[SPARK-33364][SQL][FOLLOWUP] Refine the catalog v2 API to purge a table

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/DelegatingCatalogExtension.java

@@ -99,6 +99,11 @@ public boolean dropTable(Identifier ident) {
     return asTableCatalog().dropTable(ident);
   }
 
+  @Override
+  public boolean purgeTable(Identifier ident) {
+    return asTableCatalog().purgeTable(ident);
+  }
+
   @Override
   public void renameTable(
       Identifier oldIdent,

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java

@@ -173,26 +173,23 @@ Table alterTable(
   boolean dropTable(Identifier ident);
 
   /**
-   * Drop a table in the catalog with an option to purge.
+   * Drop a table in the catalog and completely remove its data by skipping a trash even if it is
+   * supported.
    * <p>
    * If the catalog supports views and contains a view for the identifier and not a table, this
    * must not drop the view and must return false.
    * <p>
-   * If the catalog supports the option to purge a table, this method must be overridden.
-   * The default implementation falls back to {@link #dropTable(Identifier)} dropTable} if the
-   * purge option is set to false. Otherwise, it throws {@link UnsupportedOperationException}.
+   * If the catalog supports to purge a table, this method should be overridden.
+   * The default implementation throws {@link UnsupportedOperationException}.
    *
    * @param ident a table identifier
-   * @param purge whether a table should be purged
    * @return true if a table was deleted, false if no table exists for the identifier
+   * @throws UnsupportedOperationException If table purging is not supported
    *
    * @since 3.1.0
    */
-  default boolean dropTable(Identifier ident, boolean purge) {
-    if (purge) {
-      throw new UnsupportedOperationException("Purge option is not supported.");
-    }
-    return dropTable(ident);
+  default boolean purgeTable(Identifier ident) throws UnsupportedOperationException {
+    throw new UnsupportedOperationException("Purge table is not supported.");
   }
 
   /**

sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/TableCatalogSuite.scala

@@ -643,6 +643,11 @@ class TableCatalogSuite extends SparkFunSuite {
     assert(!catalog.tableExists(testIdent))
   }
 
+  test("purgeTable") {
+    val catalog = newCatalog()
+    intercept[UnsupportedOperationException](catalog.purgeTable(testIdent))
+  }
+
   test("renameTable") {
     val catalog = newCatalog()
 

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DropTableExec.scala

@@ -35,7 +35,7 @@ case class DropTableExec(
   override def run(): Seq[InternalRow] = {
     if (catalog.tableExists(ident)) {
       invalidateCache()
-      catalog.dropTable(ident, purge)
+      if (purge) catalog.purgeTable(ident) else catalog.dropTable(ident)
     } else if (!ifExists) {
       throw new NoSuchTableException(ident)
     }

sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/DropTableSuite.scala

@@ -29,8 +29,8 @@ class DropTableSuite extends command.DropTableSuiteBase with CommandSuiteBase {
       val errMsg = intercept[UnsupportedOperationException] {
         sql(s"DROP TABLE $catalog.ns.tbl PURGE")
       }.getMessage
-      // The default TableCatalog.dropTable implementation doesn't support the purge option.
-      assert(errMsg.contains("Purge option is not supported"))
+      // The default TableCatalog.purgeTable implementation throws an exception.
+      assert(errMsg.contains("Purge table is not supported"))
     }
   }