Support force option on RegisterTable procedure

api/src/main/java/org/apache/iceberg/catalog/Catalog.java

@@ -336,14 +336,17 @@ default boolean dropTable(TableIdentifier identifier) {
   default void invalidateTable(TableIdentifier identifier) {}
 
   /**
-   * Register a table with the catalog if it does not exist.
+   * Register a table with the catalog.
    *
    * @param identifier a table identifier
    * @param metadataFileLocation the location of a metadata file
+   * @param force If true will register the table even if the table already existed, otherwise
+   *     it will throw table already existing error. Default is false.
    * @return a Table instance
    * @throws AlreadyExistsException if the table already exists in the catalog.
    */
-  default Table registerTable(TableIdentifier identifier, String metadataFileLocation) {
+  default Table registerTable(
+      TableIdentifier identifier, String metadataFileLocation, boolean force) {
     throw new UnsupportedOperationException("Registering tables is not supported");
   }
 

aws/src/integration/java/org/apache/iceberg/aws/dynamodb/TestDynamoDbCatalog.java

@@ -308,7 +308,7 @@ public void testRegisterTable() {
     Assertions.assertThat(catalog.dropTable(identifier, false)).isTrue();
     TableOperations ops = ((HasTableOperations) registeringTable).operations();
     String metadataLocation = ((DynamoDbTableOperations) ops).currentMetadataLocation();
-    Assertions.assertThat(catalog.registerTable(identifier, metadataLocation)).isNotNull();
+    Assertions.assertThat(catalog.registerTable(identifier, metadataLocation, false)).isNotNull();
     Assertions.assertThat(catalog.loadTable(identifier)).isNotNull();
     Assertions.assertThat(catalog.dropTable(identifier, true)).isTrue();
     Assertions.assertThat(catalog.dropNamespace(namespace)).isTrue();
@@ -323,7 +323,9 @@ public void testRegisterExistingTable() {
     Table registeringTable = catalog.loadTable(identifier);
     TableOperations ops = ((HasTableOperations) registeringTable).operations();
     String metadataLocation = ((DynamoDbTableOperations) ops).currentMetadataLocation();
-    Assertions.assertThatThrownBy(() -> catalog.registerTable(identifier, metadataLocation))
+    catalog.registerTable(identifier, metadataLocation, true);
+    Assertions.assertThat(catalog.loadTable(identifier)).isNotNull();
+    Assertions.assertThatThrownBy(() -> catalog.registerTable(identifier, metadataLocation, false))
         .isInstanceOf(AlreadyExistsException.class);
     Assertions.assertThat(catalog.dropTable(identifier, true)).isTrue();
     Assertions.assertThat(catalog.dropNamespace(namespace)).isTrue();

aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java

@@ -445,7 +445,7 @@ public void testRegisterTable() {
     Table table = glueCatalog.loadTable(identifier);
     String metadataLocation = ((BaseTable) table).operations().current().metadataFileLocation();
     Assertions.assertThat(glueCatalog.dropTable(identifier, false)).isTrue();
-    Assertions.assertThat(glueCatalog.registerTable(identifier, metadataLocation)).isNotNull();
+    Assertions.assertThat(glueCatalog.registerTable(identifier, metadataLocation, false)).isNotNull();
     Assertions.assertThat(glueCatalog.loadTable(identifier)).isNotNull();
     Assertions.assertThat(glueCatalog.dropTable(identifier, true)).isTrue();
     Assertions.assertThat(glueCatalog.dropNamespace(Namespace.of(namespace))).isTrue();
@@ -459,7 +459,9 @@ public void testRegisterTableAlreadyExists() {
     TableIdentifier identifier = TableIdentifier.of(namespace, tableName);
     Table table = glueCatalog.loadTable(identifier);
     String metadataLocation = ((BaseTable) table).operations().current().metadataFileLocation();
-    Assertions.assertThatThrownBy(() -> glueCatalog.registerTable(identifier, metadataLocation))
+    glueCatalog.registerTable(identifier, metadataLocation, true);
+    Assertions.assertThat(glueCatalog.loadTable(identifier)).isNotNull();
+    Assertions.assertThatThrownBy(() -> glueCatalog.registerTable(identifier, metadataLocation, false))
         .isInstanceOf(AlreadyExistsException.class);
     Assertions.assertThat(glueCatalog.dropTable(identifier, true)).isTrue();
     Assertions.assertThat(glueCatalog.dropNamespace(Namespace.of(namespace))).isTrue();

core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java

@@ -66,19 +66,28 @@ public Table loadTable(TableIdentifier identifier) {
   }
 
   @Override
-  public Table registerTable(TableIdentifier identifier, String metadataFileLocation) {
+  public Table registerTable(
+      TableIdentifier identifier, String metadataFileLocation, boolean force) {
     Preconditions.checkArgument(
         identifier != null && isValidIdentifier(identifier), "Invalid identifier: %s", identifier);
     Preconditions.checkArgument(
         metadataFileLocation != null && !metadataFileLocation.isEmpty(),
         "Cannot register an empty metadata file location as a table");
 
-    // Throw an exception if this table already exists in the catalog.
+    TableOperations ops = newTableOps(identifier);
     if (tableExists(identifier)) {
-      throw new AlreadyExistsException("Table already exists: %s", identifier);
+      if (!force) {
+        // Throw an exception if this table already exists in the catalog.
+        throw new AlreadyExistsException("Table already exists: %s", identifier);
+      }
+      TableMetadata current = ops.current();
+      if (metadataFileLocation.equals(current.metadataFileLocation())) {
+        LOG.warn("The metadata file location is the same as the existing table, ignore this operation.");
+        return new BaseTable(ops, identifier.toString());
+      }
+      dropTable(identifier, false);
     }
 
-    TableOperations ops = newTableOps(identifier);
     InputFile metadataFile = ops.io().newInputFile(metadataFileLocation);
     TableMetadata metadata = TableMetadataParser.read(ops.io(), metadataFile);
     ops.commit(null, metadata);

core/src/main/java/org/apache/iceberg/CachingCatalog.java

@@ -188,8 +188,9 @@ public void invalidateTable(TableIdentifier ident) {
   }
 
   @Override
-  public Table registerTable(TableIdentifier identifier, String metadataFileLocation) {
-    Table table = catalog.registerTable(identifier, metadataFileLocation);
+  public Table registerTable(
+      TableIdentifier identifier, String metadataFileLocation, boolean force) {
+    Table table = catalog.registerTable(identifier, metadataFileLocation, force);
     invalidateTable(identifier);
     return table;
   }

core/src/main/java/org/apache/iceberg/catalog/BaseSessionCatalog.java

@@ -85,7 +85,7 @@ public TableBuilder buildTable(TableIdentifier ident, Schema schema) {
     }
 
     @Override
-    public Table registerTable(TableIdentifier ident, String metadataFileLocation) {
+    public Table registerTable(TableIdentifier ident, String metadataFileLocation, boolean force) {
       return BaseSessionCatalog.this.registerTable(context, ident, metadataFileLocation);
     }
 

core/src/main/java/org/apache/iceberg/rest/RESTCatalog.java

@@ -203,8 +203,8 @@ public void renameTable(TableIdentifier from, TableIdentifier to) {
   }
 
   @Override
-  public Table registerTable(TableIdentifier ident, String metadataFileLocation) {
-    return delegate.registerTable(ident, metadataFileLocation);
+  public Table registerTable(TableIdentifier ident, String metadataFileLocation, boolean force) {
+    return delegate.registerTable(ident, metadataFileLocation, force);
   }
 
   @Override

core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java

@@ -643,7 +643,7 @@ public void testRegisterTable() throws IOException {
     Table registeringTable = catalog.loadTable(identifier);
     TableOperations ops = ((HasTableOperations) registeringTable).operations();
     String metadataLocation = ((HadoopTableOperations) ops).current().metadataFileLocation();
-    Assertions.assertThat(catalog.registerTable(identifier2, metadataLocation)).isNotNull();
+    Assertions.assertThat(catalog.registerTable(identifier2, metadataLocation, false)).isNotNull();
     Assertions.assertThat(catalog.loadTable(identifier2)).isNotNull();
     Assertions.assertThat(catalog.dropTable(identifier)).isTrue();
     Assertions.assertThat(catalog.dropTable(identifier2)).isTrue();
@@ -657,7 +657,9 @@ public void testRegisterExistingTable() throws IOException {
     Table registeringTable = catalog.loadTable(identifier);
     TableOperations ops = ((HasTableOperations) registeringTable).operations();
     String metadataLocation = ((HadoopTableOperations) ops).current().metadataFileLocation();
-    Assertions.assertThatThrownBy(() -> catalog.registerTable(identifier, metadataLocation))
+    catalog.registerTable(identifier, metadataLocation, true);
+    Assertions.assertThat(catalog.loadTable(identifier)).isNotNull();
+    Assertions.assertThatThrownBy(() -> catalog.registerTable(identifier, metadataLocation, false))
         .isInstanceOf(AlreadyExistsException.class)
         .hasMessage("Table already exists: a.t1");
     Assertions.assertThat(catalog.dropTable(identifier)).isTrue();

core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java

@@ -694,7 +694,7 @@ public void testRegisterTable() {
     catalog.dropTable(identifier, false);
     TableOperations ops = ((HasTableOperations) registeringTable).operations();
     String metadataLocation = ((JdbcTableOperations) ops).currentMetadataLocation();
-    Assertions.assertThat(catalog.registerTable(identifier, metadataLocation)).isNotNull();
+    Assertions.assertThat(catalog.registerTable(identifier, metadataLocation, false)).isNotNull();
     Assertions.assertThat(catalog.loadTable(identifier)).isNotNull();
     Assertions.assertThat(catalog.dropTable(identifier)).isTrue();
   }
@@ -706,7 +706,9 @@ public void testRegisterExistingTable() {
     Table registeringTable = catalog.loadTable(identifier);
     TableOperations ops = ((HasTableOperations) registeringTable).operations();
     String metadataLocation = ((JdbcTableOperations) ops).currentMetadataLocation();
-    Assertions.assertThatThrownBy(() -> catalog.registerTable(identifier, metadataLocation))
+    catalog.registerTable(identifier, metadataLocation, true);
+    Assertions.assertThat(catalog.loadTable(identifier)).isNotNull();
+    Assertions.assertThatThrownBy(() -> catalog.registerTable(identifier, metadataLocation, false))
         .isInstanceOf(AlreadyExistsException.class)
         .hasMessage("Table already exists: a.t1");
     Assertions.assertThat(catalog.dropTable(identifier)).isTrue();

dell/src/test/java/org/apache/iceberg/dell/ecs/TestEcsCatalog.java

@@ -190,7 +190,8 @@ public void testRegisterTable() {
     ecsCatalog.dropTable(identifier, false);
     TableOperations ops = ((HasTableOperations) registeringTable).operations();
     String metadataLocation = ((EcsTableOperations) ops).currentMetadataLocation();
-    Assertions.assertThat(ecsCatalog.registerTable(identifier, metadataLocation)).isNotNull();
+    Assertions.assertThat(ecsCatalog.registerTable(identifier, metadataLocation, false))
+        .isNotNull();
     Assertions.assertThat(ecsCatalog.loadTable(identifier)).isNotNull();
     Assertions.assertThat(ecsCatalog.dropTable(identifier, true)).isTrue();
   }
@@ -202,7 +203,10 @@ public void testRegisterExistingTable() {
     Table registeringTable = ecsCatalog.loadTable(identifier);
     TableOperations ops = ((HasTableOperations) registeringTable).operations();
     String metadataLocation = ((EcsTableOperations) ops).currentMetadataLocation();
-    Assertions.assertThatThrownBy(() -> ecsCatalog.registerTable(identifier, metadataLocation))
+    ecsCatalog.registerTable(identifier, metadataLocation, true);
+    Assertions.assertThat(ecsCatalog.loadTable(identifier)).isNotNull();
+    Assertions.assertThatThrownBy(
+            () -> ecsCatalog.registerTable(identifier, metadataLocation, false))
         .isInstanceOf(AlreadyExistsException.class)
         .hasMessage("Table already exists: a.t1");
     Assertions.assertThat(ecsCatalog.dropTable(identifier, true)).isTrue();

hive-metastore/src/main/java/org/apache/iceberg/hive/HiveCatalog.java

@@ -45,6 +45,7 @@
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.hadoop.HadoopFileIO;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
@@ -169,10 +170,19 @@ public boolean dropTable(TableIdentifier identifier, boolean purge) {
 
     TableOperations ops = newTableOps(identifier);
     TableMetadata lastMetadata;
-    if (purge && ops.current() != null) {
-      lastMetadata = ops.current();
-    } else {
-      lastMetadata = null;
+    try {
+      if (purge && ops.current() != null) {
+        lastMetadata = ops.current();
+      } else {
+        lastMetadata = null;
+      }
+    } catch (Exception e) {
+      if (e instanceof NotFoundException) {
+        LOG.warn("The metadata file doesn't exist any more. Continue to drop the table.", e);
+        lastMetadata = null;
+      } else {
+        throw e;
+      }
     }
 
     try {

hive-metastore/src/test/java/org/apache/iceberg/hive/HiveTableTest.java

@@ -431,7 +431,7 @@ public void testRegisterTable() throws TException {
     List<String> metadataVersionFiles = metadataVersionFiles(TABLE_NAME);
     Assert.assertEquals(1, metadataVersionFiles.size());
 
-    catalog.registerTable(TABLE_IDENTIFIER, "file:" + metadataVersionFiles.get(0));
+    catalog.registerTable(TABLE_IDENTIFIER, "file:" + metadataVersionFiles.get(0), false);
 
     org.apache.hadoop.hive.metastore.api.Table newTable =
         metastoreClient.getTable(DB_NAME, TABLE_NAME);
@@ -482,7 +482,7 @@ public void testRegisterHadoopTableToHiveCatalog() throws IOException, TExceptio
 
     // register the table to hive catalog using the latest metadata file
     String latestMetadataFile = ((BaseTable) table).operations().current().metadataFileLocation();
-    catalog.registerTable(identifier, "file:" + latestMetadataFile);
+    catalog.registerTable(identifier, "file:" + latestMetadataFile, false);
     Assert.assertNotNull(metastoreClient.getTable(DB_NAME, "table1"));
 
     // load the table in hive catalog
@@ -546,7 +546,8 @@ public void testRegisterExistingTable() throws TException {
         "Should complain that the table already exists",
         AlreadyExistsException.class,
         "Table already exists",
-        () -> catalog.registerTable(TABLE_IDENTIFIER, "file:" + metadataVersionFiles.get(0)));
+        () ->
+            catalog.registerTable(TABLE_IDENTIFIER, "file:" + metadataVersionFiles.get(0), false));
   }
 
   @Test

hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCatalog.java

@@ -53,6 +53,7 @@
 import org.apache.iceberg.SortOrderParser;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.Transaction;
 import org.apache.iceberg.UpdateSchema;
@@ -62,6 +63,7 @@
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -443,6 +445,7 @@ public void testDropNamespace() throws TException {
         () -> {
           catalog.dropNamespace(namespace);
         });
+    final String s = catalog.newTableOps(identifier).current().metadataFileLocation();
     Assert.assertTrue(catalog.dropTable(identifier, true));
     Assert.assertTrue(
         "Should fail to drop namespace if it is not empty", catalog.dropNamespace(namespace));
@@ -458,6 +461,38 @@ public void testDropNamespace() throws TException {
         });
   }
 
+  @Test
+  public void testDropTable() throws TException {
+    Namespace namespace = Namespace.of("dbname_drop");
+    TableIdentifier identifier = TableIdentifier.of(namespace, "table");
+    Schema tableSchema =
+        new Schema(Types.StructType.of(required(1, "id", Types.LongType.get())).fields());
+    catalog.createNamespace(namespace, meta);
+    catalog.createTable(identifier, tableSchema);
+    TableOperations ops = catalog.newTableOps(identifier);
+
+    Assert.assertTrue(catalog.dropTable(identifier, true));
+    AssertHelpers.assertThrows(
+        "Should fail to load table after dropping it",
+        NoSuchTableException.class,
+        () -> catalog.loadTable(identifier)
+    );
+
+    // delete corrupted table
+    catalog.createTable(identifier, tableSchema);
+    String metadataFileLocation = catalog.newTableOps(identifier).current().metadataFileLocation();
+    ops.io().deleteFile(metadataFileLocation);
+    Assert.assertTrue(catalog.dropTable(identifier, true));
+    AssertHelpers.assertThrows(
+        "Should fail to load table after dropping it",
+        NoSuchTableException.class,
+        () -> catalog.loadTable(identifier)
+    );
+
+    // delete table that does not exist
+    Assert.assertFalse(catalog.dropTable(identifier, true));
+  }
+
   @Test
   public void testTableName() {
     Schema schema =

nessie/src/test/java/org/apache/iceberg/nessie/TestNessieTable.java

@@ -414,7 +414,7 @@ public void testDropTable() throws IOException {
   }
 
   private void validateRegister(TableIdentifier identifier, String metadataVersionFiles) {
-    Assertions.assertThat(catalog.registerTable(identifier, "file:" + metadataVersionFiles))
+    Assertions.assertThat(catalog.registerTable(identifier, "file:" + metadataVersionFiles, false))
         .isNotNull();
     Table newTable = catalog.loadTable(identifier);
     Assertions.assertThat(newTable).isNotNull();
@@ -445,12 +445,16 @@ public void testRegisterTableFailureScenarios()
     TableIdentifier defaultIdentifier =
         TableIdentifier.of(DB_NAME, defaultTableReference.toString());
     Assertions.assertThatThrownBy(
-            () -> catalog.registerTable(defaultIdentifier, "file:" + metadataVersionFiles.get(0)))
+            () ->
+                catalog.registerTable(
+                    defaultIdentifier, "file:" + metadataVersionFiles.get(0), false))
         .isInstanceOf(IllegalArgumentException.class)
         .hasMessage("Nessie ref 'default' does not exist");
     // Case 2: Table Already Exists
     Assertions.assertThatThrownBy(
-            () -> catalog.registerTable(TABLE_IDENTIFIER, "file:" + metadataVersionFiles.get(0)))
+            () ->
+                catalog.registerTable(
+                    TABLE_IDENTIFIER, "file:" + metadataVersionFiles.get(0), false))
         .isInstanceOf(AlreadyExistsException.class)
         .hasMessage("Table already exists: db.tbl");
     // Case 3: Registering using a tag
@@ -464,19 +468,21 @@ public void testRegisterTableFailureScenarios()
         ImmutableTableReference.builder().reference("tag_1").name(TABLE_NAME).build();
     TableIdentifier tagIdentifier = TableIdentifier.of(DB_NAME, tagTableReference.toString());
     Assertions.assertThatThrownBy(
-            () -> catalog.registerTable(tagIdentifier, "file:" + metadataVersionFiles.get(0)))
+            () ->
+                catalog.registerTable(tagIdentifier, "file:" + metadataVersionFiles.get(0), false))
         .isInstanceOf(IllegalArgumentException.class)
         .hasMessage("You can only mutate tables when using a branch without a hash or timestamp.");
     // Case 4: non-null metadata path with null metadata location
     Assertions.assertThatThrownBy(
             () ->
                 catalog.registerTable(
-                    TABLE_IDENTIFIER, "file:" + metadataVersionFiles.get(0) + "invalidName"))
+                    TABLE_IDENTIFIER, "file:" + metadataVersionFiles.get(0) + "invalidName", false))
         .isInstanceOf(NotFoundException.class);
     // Case 5: null identifier
     Assertions.assertThatThrownBy(
             () ->
-                catalog.registerTable(null, "file:" + metadataVersionFiles.get(0) + "invalidName"))
+                catalog.registerTable(
+                    null, "file:" + metadataVersionFiles.get(0) + "invalidName", false))
         .isInstanceOf(IllegalArgumentException.class)
         .hasMessage("Invalid identifier: null");
   }