Core: Implement BaseMetastoreCatalog.registerTable()

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

@@ -27,8 +27,10 @@
 import java.util.stream.IntStream;
 import org.apache.iceberg.AssertHelpers;
 import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.aws.AwsClientFactories;
 import org.apache.iceberg.aws.AwsClientFactory;
 import org.apache.iceberg.aws.AwsIntegTestUtil;
@@ -42,9 +44,11 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.types.Types;
+import org.assertj.core.api.Assertions;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
+import org.junit.Ignore;
 import org.junit.Test;
 import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
 import software.amazon.awssdk.services.dynamodb.model.DeleteTableRequest;
@@ -295,6 +299,30 @@ public void testDropNamespace() {
     Assert.assertFalse("namespace must not exist", response.hasItem());
   }
 
+  @Ignore
+  public void testRegisterTable() {
+    TableIdentifier identifier = TableIdentifier.of("a", "t1");
+    catalog.createTable(identifier, SCHEMA);
+    Table registeringTable = catalog.loadTable(identifier);
+    catalog.dropTable(identifier, false);
+    TableOperations ops = ((HasTableOperations) registeringTable).operations();
+    String metadataLocation = ((DynamoDbTableOperations) ops).currentMetadataLocation();
+    Assertions.assertThat(catalog.registerTable(identifier, metadataLocation)).isNotNull();
+    Table newTable = catalog.loadTable(identifier);
+    Assertions.assertThat(newTable).isNotNull();
+  }
+
+  @Ignore
+  public void testRegisterExistingTable() {
+    catalog.createTable(TableIdentifier.of("a", "t1"), SCHEMA);
+    Table registeringTable = catalog.loadTable(TableIdentifier.of("a", "t1"));
+    TableOperations ops = ((HasTableOperations) registeringTable).operations();
+    String metadataLocation = ((DynamoDbTableOperations) ops).currentMetadataLocation();
+    Assertions.assertThatThrownBy(() -> catalog.registerTable(TableIdentifier.of("a", "t1"), metadataLocation))
+        .isInstanceOf(AlreadyExistsException.class)
+        .hasMessage("Table already exists: a.t1");
+  }
+
   private static String genRandomName() {
     return UUID.randomUUID().toString().replace("-", "");
   }

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

@@ -25,6 +25,7 @@
 import java.util.stream.Collectors;
 import org.apache.iceberg.AssertHelpers;
 import org.apache.iceberg.BaseMetastoreTableOperations;
+import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
 import org.apache.iceberg.PartitionSpec;
@@ -43,6 +44,7 @@
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.types.Types.NestedField;
 import org.junit.Assert;
+import org.junit.Ignore;
 import org.junit.Test;
 import software.amazon.awssdk.services.glue.model.Column;
 import software.amazon.awssdk.services.glue.model.CreateTableRequest;
@@ -433,4 +435,28 @@ public void testTablePropsDefinedAtCatalogLevel() {
         "table-key5",
         table.properties().get("key5"));
   }
+
+  @Ignore
+  public void testRegisterTable() {
+    String namespace = createNamespace();
+    String tableName = getRandomName();
+    createTable(namespace, tableName);
+    Table table = glueCatalog.loadTable(TableIdentifier.of(namespace, tableName));
+    String metadataLocation = ((BaseTable) table).operations().current().metadataFileLocation();
+    glueCatalog.dropTable(TableIdentifier.of(namespace, tableName), false);
+    glueCatalog.registerTable(TableIdentifier.of(namespace, tableName), metadataLocation);
+  }
+
+  @Ignore
+  public void testRegisterTableAlreadyExists() {
+    String namespace = createNamespace();
+    String tableName = getRandomName();
+    createTable(namespace, tableName);
+    Table table = glueCatalog.loadTable(TableIdentifier.of(namespace, tableName));
+    String metadataLocation = ((BaseTable) table).operations().current().metadataFileLocation();
+    AssertHelpers.assertThrows("Should fail to register to an existing Glue table",
+        AlreadyExistsException.class,
+        "already exists in Glue",
+        () -> glueCatalog.registerTable(TableIdentifier.of(namespace, tableName), metadataLocation));
+  }
 }

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

@@ -25,6 +25,7 @@
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
@@ -65,6 +66,26 @@ public Table loadTable(TableIdentifier identifier) {
     return result;
   }
 
+  @Override
+  public Table registerTable(TableIdentifier identifier, String metadataFileLocation) {
+    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.
+    if (tableExists(identifier)) {
+      throw new AlreadyExistsException("Table already exists: %s", identifier);
+    }
+
+    TableOperations ops = newTableOps(identifier);
+    InputFile metadataFile = ops.io().newInputFile(metadataFileLocation);
+    TableMetadata metadata = TableMetadataParser.read(ops.io(), metadataFile);
+    ops.commit(null, metadata);
+
+    return new BaseTable(ops, identifier.toString());
+  }
+
   @Override
   public TableBuilder buildTable(TableIdentifier identifier, Schema schema) {
     return new BaseMetastoreCatalogTableBuilder(identifier, schema);

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

@@ -30,12 +30,15 @@
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.SortOrder;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.Transaction;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
@@ -47,6 +50,7 @@
 import org.apache.iceberg.transforms.Transform;
 import org.apache.iceberg.transforms.Transforms;
 import org.apache.iceberg.types.Types;
+import org.assertj.core.api.Assertions;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -589,4 +593,31 @@ public void testTablePropsDefinedAtCatalogLevel() throws IOException {
         "table-key5",
         table.properties().get("key5"));
   }
+
+  @Test
+  public void testRegisterTable() throws IOException {
+    TableIdentifier identifier = TableIdentifier.of("a", "t1");
+    TableIdentifier identifier2 = TableIdentifier.of("a", "t2");
+    HadoopCatalog catalog = hadoopCatalog();
+    catalog.createTable(identifier, SCHEMA);
+    Table registeringTable = catalog.loadTable(identifier);
+    TableOperations ops = ((HasTableOperations) registeringTable).operations();
+    String metadataLocation = ((HadoopTableOperations) ops).current().metadataFileLocation();
+    Assertions.assertThat(catalog.registerTable(identifier2, metadataLocation)).isNotNull();
+    Table newTable = catalog.loadTable(identifier2);
+    Assertions.assertThat(newTable).isNotNull();
+  }
+
+  @Test
+  public void testRegisterExistingTable() throws IOException {
+    TableIdentifier identifier = TableIdentifier.of("a", "t1");
+    HadoopCatalog catalog = hadoopCatalog();
+    catalog.createTable(identifier, SCHEMA);
+    Table registeringTable = catalog.loadTable(identifier);
+    TableOperations ops = ((HasTableOperations) registeringTable).operations();
+    String metadataLocation = ((HadoopTableOperations) ops).current().metadataFileLocation();
+    Assertions.assertThatThrownBy(() -> catalog.registerTable(identifier, metadataLocation))
+        .isInstanceOf(AlreadyExistsException.class)
+        .hasMessage("Table already exists: a.t1");
+  }
 }

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

@@ -37,10 +37,12 @@
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SortOrder;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.Transaction;
 import org.apache.iceberg.catalog.CatalogTests;
 import org.apache.iceberg.catalog.Namespace;
@@ -58,6 +60,7 @@
 import org.apache.iceberg.transforms.Transform;
 import org.apache.iceberg.transforms.Transforms;
 import org.apache.iceberg.types.Types;
+import org.assertj.core.api.Assertions;
 import org.junit.Assert;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -639,4 +642,28 @@ public void testConversions() {
     Assert.assertEquals(ns, JdbcUtil.stringToNamespace(nsString));
   }
 
+  @Test
+  public void testRegisterTable() {
+    TableIdentifier identifier = TableIdentifier.of("a", "t1");
+    catalog.createTable(identifier, SCHEMA);
+    Table registeringTable = catalog.loadTable(identifier);
+    catalog.dropTable(identifier, false);
+    TableOperations ops = ((HasTableOperations) registeringTable).operations();
+    String metadataLocation = ((JdbcTableOperations) ops).currentMetadataLocation();
+    Assertions.assertThat(catalog.registerTable(identifier, metadataLocation)).isNotNull();
+    Table newTable = catalog.loadTable(identifier);
+    Assertions.assertThat(newTable).isNotNull();
+  }
+
+  @Test
+  public void testRegisterExistingTable() {
+    TableIdentifier identifier = TableIdentifier.of("a", "t1");
+    catalog.createTable(identifier, SCHEMA);
+    Table registeringTable = catalog.loadTable(identifier);
+    TableOperations ops = ((HasTableOperations) registeringTable).operations();
+    String metadataLocation = ((JdbcTableOperations) ops).currentMetadataLocation();
+    Assertions.assertThatThrownBy(() -> catalog.registerTable(identifier, metadataLocation))
+        .isInstanceOf(AlreadyExistsException.class)
+        .hasMessage("Table already exists: a.t1");
+  }
 }

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

@@ -23,10 +23,14 @@
 import java.util.Map;
 import org.apache.iceberg.AssertHelpers;
 import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.dell.mock.ecs.EcsS3MockRule;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
@@ -35,6 +39,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.types.Types;
+import org.assertj.core.api.Assertions;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -170,4 +175,29 @@ public void testRenameTable() {
     Assert.assertFalse("Old table does not exist", ecsCatalog.tableExists(TableIdentifier.of("a", "t1")));
     Assert.assertTrue("New table exists", ecsCatalog.tableExists(TableIdentifier.of("b", "t2")));
   }
+
+  @Test
+  public void testRegisterTable() {
+    TableIdentifier identifier = TableIdentifier.of("a", "t1");
+    ecsCatalog.createTable(identifier, SCHEMA);
+    Table registeringTable = ecsCatalog.loadTable(identifier);
+    ecsCatalog.dropTable(identifier, false);
+    TableOperations ops = ((HasTableOperations) registeringTable).operations();
+    String metadataLocation = ((EcsTableOperations) ops).currentMetadataLocation();
+    Assertions.assertThat(ecsCatalog.registerTable(identifier, metadataLocation)).isNotNull();
+    Table newTable = ecsCatalog.loadTable(identifier);
+    Assertions.assertThat(newTable).isNotNull();
+  }
+
+  @Test
+  public void testRegisterExistingTable() {
+    TableIdentifier identifier = TableIdentifier.of("a", "t1");
+    ecsCatalog.createTable(identifier, SCHEMA);
+    Table registeringTable = ecsCatalog.loadTable(identifier);
+    TableOperations ops = ((HasTableOperations) registeringTable).operations();
+    String metadataLocation = ((EcsTableOperations) ops).currentMetadataLocation();
+    Assertions.assertThatThrownBy(() -> ecsCatalog.registerTable(identifier, metadataLocation))
+        .isInstanceOf(AlreadyExistsException.class)
+        .hasMessage("Table already exists: a.t1");
+  }
 }

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

@@ -36,12 +36,10 @@
 import org.apache.hadoop.hive.metastore.api.UnknownDBException;
 import org.apache.iceberg.BaseMetastoreCatalog;
 import org.apache.iceberg.BaseMetastoreTableOperations;
-import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.CatalogUtil;
 import org.apache.iceberg.ClientPool;
 import org.apache.iceberg.TableMetadata;
-import org.apache.iceberg.TableMetadataParser;
 import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
@@ -51,7 +49,6 @@
 import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.hadoop.HadoopFileIO;
 import org.apache.iceberg.io.FileIO;
-import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -234,23 +231,6 @@ public void renameTable(TableIdentifier from, TableIdentifier originalTo) {
     }
   }
 
-  @Override
-  public org.apache.iceberg.Table registerTable(TableIdentifier identifier, String metadataFileLocation) {
-    Preconditions.checkArgument(isValidIdentifier(identifier), "Invalid identifier: %s", identifier);
-
-    // Throw an exception if this table already exists in the catalog.
-    if (tableExists(identifier)) {
-      throw new org.apache.iceberg.exceptions.AlreadyExistsException("Table already exists: %s", identifier);
-    }
-
-    TableOperations ops = newTableOps(identifier);
-    InputFile metadataFile = fileIO.newInputFile(metadataFileLocation);
-    TableMetadata metadata = TableMetadataParser.read(ops.io(), metadataFile);
-    ops.commit(null, metadata);
-
-    return new BaseTable(ops, identifier.toString());
-  }
-
   @Override
   public void createNamespace(Namespace namespace, Map<String, String> meta) {
     Preconditions.checkArgument(

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

@@ -152,7 +152,8 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
       }
     }
 
-    String newMetadataLocation = writeNewMetadata(metadata, currentVersion() + 1);
+    String newMetadataLocation = (base == null) && (metadata.metadataFileLocation() != null) ?
+        metadata.metadataFileLocation() : writeNewMetadata(metadata, currentVersion() + 1);
 
     boolean delete = true;
     try {