You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2020/09/27 11:34:45 UTC
[GitHub] [spark] stczwd commented on a change in pull request #29339: [SPARK-32512][SQL] add alter table add/drop partition command for datasourcev2
stczwd commented on a change in pull request #29339:
URL: https://github.com/apache/spark/pull/29339#discussion_r495562634
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AlterTableAddPartitionExec.scala
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.v2
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.PartitionsAlreadyExistException
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.{SupportsAtomicPartitionManagement, SupportsPartitionManagement}
+
+/**
+ * Physical plan node for adding partitions of table.
+ */
+case class AlterTableAddPartitionExec(
+ table: SupportsPartitionManagement,
+ partitions: Seq[(InternalRow, Map[String, String])],
+ ignoreIfExists: Boolean) extends V2CommandExec {
+ import DataSourceV2Implicits._
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def run(): Seq[InternalRow] = {
+ val existsPartIdents = partitions.map(_._1).filter(table.partitionExists)
Review comment:
good
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
##########
@@ -276,6 +276,13 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
case r @ ShowTableProperties(rt: ResolvedTable, propertyKey) =>
ShowTablePropertiesExec(r.output, rt.table, propertyKey) :: Nil
+ case AlterTableAddPartition(table, parts, ignoreIfExists) =>
+ AlterTableAddPartitionExec(table, parts, ignoreIfExists) :: Nil
+
+ case AlterTableDropPartition(table, partIdents, ignoreIfNotExists) =>
+ AlterTableDropPartitionExec(
+ table, partIdents, ignoreIfNotExists) :: Nil
Review comment:
ok
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Implicits.scala
##########
@@ -62,4 +85,42 @@ object DataSourceV2Implicits {
new CaseInsensitiveStringMap(options.asJava)
}
}
+
+ implicit class TablePartitionSpecHelper(partSpec: TablePartitionSpec) {
+ def asPartitionIdentifier(partSchema: StructType): InternalRow = {
+ val conflictKeys = partSpec.keys.toSeq.diff(partSchema.map(_.name))
+ if (conflictKeys.nonEmpty) {
+ throw new AnalysisException(
+ s"Partition key ${conflictKeys.mkString(",")} not exists")
Review comment:
ok
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AlterTableDropPartitionExec.scala
##########
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.v2
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionsException
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.{SupportsAtomicPartitionManagement, SupportsPartitionManagement}
+
+/**
+ * Physical plan node for dropping partitions of table.
+ */
+case class AlterTableDropPartitionExec(
+ table: SupportsPartitionManagement,
+ partIdents: Seq[InternalRow],
+ ignoreIfNotExists: Boolean) extends V2CommandExec {
+ import DataSourceV2Implicits._
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def run(): Seq[InternalRow] = {
+ val notExistsPartIdents = partIdents.filterNot(table.partitionExists)
Review comment:
good
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
##########
@@ -2066,13 +2069,71 @@ class DataSourceV2SQLSuite
}
test("ALTER TABLE ADD PARTITION") {
- val t = "testcat.ns1.ns2.tbl"
+ val t = "testpart.ns1.ns2.tbl"
withTable(t) {
spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
- val e = intercept[AnalysisException] {
- sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
- }
- assert(e.message.contains("ALTER TABLE ADD PARTITION is only supported with v1 tables"))
+ spark.sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
+
+ val partTable = catalog("testpart").asTableCatalog
+ .loadTable(Identifier.of(Array("ns1", "ns2"), "tbl")).asInstanceOf[InMemoryPartitionTable]
+ assert(partTable.partitionExists(InternalRow.fromSeq(Seq(1))))
+
+ val partMetadata =
+ partTable.loadPartitionMetadata(InternalRow.fromSeq(Seq(1)))
+ assert(partMetadata.containsKey("location"))
+ assert(partMetadata.get("location") == "loc")
+
+ partTable.clearPartitions()
+ }
+ }
+
+ test("ALTER TABLE ADD PARTITIONS") {
+ val t = "testpart.ns1.ns2.tbl"
+ withTable(t) {
+ spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
+ spark.sql(
+ s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc' PARTITION (id=2) LOCATION 'loc1'")
+
+ val partTable = catalog("testpart").asTableCatalog
+ .loadTable(Identifier.of(Array("ns1", "ns2"), "tbl")).asInstanceOf[InMemoryPartitionTable]
+ assert(partTable.partitionExists(InternalRow.fromSeq(Seq(1))))
+ assert(partTable.partitionExists(InternalRow.fromSeq(Seq(2))))
+
+ val partMetadata =
+ partTable.loadPartitionMetadata(InternalRow.fromSeq(Seq(1)))
+ assert(partMetadata.containsKey("location"))
+ assert(partMetadata.get("location") == "loc")
+
+ val partMetadata1 =
+ partTable.loadPartitionMetadata(InternalRow.fromSeq(Seq(2)))
+ assert(partMetadata1.containsKey("location"))
+ assert(partMetadata1.get("location") == "loc1")
+
+ partTable.clearPartitions()
+ }
+ }
+
+ test("ALTER TABLE ADD PARTITIONS: partition already exists") {
+ val t = "testpart.ns1.ns2.tbl"
+ withTable(t) {
+ spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
+ spark.sql(
+ s"ALTER TABLE $t ADD PARTITION (id=2) LOCATION 'loc1'")
+
+ assertThrows[PartitionsAlreadyExistException](
+ spark.sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'" +
+ s" PARTITION (id=2) LOCATION 'loc1'"))
Review comment:
ok
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
##########
@@ -2066,13 +2069,71 @@ class DataSourceV2SQLSuite
}
test("ALTER TABLE ADD PARTITION") {
- val t = "testcat.ns1.ns2.tbl"
+ val t = "testpart.ns1.ns2.tbl"
withTable(t) {
spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
- val e = intercept[AnalysisException] {
- sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
- }
- assert(e.message.contains("ALTER TABLE ADD PARTITION is only supported with v1 tables"))
+ spark.sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
+
+ val partTable = catalog("testpart").asTableCatalog
+ .loadTable(Identifier.of(Array("ns1", "ns2"), "tbl")).asInstanceOf[InMemoryPartitionTable]
+ assert(partTable.partitionExists(InternalRow.fromSeq(Seq(1))))
+
+ val partMetadata =
+ partTable.loadPartitionMetadata(InternalRow.fromSeq(Seq(1)))
+ assert(partMetadata.containsKey("location"))
+ assert(partMetadata.get("location") == "loc")
+
+ partTable.clearPartitions()
+ }
+ }
+
+ test("ALTER TABLE ADD PARTITIONS") {
+ val t = "testpart.ns1.ns2.tbl"
+ withTable(t) {
+ spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
+ spark.sql(
+ s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc' PARTITION (id=2) LOCATION 'loc1'")
+
+ val partTable = catalog("testpart").asTableCatalog
+ .loadTable(Identifier.of(Array("ns1", "ns2"), "tbl")).asInstanceOf[InMemoryPartitionTable]
+ assert(partTable.partitionExists(InternalRow.fromSeq(Seq(1))))
+ assert(partTable.partitionExists(InternalRow.fromSeq(Seq(2))))
+
+ val partMetadata =
+ partTable.loadPartitionMetadata(InternalRow.fromSeq(Seq(1)))
+ assert(partMetadata.containsKey("location"))
+ assert(partMetadata.get("location") == "loc")
+
+ val partMetadata1 =
+ partTable.loadPartitionMetadata(InternalRow.fromSeq(Seq(2)))
+ assert(partMetadata1.containsKey("location"))
+ assert(partMetadata1.get("location") == "loc1")
+
+ partTable.clearPartitions()
+ }
+ }
+
+ test("ALTER TABLE ADD PARTITIONS: partition already exists") {
+ val t = "testpart.ns1.ns2.tbl"
+ withTable(t) {
+ spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
+ spark.sql(
+ s"ALTER TABLE $t ADD PARTITION (id=2) LOCATION 'loc1'")
+
+ assertThrows[PartitionsAlreadyExistException](
+ spark.sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'" +
+ s" PARTITION (id=2) LOCATION 'loc1'"))
+
+ val partTable = catalog("testpart").asTableCatalog
+ .loadTable(Identifier.of(Array("ns1", "ns2"), "tbl")).asInstanceOf[InMemoryPartitionTable]
+ assert(!partTable.partitionExists(InternalRow.fromSeq(Seq(1))))
+
+ spark.sql(s"ALTER TABLE $t ADD IF NOT EXISTS PARTITION (id=1) LOCATION 'loc'" +
+ s" PARTITION (id=2) LOCATION 'loc1'")
Review comment:
ok
##########
File path: sql/catalyst/src/test/scala/org/apache/spark/sql/connector/InMemoryPartitionTableCatalog.scala
##########
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector
+import java.util
Review comment:
ok
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AlterTableDropPartitionExec.scala
##########
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.v2
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionsException
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.{SupportsAtomicPartitionManagement, SupportsPartitionManagement}
+
+/**
+ * Physical plan node for dropping partitions of table.
+ */
+case class AlterTableDropPartitionExec(
+ table: SupportsPartitionManagement,
+ partIdents: Seq[InternalRow],
+ ignoreIfNotExists: Boolean) extends V2CommandExec {
+ import DataSourceV2Implicits._
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def run(): Seq[InternalRow] = {
+ val notExistsPartIdents = partIdents.filterNot(table.partitionExists)
+ if (notExistsPartIdents.nonEmpty && !ignoreIfNotExists) {
+ throw new NoSuchPartitionsException(
+ table.name(), notExistsPartIdents, table.partitionSchema())
+ }
+
+ val existsPartitions = partIdents.filterNot(notExistsPartIdents.contains)
+ existsPartitions match {
+ case Seq() => // Nothing will be done
+ case Seq(partIdent) =>
+ table.dropPartition(partIdent)
+ case Seq(_ *) if table.isInstanceOf[SupportsAtomicPartitionManagement] =>
+ table.asAtomicPartitionable
+ .dropPartitions(existsPartitions.toArray)
Review comment:
ok
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AlterTableAddPartitionExec.scala
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.v2
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.PartitionsAlreadyExistException
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.{SupportsAtomicPartitionManagement, SupportsPartitionManagement}
+
+/**
+ * Physical plan node for adding partitions of table.
+ */
+case class AlterTableAddPartitionExec(
+ table: SupportsPartitionManagement,
+ partitions: Seq[(InternalRow, Map[String, String])],
+ ignoreIfExists: Boolean) extends V2CommandExec {
+ import DataSourceV2Implicits._
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def run(): Seq[InternalRow] = {
+ val existsPartIdents = partitions.map(_._1).filter(table.partitionExists)
+ if (existsPartIdents.nonEmpty && !ignoreIfExists) {
+ throw new PartitionsAlreadyExistException(
+ table.name(), existsPartIdents, table.partitionSchema())
+ }
+
+ val notExistsPartitions =
+ partitions.filterNot(part => existsPartIdents.contains(part._1))
+ notExistsPartitions match {
+ case Seq() => // Nothing will be done
+ case Seq((partIdent, properties)) =>
+ table.createPartition(partIdent, properties.asJava)
+ case Seq(_ *) if table.isInstanceOf[SupportsAtomicPartitionManagement] =>
+ table.asAtomicPartitionable
+ .createPartitions(
+ notExistsPartitions.map(_._1).toArray,
Review comment:
good
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
##########
@@ -2066,13 +2069,71 @@ class DataSourceV2SQLSuite
}
test("ALTER TABLE ADD PARTITION") {
- val t = "testcat.ns1.ns2.tbl"
+ val t = "testpart.ns1.ns2.tbl"
withTable(t) {
spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
- val e = intercept[AnalysisException] {
- sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
- }
- assert(e.message.contains("ALTER TABLE ADD PARTITION is only supported with v1 tables"))
+ spark.sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
+
+ val partTable = catalog("testpart").asTableCatalog
+ .loadTable(Identifier.of(Array("ns1", "ns2"), "tbl")).asInstanceOf[InMemoryPartitionTable]
+ assert(partTable.partitionExists(InternalRow.fromSeq(Seq(1))))
+
+ val partMetadata =
+ partTable.loadPartitionMetadata(InternalRow.fromSeq(Seq(1)))
Review comment:
ok
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
##########
@@ -2066,13 +2069,71 @@ class DataSourceV2SQLSuite
}
test("ALTER TABLE ADD PARTITION") {
- val t = "testcat.ns1.ns2.tbl"
+ val t = "testpart.ns1.ns2.tbl"
withTable(t) {
spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
- val e = intercept[AnalysisException] {
- sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
- }
- assert(e.message.contains("ALTER TABLE ADD PARTITION is only supported with v1 tables"))
+ spark.sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
+
+ val partTable = catalog("testpart").asTableCatalog
+ .loadTable(Identifier.of(Array("ns1", "ns2"), "tbl")).asInstanceOf[InMemoryPartitionTable]
+ assert(partTable.partitionExists(InternalRow.fromSeq(Seq(1))))
+
+ val partMetadata =
+ partTable.loadPartitionMetadata(InternalRow.fromSeq(Seq(1)))
+ assert(partMetadata.containsKey("location"))
+ assert(partMetadata.get("location") == "loc")
+
+ partTable.clearPartitions()
+ }
+ }
+
+ test("ALTER TABLE ADD PARTITIONS") {
+ val t = "testpart.ns1.ns2.tbl"
+ withTable(t) {
+ spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
+ spark.sql(
+ s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc' PARTITION (id=2) LOCATION 'loc1'")
+
+ val partTable = catalog("testpart").asTableCatalog
+ .loadTable(Identifier.of(Array("ns1", "ns2"), "tbl")).asInstanceOf[InMemoryPartitionTable]
+ assert(partTable.partitionExists(InternalRow.fromSeq(Seq(1))))
+ assert(partTable.partitionExists(InternalRow.fromSeq(Seq(2))))
+
+ val partMetadata =
+ partTable.loadPartitionMetadata(InternalRow.fromSeq(Seq(1)))
Review comment:
ok
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
##########
@@ -2087,14 +2148,52 @@ class DataSourceV2SQLSuite
}
}
+ test("ALTER TABLE DROP PARTITION") {
+ val t = "testpart.ns1.ns2.tbl"
+ withTable(t) {
+ spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
+ spark.sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
+ spark.sql(s"ALTER TABLE $t DROP PARTITION (id=1)")
+
+ val partTable =
+ catalog("testpart").asTableCatalog.loadTable(Identifier.of(Array("ns1", "ns2"), "tbl"))
+ assert(!partTable.asPartitionable.partitionExists(InternalRow.fromSeq(Seq(1))))
+ }
+ }
+
test("ALTER TABLE DROP PARTITIONS") {
- val t = "testcat.ns1.ns2.tbl"
+ val t = "testpart.ns1.ns2.tbl"
withTable(t) {
spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
- val e = intercept[AnalysisException] {
- sql(s"ALTER TABLE $t DROP PARTITION (id=1)")
- }
- assert(e.message.contains("ALTER TABLE DROP PARTITION is only supported with v1 tables"))
+ spark.sql(s"ALTER TABLE $t ADD IF NOT EXISTS PARTITION (id=1) LOCATION 'loc'" +
+ s" PARTITION (id=2) LOCATION 'loc1'")
Review comment:
ok
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Implicits.scala
##########
@@ -62,4 +85,42 @@ object DataSourceV2Implicits {
new CaseInsensitiveStringMap(options.asJava)
}
}
+
+ implicit class TablePartitionSpecHelper(partSpec: TablePartitionSpec) {
+ def asPartitionIdentifier(partSchema: StructType): InternalRow = {
+ val conflictKeys = partSpec.keys.toSeq.diff(partSchema.map(_.name))
+ if (conflictKeys.nonEmpty) {
+ throw new AnalysisException(
+ s"Partition key ${conflictKeys.mkString(",")} not exists")
+ }
+
+ val partValues = partSchema.map { part =>
+ val partValue = partSpec.get(part.name).orNull
+ if (partValue == null) {
+ null
+ } else {
+ part.dataType match {
Review comment:
a `TODO` sounds good to me.
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
##########
@@ -2066,13 +2069,71 @@ class DataSourceV2SQLSuite
}
test("ALTER TABLE ADD PARTITION") {
- val t = "testcat.ns1.ns2.tbl"
+ val t = "testpart.ns1.ns2.tbl"
withTable(t) {
spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
- val e = intercept[AnalysisException] {
- sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
- }
- assert(e.message.contains("ALTER TABLE ADD PARTITION is only supported with v1 tables"))
+ spark.sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'")
+
+ val partTable = catalog("testpart").asTableCatalog
+ .loadTable(Identifier.of(Array("ns1", "ns2"), "tbl")).asInstanceOf[InMemoryPartitionTable]
+ assert(partTable.partitionExists(InternalRow.fromSeq(Seq(1))))
+
+ val partMetadata =
+ partTable.loadPartitionMetadata(InternalRow.fromSeq(Seq(1)))
+ assert(partMetadata.containsKey("location"))
+ assert(partMetadata.get("location") == "loc")
+
+ partTable.clearPartitions()
+ }
+ }
+
+ test("ALTER TABLE ADD PARTITIONS") {
+ val t = "testpart.ns1.ns2.tbl"
+ withTable(t) {
+ spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
+ spark.sql(
+ s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc' PARTITION (id=2) LOCATION 'loc1'")
+
+ val partTable = catalog("testpart").asTableCatalog
+ .loadTable(Identifier.of(Array("ns1", "ns2"), "tbl")).asInstanceOf[InMemoryPartitionTable]
+ assert(partTable.partitionExists(InternalRow.fromSeq(Seq(1))))
+ assert(partTable.partitionExists(InternalRow.fromSeq(Seq(2))))
+
+ val partMetadata =
+ partTable.loadPartitionMetadata(InternalRow.fromSeq(Seq(1)))
+ assert(partMetadata.containsKey("location"))
+ assert(partMetadata.get("location") == "loc")
+
+ val partMetadata1 =
+ partTable.loadPartitionMetadata(InternalRow.fromSeq(Seq(2)))
Review comment:
ok
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org