change dropPartition returns

stczwd · stczwd · commit 4686a2416bf5 · 2020-08-12T20:13:00.000+08:00
Change-Id: Id45c2cfd3acbbd6fbcb88d0ac1452fc8aa4c19fa
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsAtomicPartitionManagement.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsAtomicPartitionManagement.java
@@ -21,7 +21,6 @@
 
 import org.apache.spark.annotation.Experimental;
 import org.apache.spark.sql.catalyst.InternalRow;
-import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionsException;
 import org.apache.spark.sql.catalyst.analysis.PartitionAlreadyExistsException;
 import org.apache.spark.sql.catalyst.analysis.PartitionsAlreadyExistException;
 
@@ -54,12 +53,7 @@ default void createPartition(
 
   @Override
   default boolean dropPartition(InternalRow ident) {
-    try {
-      dropPartitions(new InternalRow[]{ident});
-      return true;
-    } catch (NoSuchPartitionsException e) {
-      return false;
-    }
+    return dropPartitions(new InternalRow[]{ident});
   }
 
   /**
@@ -85,8 +79,7 @@ void createPartitions(
    * the operation of dropPartitions need to be safely rolled back.
    *
    * @param idents an array of partition identifiers
-   * @throws NoSuchPartitionsException If any partition identifier to drop doesn't exist
+   * @return true if partitions were deleted, false if any partition not exists
    */
-  void dropPartitions(
-      InternalRow[] idents) throws NoSuchPartitionsException;
+  boolean dropPartitions(InternalRow[] idents);
 }
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsPartitionManagement.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsPartitionManagement.java
@@ -45,6 +45,8 @@
 public interface SupportsPartitionManagement extends Table {
 
     /**
+     * Get the partition schema of table,
+     * this must be consistent with ${@link Table#partitioning()}
      * @return the partition schema of table
      */
     StructType partitionSchema();
@@ -71,7 +73,7 @@ void createPartition(
     boolean dropPartition(InternalRow ident);
 
     /**
-     * Test whether a partition exists using an {@link Identifier identifier} from the table.
+     * Test whether a partition exists using an {@link InternalRow ident} from the table.
      *
      * @param ident a partition identifier
      * @return true if the partition exists, false otherwise
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/InMemoryAtomicPartitionTable.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/InMemoryAtomicPartitionTable.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.connector
 import java.util
 
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionsException, PartitionAlreadyExistsException, PartitionsAlreadyExistException}
+import org.apache.spark.sql.catalyst.analysis.{PartitionAlreadyExistsException, PartitionsAlreadyExistException}
 import org.apache.spark.sql.connector.catalog.SupportsAtomicPartitionManagement
 import org.apache.spark.sql.connector.expressions.Transform
 import org.apache.spark.sql.types.StructType
@@ -67,11 +67,10 @@ class InMemoryAtomicPartitionTable (
     }
   }
 
-  override def dropPartitions(idents: Array[InternalRow]): Unit = {
+  override def dropPartitions(idents: Array[InternalRow]): Boolean = {
     if (!idents.forall(partitionExists)) {
-      throw new NoSuchPartitionsException(
-        name, idents.filterNot(partitionExists), partitionSchema)
+      return false;
     }
-    idents.foreach(dropPartition)
+    idents.forall(dropPartition)
   }
 }
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/SupportsAtomicPartitionManagementSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/SupportsAtomicPartitionManagementSuite.scala
@@ -21,7 +21,7 @@ import java.util
 
 import org.apache.spark.SparkFunSuite
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionsException, PartitionsAlreadyExistException}
+import org.apache.spark.sql.catalyst.analysis.PartitionsAlreadyExistException
 import org.apache.spark.sql.connector.{InMemoryAtomicPartitionTable, InMemoryTableCatalog}
 import org.apache.spark.sql.connector.expressions.{LogicalExpressions, NamedReference}
 import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
@@ -117,8 +117,7 @@ class SupportsAtomicPartitionManagementSuite extends SparkFunSuite {
     assert(partTable.listPartitionIdentifiers(InternalRow.empty).length == 1)
 
     val partIdents = Array(InternalRow.apply("3"), InternalRow.apply("4"))
-    assertThrows[NoSuchPartitionsException](
-      partTable.dropPartitions(partIdents))
+    assert(!partTable.dropPartitions(partIdents))
     assert(partTable.partitionExists(partIdent))
 
     partTable.dropPartition(partIdent)

Original file line number	Diff line number	Diff line change
`@@ -20,7 +20,7 @@ package org.apache.spark.sql.connector`
`20`	`20`	`import java.util`
`21`	`21`
`22`	`22`	`import org.apache.spark.sql.catalyst.InternalRow`
`23`		`-import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionsException, PartitionAlreadyExistsException, PartitionsAlreadyExistException}`
	`23`	`+import org.apache.spark.sql.catalyst.analysis.{PartitionAlreadyExistsException, PartitionsAlreadyExistException}`
`24`	`24`	`import org.apache.spark.sql.connector.catalog.SupportsAtomicPartitionManagement`
`25`	`25`	`import org.apache.spark.sql.connector.expressions.Transform`
`26`	`26`	`import org.apache.spark.sql.types.StructType`
`@@ -67,11 +67,10 @@ class InMemoryAtomicPartitionTable (`
`67`	`67`	`}`
`68`	`68`	`}`
`69`	`69`
`70`		`- override def dropPartitions(idents: Array[InternalRow]): Unit = {`
	`70`	`+ override def dropPartitions(idents: Array[InternalRow]): Boolean = {`
`71`	`71`	`if (!idents.forall(partitionExists)) {`
`72`		`- throw new NoSuchPartitionsException(`
`73`		`- name, idents.filterNot(partitionExists), partitionSchema)`
	`72`	`+ return false;`
`74`	`73`	`}`
`75`		`- idents.foreach(dropPartition)`
	`74`	`+ idents.forall(dropPartition)`
`76`	`75`	`}`
`77`	`76`	`}`