[#4814] Improvement (catalog-lakehouse-paimon): use purgeTable for Paimon instead of dropTable inGravitino

catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalogOperations.java

@@ -438,15 +438,8 @@ public GravitinoPaimonTable alterTable(NameIdentifier identifier, TableChange...
    */
   @Override
   public boolean dropTable(NameIdentifier identifier) {
-    try {
-      NameIdentifier tableIdentifier = buildPaimonNameIdentifier(identifier);
-      paimonCatalogOps.dropTable(tableIdentifier.toString());
-    } catch (Catalog.TableNotExistException e) {
-      LOG.warn("Paimon table {} does not exist.", identifier);
-      return false;
-    }
-    LOG.info("Dropped Paimon table {}.", identifier);
-    return true;
+    throw new UnsupportedOperationException(
+        "Paimon dropTable will both remove the metadata and data, please use purgeTable instead in Gravitino.");
   }
 
   /**
@@ -458,7 +451,15 @@ public boolean dropTable(NameIdentifier identifier) {
    */
   @Override
   public boolean purgeTable(NameIdentifier identifier) throws UnsupportedOperationException {
-    throw new UnsupportedOperationException("purgeTable is unsupported now for Paimon Catalog.");
+    try {
+      NameIdentifier tableIdentifier = buildPaimonNameIdentifier(identifier);
+      paimonCatalogOps.purgeTable(tableIdentifier.toString());
+    } catch (Catalog.TableNotExistException e) {
+      LOG.warn("Paimon table {} does not exist.", identifier);
+      return false;
+    }
+    LOG.info("Purged Paimon table {}.", identifier);
+    return true;
   }
 
   @Override

catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/ops/PaimonCatalogOps.java

@@ -90,7 +90,7 @@ public void createTable(String tableName, Schema schema)
     catalog.createTable(tableIdentifier(tableName), schema, false);
   }
 
-  public void dropTable(String tableName) throws TableNotExistException {
+  public void purgeTable(String tableName) throws TableNotExistException {
     catalog.dropTable(tableIdentifier(tableName), false);
   }
 

catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/TestGravitinoPaimonTable.java

@@ -101,7 +101,7 @@ void resetSchema() {
                 return NameIdentifier.of(
                     Namespace.of(levels[levels.length - 1]), nameIdentifier.name());
               })
-          .forEach(nameIdentifier -> paimonCatalogOperations.dropTable(nameIdentifier));
+          .forEach(nameIdentifier -> paimonCatalogOperations.purgeTable(nameIdentifier));
     }
     paimonCatalogOperations.dropSchema(schemaIdent, false);
     initPaimonSchema();
@@ -358,7 +358,7 @@ void testDropPaimonTable() {
         new SortOrder[0]);
 
     Assertions.assertTrue(paimonCatalogOperations.tableExists(tableIdentifier));
-    paimonCatalogOperations.dropTable(tableIdentifier);
+    paimonCatalogOperations.purgeTable(tableIdentifier);
     Assertions.assertFalse(paimonCatalogOperations.tableExists(tableIdentifier));
   }
 

catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/integration/test/CatalogPaimonBaseIT.java

@@ -595,13 +595,13 @@ void testListAndDropPaimonTable() throws DatabaseNotExistException {
     Assertions.assertEquals("table_1", tableIdentifiers.get(0));
     Assertions.assertEquals("table_2", tableIdentifiers.get(1));
 
-    Assertions.assertDoesNotThrow(() -> tableCatalog.dropTable(table1));
+    Assertions.assertDoesNotThrow(() -> tableCatalog.purgeTable(table1));
 
     nameIdentifiers = tableCatalog.listTables(Namespace.of(schemaName));
     Assertions.assertEquals(1, nameIdentifiers.length);
     Assertions.assertEquals("table_2", nameIdentifiers[0].name());
 
-    Assertions.assertDoesNotThrow(() -> tableCatalog.dropTable(table2));
+    Assertions.assertDoesNotThrow(() -> tableCatalog.purgeTable(table2));
     Namespace schemaNamespace = Namespace.of(schemaName);
     nameIdentifiers = tableCatalog.listTables(schemaNamespace);
     Assertions.assertEquals(0, nameIdentifiers.length);

catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/integration/test/CatalogPaimonKerberosFilesystemIT.java

@@ -266,8 +266,8 @@ void testPaimonWithKerberos() {
     Table loadTable = catalog.asTableCatalog().loadTable(tableNameIdentifier);
     Assertions.assertEquals(loadTable.name(), tableNameIdentifier.name());
 
-    // Drop table
-    catalog.asTableCatalog().dropTable(tableNameIdentifier);
+    // Purge table
+    catalog.asTableCatalog().purgeTable(tableNameIdentifier);
     Assertions.assertFalse(catalog.asTableCatalog().tableExists(tableNameIdentifier));
 
     // Drop schema

catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/ops/TestPaimonCatalogOps.java

@@ -153,10 +153,10 @@ void testLoadListAndDropTableOperations() throws Exception {
     assertEquals(OPTIONS.get(BUCKET.key()), table.options().get(BUCKET.key()));
 
     // drop table
-    Assertions.assertDoesNotThrow(() -> paimonCatalogOps.dropTable(IDENTIFIER.toString()));
+    Assertions.assertDoesNotThrow(() -> paimonCatalogOps.purgeTable(IDENTIFIER.toString()));
     Assertions.assertThrowsExactly(
         Catalog.TableNotExistException.class,
-        () -> paimonCatalogOps.dropTable(IDENTIFIER.toString()));
+        () -> paimonCatalogOps.purgeTable(IDENTIFIER.toString()));
 
     // list table again
     Assertions.assertEquals(

docs/lakehouse-paimon-catalog.md

@@ -59,9 +59,9 @@ Please refer to [Manage Relational Metadata Using Gravitino](./manage-relational
 
 ### Schema properties
 
-- Doesn't support specify location and store any schema properties when createSchema for FilesystemCatalog now.
-- Doesn't return any schema properties when loadSchema for FilesystemCatalog now.
-- Doesn't support store schema comment for FilesystemCatalog now.
+- Doesn't support specify location and store any schema properties when createSchema for FilesystemCatalog.
+- Doesn't return any schema properties when loadSchema for FilesystemCatalog.
+- Doesn't support store schema comment for FilesystemCatalog.
 
 ### Schema operations
 
@@ -71,20 +71,23 @@ Please refer to [Manage Relational Metadata Using Gravitino](./manage-relational
 
 ### Table capabilities
 
-- Supporting createTable, dropTable, alterTable, loadTable and listTable.
-```
-dropTable will delete the table location directly, similar with purgeTable.
-```
+- Supporting createTable, purgeTable, alterTable, loadTable and listTable.
 - Supporting Column default value through table properties, such as `fields.{columnName}.default-value`, not column expression.
-
+
+- Doesn't support dropTable.
 - Doesn't support table distribution and sort orders.
 
+:::info
+Gravitino Paimon Catalog does not support dropTable, because the dropTable in Paimon will both remove the table metadata and the table location from the file system and skip the trash, we should use purgeTable instead in Gravitino.
+:::
+
 :::info
 Paimon does not support auto increment column.
 :::
 
 #### Table changes
 
+- RenameTable
 - AddColumn
 - DeleteColumn
 - RenameColumn