Core: Avoid creating new metadata file when registerTable API is used

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

@@ -369,7 +369,11 @@ 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();
+    Table registeredTable = catalog.registerTable(identifier, metadataLocation);
+    Assertions.assertThat(registeredTable).isNotNull();
+    String expectedMetadataLocation =
+        ((HasTableOperations) registeredTable).operations().current().metadataFileLocation();
+    Assertions.assertThat(metadataLocation).isEqualTo(expectedMetadataLocation);
     Assertions.assertThat(catalog.loadTable(identifier)).isNotNull();
     Assertions.assertThat(catalog.dropTable(identifier, true)).isTrue();
     Assertions.assertThat(catalog.dropNamespace(namespace)).isTrue();

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

@@ -536,7 +536,11 @@ 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();
+    Table registeredTable = glueCatalog.registerTable(identifier, metadataLocation);
+    Assertions.assertThat(registeredTable).isNotNull();
+    String expectedMetadataLocation =
+        ((BaseTable) table).operations().current().metadataFileLocation();
+    Assertions.assertThat(metadataLocation).isEqualTo(expectedMetadataLocation);
     Assertions.assertThat(glueCatalog.loadTable(identifier)).isNotNull();
     Assertions.assertThat(glueCatalog.dropTable(identifier, true)).isTrue();
     Assertions.assertThat(glueCatalog.dropNamespace(Namespace.of(namespace))).isTrue();

aws/src/main/java/org/apache/iceberg/aws/dynamodb/DynamoDbTableOperations.java

@@ -102,7 +102,8 @@ protected void doRefresh() {
 
   @Override
   protected void doCommit(TableMetadata base, TableMetadata metadata) {
-    String newMetadataLocation = writeNewMetadata(metadata, currentVersion() + 1);
+    boolean newTable = base == null;
+    String newMetadataLocation = writeNewMetadataIfRequired(newTable, metadata);
     CommitStatus commitStatus = CommitStatus.FAILURE;
     Map<String, AttributeValue> tableKey = DynamoDbCatalog.tablePrimaryKey(tableIdentifier);
     try {

aws/src/main/java/org/apache/iceberg/aws/glue/GlueTableOperations.java

@@ -148,7 +148,8 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
     try {
       glueTempTableCreated = createGlueTempTableIfNecessary(base, metadata.location());
 
-      newMetadataLocation = writeNewMetadata(metadata, currentVersion() + 1);
+      boolean newTable = base == null;
+      newMetadataLocation = writeNewMetadataIfRequired(newTable, metadata);
       lock(newMetadataLocation);
       Table glueTable = getGlueTable();
       checkMetadataLocation(glueTable, base);

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

@@ -154,6 +154,12 @@ protected void disableRefresh() {
     this.shouldRefresh = false;
   }
 
+  protected String writeNewMetadataIfRequired(boolean newTable, TableMetadata metadata) {
+    return newTable && metadata.metadataFileLocation() != null
+        ? metadata.metadataFileLocation()
+        : writeNewMetadata(metadata, currentVersion() + 1);
+  }
+
   protected String writeNewMetadata(TableMetadata metadata, int newVersion) {
     String newTableMetadataFilePath = newTableMetadataFilePath(metadata, newVersion);
     OutputFile newMetadataLocation = io().newOutputFile(newTableMetadataFilePath);

core/src/main/java/org/apache/iceberg/jdbc/JdbcTableOperations.java

@@ -102,7 +102,8 @@ public void doRefresh() {
 
   @Override
   public void doCommit(TableMetadata base, TableMetadata metadata) {
-    String newMetadataLocation = writeNewMetadata(metadata, currentVersion() + 1);
+    boolean newTable = base == null;
+    String newMetadataLocation = writeNewMetadataIfRequired(newTable, metadata);
     try {
       Map<String, String> table = getTable();
 

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

@@ -782,6 +782,9 @@ public void testRegisterTable() {
     Table registeredTable = catalog.registerTable(identifier, metadataLocation);
     Assertions.assertThat(registeredTable).isNotNull();
     TestHelpers.assertSerializedAndLoadedMetadata(registeringTable, registeredTable);
+    String expectedMetadataLocation =
+        ((HasTableOperations) registeredTable).operations().current().metadataFileLocation();
+    Assertions.assertThat(metadataLocation).isEqualTo(expectedMetadataLocation);
     Assertions.assertThat(catalog.loadTable(identifier)).isNotNull();
     Assertions.assertThat(catalog.dropTable(identifier)).isTrue();
   }

dell/src/main/java/org/apache/iceberg/dell/ecs/EcsTableOperations.java

@@ -87,7 +87,8 @@ protected void doRefresh() {
 
   @Override
   protected void doCommit(TableMetadata base, TableMetadata metadata) {
-    String newMetadataLocation = writeNewMetadata(metadata, currentVersion() + 1);
+    boolean newTable = base == null;
+    String newMetadataLocation = writeNewMetadataIfRequired(newTable, metadata);
     if (base == null) {
       // create a new table, the metadataKey should be absent
       if (!catalog.putNewProperties(tableObject, buildProperties(newMetadataLocation))) {

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

@@ -190,7 +190,11 @@ public void testRegisterTable() {
     ecsCatalog.dropTable(identifier, false);
     TableOperations ops = ((HasTableOperations) registeringTable).operations();
     String metadataLocation = ((EcsTableOperations) ops).currentMetadataLocation();
-    Assertions.assertThat(ecsCatalog.registerTable(identifier, metadataLocation)).isNotNull();
+    Table registeredTable = ecsCatalog.registerTable(identifier, metadataLocation);
+    Assertions.assertThat(registeredTable).isNotNull();
+    String expectedMetadataLocation =
+        ((HasTableOperations) registeredTable).operations().current().metadataFileLocation();
+    Assertions.assertThat(metadataLocation).isEqualTo(expectedMetadataLocation);
     Assertions.assertThat(ecsCatalog.loadTable(identifier)).isNotNull();
     Assertions.assertThat(ecsCatalog.dropTable(identifier, true)).isTrue();
   }

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

@@ -268,10 +268,8 @@ protected void doRefresh() {
   @SuppressWarnings("checkstyle:CyclomaticComplexity")
   @Override
   protected void doCommit(TableMetadata base, TableMetadata metadata) {
-    String newMetadataLocation =
-        base == null && metadata.metadataFileLocation() != null
-            ? metadata.metadataFileLocation()
-            : writeNewMetadata(metadata, currentVersion() + 1);
+    boolean newTable = base == null;
+    String newMetadataLocation = writeNewMetadataIfRequired(newTable, metadata);
     boolean hiveEngineEnabled = hiveEngineEnabled(metadata, conf);
     boolean keepHiveStats = conf.getBoolean(ConfigProperties.KEEP_HIVE_STATS, false);
 
@@ -296,7 +294,7 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
       if (tbl != null) {
         // If we try to create the table but the metadata location is already set, then we had a
         // concurrent commit
-        if (base == null
+        if (newTable
             && tbl.getParameters().get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP)
                 != null) {
           throw new AlreadyExistsException("Table already exists: %s.%s", database, tableName);

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

@@ -28,6 +28,7 @@
 import static org.apache.iceberg.TableProperties.DEFAULT_SORT_ORDER;
 import static org.apache.iceberg.expressions.Expressions.bucket;
 import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -49,6 +50,7 @@
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
 import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.PartitionSpecParser;
 import org.apache.iceberg.Schema;
@@ -60,6 +62,7 @@
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TestHelpers;
 import org.apache.iceberg.Transaction;
 import org.apache.iceberg.UpdateSchema;
 import org.apache.iceberg.catalog.Catalog;
@@ -1176,4 +1179,35 @@ public void testDatabaseLocationWithSlashInWarehouseDir() {
 
     Assert.assertEquals("s3://bucket/database.db", database.getLocationUri());
   }
+
+  @Test
+  public void testRegisterTable() {
+    TableIdentifier identifier = TableIdentifier.of(DB_NAME, "t1");
+    catalog.createTable(identifier, getTestSchema());
+    Table registeringTable = catalog.loadTable(identifier);
+    catalog.dropTable(identifier, false);
+    TableOperations ops = ((HasTableOperations) registeringTable).operations();
+    String metadataLocation = ((HiveTableOperations) ops).currentMetadataLocation();
+    Table registeredTable = catalog.registerTable(identifier, metadataLocation);
+    assertThat(registeredTable).isNotNull();
+    TestHelpers.assertSerializedAndLoadedMetadata(registeringTable, registeredTable);
+    String expectedMetadataLocation =
+        ((HasTableOperations) registeredTable).operations().current().metadataFileLocation();
+    assertThat(metadataLocation).isEqualTo(expectedMetadataLocation);
+    assertThat(catalog.loadTable(identifier)).isNotNull();
+    assertThat(catalog.dropTable(identifier)).isTrue();
+  }
+
+  @Test
+  public void testRegisterExistingTable() {
+    TableIdentifier identifier = TableIdentifier.of(DB_NAME, "t1");
+    catalog.createTable(identifier, getTestSchema());
+    Table registeringTable = catalog.loadTable(identifier);
+    TableOperations ops = ((HasTableOperations) registeringTable).operations();
+    String metadataLocation = ((HiveTableOperations) ops).currentMetadataLocation();
+    assertThatThrownBy(() -> catalog.registerTable(identifier, metadataLocation))
+        .isInstanceOf(AlreadyExistsException.class)
+        .hasMessage("Table already exists: hivedb.t1");
+    assertThat(catalog.dropTable(identifier, true)).isTrue();
+  }
 }

nessie/src/main/java/org/apache/iceberg/nessie/NessieTableOperations.java

@@ -143,10 +143,8 @@ protected void doRefresh() {
 
   @Override
   protected void doCommit(TableMetadata base, TableMetadata metadata) {
-    String newMetadataLocation =
-        (base == null) && (metadata.metadataFileLocation() != null)
-            ? metadata.metadataFileLocation()
-            : writeNewMetadata(metadata, currentVersion() + 1);
+    boolean newTable = base == null;
+    String newMetadataLocation = writeNewMetadataIfRequired(newTable, metadata);
 
     String refName = client.refName();
     boolean delete = true;