From 73ecc5d3d9bd6c0966e03502b10f450168e4456b Mon Sep 17 00:00:00 2001 From: Theo Xu Date: Fri, 3 Feb 2023 16:33:29 -0800 Subject: [PATCH 1/8] support registerTable in GlueCatalog --- .../aws/glue/TestGlueCatalogTable.java | 63 +++++++++++++++++-- .../apache/iceberg/aws/glue/GlueCatalog.java | 40 ++++++++++++ .../iceberg/aws/glue/GlueTableOperations.java | 2 +- .../iceberg/aws/glue/TestGlueCatalog.java | 26 ++++++++ 4 files changed, 125 insertions(+), 6 deletions(-) diff --git a/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java b/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java index 427952fa16ce..30b0dc0cf719 100644 --- a/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java +++ b/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java @@ -42,6 +42,7 @@ import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.exceptions.NoSuchNamespaceException; import org.apache.iceberg.exceptions.NoSuchTableException; import org.apache.iceberg.exceptions.ValidationException; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; @@ -563,13 +564,37 @@ public void testRegisterTable() { Table registeredTable = glueCatalog.registerTable(identifier, metadataLocation); Assertions.assertThat(registeredTable).isNotNull(); String expectedMetadataLocation = - ((BaseTable) table).operations().current().metadataFileLocation(); + ((BaseTable) registeredTable).operations().current().metadataFileLocation(); Assertions.assertThat(metadataLocation).isEqualTo(expectedMetadataLocation); + + GetTableResponse response = + glue.getTable(GetTableRequest.builder().databaseName(namespace).name(tableName).build()); + String actualMetadataLocationGlue = response.table() + .parameters().get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP); + + Assert.assertEquals( + "Glue Catalog Register Table should not submit a new commit", + expectedMetadataLocation, + actualMetadataLocationGlue); + Assertions.assertThat(glueCatalog.loadTable(identifier)).isNotNull(); Assertions.assertThat(glueCatalog.dropTable(identifier, true)).isTrue(); Assertions.assertThat(glueCatalog.dropNamespace(Namespace.of(namespace))).isTrue(); } + @Test + public void testRegisterTableNamespaceNotFound() { + 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 unknown namespace", + NoSuchNamespaceException.class, + "not found in Glue", + () -> glueCatalog.registerTable(TableIdentifier.of(getRandomName(), getRandomName()), metadataLocation)); + } + @Test public void testRegisterTableAlreadyExists() { String namespace = createNamespace(); @@ -578,10 +603,38 @@ 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)) - .isInstanceOf(AlreadyExistsException.class); - Assertions.assertThat(glueCatalog.dropTable(identifier, true)).isTrue(); - Assertions.assertThat(glueCatalog.dropNamespace(Namespace.of(namespace))).isTrue(); + Assertions.assertThat(glueCatalog.dropTable(identifier, false)).isTrue(); + Table registeredTable = glueCatalog.registerTable(identifier, metadataLocation); + Assertions.assertThat(registeredTable).isNotNull(); + + GetTableResponse response = + glue.getTable(GetTableRequest.builder().databaseName(namespace).name(tableName).build()); + Assert.assertEquals( + "external table type is set after register", "EXTERNAL_TABLE", response.table().tableType()); + String actualMetadataLocation = response.table() + .parameters().get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP); + Assert.assertEquals("metadata location should be updated with registerTable call", + metadataLocation, actualMetadataLocation); + + System.out.println("===" + metadataLocation); + + // commit new transaction, should create a new metadata file + DataFile dataFile = + DataFiles.builder(partitionSpec) + .withPath("/path/to/data-a.parquet") + .withFileSizeInBytes(10) + .withRecordCount(1) + .build(); + table.newAppend().appendFile(dataFile).commit(); + + metadataLocation = ((BaseTable) table).operations().current().metadataFileLocation(); + // update metadata location + glueCatalog.registerTable(identifier, metadataLocation); + response = glue.getTable(GetTableRequest.builder().databaseName(namespace).name(tableName).build()); + String updatedMetadataLocation = response.table() + .parameters().get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP); + Assert.assertEquals("metadata location should be updated with registerTable call", + metadataLocation, updatedMetadataLocation); } @Test diff --git a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java index db6e8ead7ea4..10b8f6f99d9f 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java +++ b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java @@ -21,6 +21,7 @@ import java.io.Closeable; import java.io.IOException; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Set; import java.util.stream.Collectors; @@ -78,6 +79,7 @@ import software.amazon.awssdk.services.glue.model.Table; import software.amazon.awssdk.services.glue.model.TableInput; import software.amazon.awssdk.services.glue.model.UpdateDatabaseRequest; +import software.amazon.awssdk.services.glue.model.UpdateTableRequest; public class GlueCatalog extends BaseMetastoreCatalog implements Closeable, SupportsNamespaces, Configurable { @@ -437,6 +439,44 @@ public void renameTable(TableIdentifier from, TableIdentifier to) { LOG.info("Successfully renamed table from {} to {}", from, to); } + @Override + public org.apache.iceberg.Table registerTable(TableIdentifier identifier, String metadataFileLocation) { + Preconditions.checkArgument(isValidIdentifier(identifier), + "Table identifier to register is invalid: " + identifier); + Preconditions.checkArgument(metadataFileLocation != null && !metadataFileLocation.isEmpty(), + "Cannot register an empty metadata file location as a table"); + + Map tableParameters = ImmutableMap.of( + BaseMetastoreTableOperations.TABLE_TYPE_PROP, + BaseMetastoreTableOperations.ICEBERG_TABLE_TYPE_VALUE.toLowerCase(Locale.ENGLISH), + BaseMetastoreTableOperations.METADATA_LOCATION_PROP, + metadataFileLocation); + + TableInput tableInput = TableInput.builder() + .name(IcebergToGlueConverter.getTableName(identifier, awsProperties.glueCatalogSkipNameValidation())) + .tableType(GlueTableOperations.GLUE_EXTERNAL_TABLE_TYPE) + .parameters(tableParameters) + .build(); + + try { + glue.createTable(CreateTableRequest.builder() + .databaseName(IcebergToGlueConverter.getDatabaseName(identifier, + awsProperties.glueCatalogSkipNameValidation())) + .tableInput(tableInput) + .build()); + } catch (software.amazon.awssdk.services.glue.model.AlreadyExistsException e) { + glue.updateTable(UpdateTableRequest.builder() + .databaseName(IcebergToGlueConverter.getDatabaseName(identifier, + awsProperties.glueCatalogSkipNameValidation())) + .tableInput(tableInput) + .build()); + } catch (EntityNotFoundException e) { + throw new NoSuchNamespaceException(e, "Namespace %s is not found in Glue", identifier.namespace()); + } + + return loadTable(identifier); + } + @Override public void createNamespace(Namespace namespace, Map metadata) { try { diff --git a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueTableOperations.java b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueTableOperations.java index 84887cf25302..afb796d63a5d 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueTableOperations.java +++ b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueTableOperations.java @@ -63,7 +63,7 @@ class GlueTableOperations extends BaseMetastoreTableOperations { // same as org.apache.hadoop.hive.metastore.TableType.EXTERNAL_TABLE // more details: https://docs.aws.amazon.com/glue/latest/webapi/API_TableInput.html - private static final String GLUE_EXTERNAL_TABLE_TYPE = "EXTERNAL_TABLE"; + static final String GLUE_EXTERNAL_TABLE_TYPE = "EXTERNAL_TABLE"; private final GlueClient glue; private final AwsProperties awsProperties; diff --git a/aws/src/test/java/org/apache/iceberg/aws/glue/TestGlueCatalog.java b/aws/src/test/java/org/apache/iceberg/aws/glue/TestGlueCatalog.java index a6877637a049..d0b9c0116d1f 100644 --- a/aws/src/test/java/org/apache/iceberg/aws/glue/TestGlueCatalog.java +++ b/aws/src/test/java/org/apache/iceberg/aws/glue/TestGlueCatalog.java @@ -387,6 +387,32 @@ public Object answer(InvocationOnMock invocation) throws Throwable { Assert.assertEquals(0, counter.get()); } + @Test + public void testRegisterTableInvalidIdentifier() { + AssertHelpers.assertThrows("Should not allow registering table with multi-level namespace", + IllegalArgumentException.class, + "Table identifier to register is invalid", + () -> glueCatalog.registerTable(TableIdentifier.of("a", "b", "name"), "s3://path")); + + AssertHelpers.assertThrows("Should not allow registering table with unsupported table name", + IllegalArgumentException.class, + "Table identifier to register is invalid", + () -> glueCatalog.registerTable(TableIdentifier.of("a", "$name"), "s3://path")); + } + + @Test + public void testRegisterTableWithBadLocation() { + AssertHelpers.assertThrows("Should not allow registering null location", + IllegalArgumentException.class, + "Cannot register an empty metadata file location as a table", + () -> glueCatalog.registerTable(TableIdentifier.of("a", "name"), null)); + + AssertHelpers.assertThrows("Should not allow registering empty location", + IllegalArgumentException.class, + "Cannot register an empty metadata file location as a table", + () -> glueCatalog.registerTable(TableIdentifier.of("a", "name"), "")); + } + @Test public void testCreateNamespace() { Mockito.doReturn(CreateDatabaseResponse.builder().build()) From 4ac803cb43802646a7e3faf208cdfa34809ebd8d Mon Sep 17 00:00:00 2001 From: Theo Xu Date: Fri, 3 Feb 2023 16:50:33 -0800 Subject: [PATCH 2/8] remove unnessary lines --- .../java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java b/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java index 30b0dc0cf719..27fd4eca60f8 100644 --- a/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java +++ b/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java @@ -616,8 +616,6 @@ public void testRegisterTableAlreadyExists() { Assert.assertEquals("metadata location should be updated with registerTable call", metadataLocation, actualMetadataLocation); - System.out.println("===" + metadataLocation); - // commit new transaction, should create a new metadata file DataFile dataFile = DataFiles.builder(partitionSpec) From 3509f3aa6bc8d49ac78bf387a108ae3c165f9369 Mon Sep 17 00:00:00 2001 From: Theo Xu Date: Tue, 7 Feb 2023 01:07:27 -0800 Subject: [PATCH 3/8] resolve comments in pr --- .../aws/glue/TestGlueCatalogTable.java | 38 +++++++---- .../apache/iceberg/aws/glue/GlueCatalog.java | 64 +++++++++++-------- .../iceberg/aws/glue/TestGlueCatalog.java | 12 ++-- 3 files changed, 71 insertions(+), 43 deletions(-) diff --git a/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java b/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java index 27fd4eca60f8..b1865a0f8094 100644 --- a/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java +++ b/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java @@ -569,8 +569,8 @@ public void testRegisterTable() { GetTableResponse response = glue.getTable(GetTableRequest.builder().databaseName(namespace).name(tableName).build()); - String actualMetadataLocationGlue = response.table() - .parameters().get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP); + String actualMetadataLocationGlue = + response.table().parameters().get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP); Assert.assertEquals( "Glue Catalog Register Table should not submit a new commit", @@ -589,10 +589,13 @@ public void testRegisterTableNamespaceNotFound() { 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 unknown namespace", + AssertHelpers.assertThrows( + "Should fail to register to an unknown namespace", NoSuchNamespaceException.class, "not found in Glue", - () -> glueCatalog.registerTable(TableIdentifier.of(getRandomName(), getRandomName()), metadataLocation)); + () -> + glueCatalog.registerTable( + TableIdentifier.of(getRandomName(), getRandomName()), metadataLocation)); } @Test @@ -610,11 +613,15 @@ public void testRegisterTableAlreadyExists() { GetTableResponse response = glue.getTable(GetTableRequest.builder().databaseName(namespace).name(tableName).build()); Assert.assertEquals( - "external table type is set after register", "EXTERNAL_TABLE", response.table().tableType()); - String actualMetadataLocation = response.table() - .parameters().get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP); - Assert.assertEquals("metadata location should be updated with registerTable call", - metadataLocation, actualMetadataLocation); + "external table type is set after register", + "EXTERNAL_TABLE", + response.table().tableType()); + String actualMetadataLocation = + response.table().parameters().get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP); + Assert.assertEquals( + "metadata location should be updated with registerTable call", + metadataLocation, + actualMetadataLocation); // commit new transaction, should create a new metadata file DataFile dataFile = @@ -628,11 +635,14 @@ public void testRegisterTableAlreadyExists() { metadataLocation = ((BaseTable) table).operations().current().metadataFileLocation(); // update metadata location glueCatalog.registerTable(identifier, metadataLocation); - response = glue.getTable(GetTableRequest.builder().databaseName(namespace).name(tableName).build()); - String updatedMetadataLocation = response.table() - .parameters().get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP); - Assert.assertEquals("metadata location should be updated with registerTable call", - metadataLocation, updatedMetadataLocation); + response = + glue.getTable(GetTableRequest.builder().databaseName(namespace).name(tableName).build()); + String updatedMetadataLocation = + response.table().parameters().get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP); + Assert.assertEquals( + "metadata location should be updated with registerTable call", + metadataLocation, + updatedMetadataLocation); } @Test diff --git a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java index 10b8f6f99d9f..300707045dcf 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java +++ b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java @@ -32,6 +32,7 @@ import org.apache.iceberg.CatalogUtil; import org.apache.iceberg.LockManager; import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.TableMetadataParser; import org.apache.iceberg.TableOperations; import org.apache.iceberg.aws.AwsClientFactories; import org.apache.iceberg.aws.AwsClientFactory; @@ -50,6 +51,7 @@ import org.apache.iceberg.hadoop.Configurable; import org.apache.iceberg.io.CloseableGroup; 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.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; @@ -440,38 +442,50 @@ public void renameTable(TableIdentifier from, TableIdentifier to) { } @Override - public org.apache.iceberg.Table registerTable(TableIdentifier identifier, String metadataFileLocation) { - Preconditions.checkArgument(isValidIdentifier(identifier), - "Table identifier to register is invalid: " + identifier); - Preconditions.checkArgument(metadataFileLocation != null && !metadataFileLocation.isEmpty(), + public org.apache.iceberg.Table registerTable( + TableIdentifier identifier, String metadataFileLocation) { + Preconditions.checkArgument( + isValidIdentifier(identifier), "Table identifier to register is invalid: " + identifier); + Preconditions.checkArgument( + metadataFileLocation != null && !metadataFileLocation.isEmpty(), "Cannot register an empty metadata file location as a table"); - Map tableParameters = ImmutableMap.of( - BaseMetastoreTableOperations.TABLE_TYPE_PROP, - BaseMetastoreTableOperations.ICEBERG_TABLE_TYPE_VALUE.toLowerCase(Locale.ENGLISH), - BaseMetastoreTableOperations.METADATA_LOCATION_PROP, - metadataFileLocation); + TableOperations ops = newTableOps(identifier); + InputFile metadataFile = ops.io().newInputFile(metadataFileLocation); + TableMetadata metadata = TableMetadataParser.read(ops.io(), metadataFile); - TableInput tableInput = TableInput.builder() - .name(IcebergToGlueConverter.getTableName(identifier, awsProperties.glueCatalogSkipNameValidation())) - .tableType(GlueTableOperations.GLUE_EXTERNAL_TABLE_TYPE) - .parameters(tableParameters) - .build(); + Map tableParameters = + ImmutableMap.of( + BaseMetastoreTableOperations.TABLE_TYPE_PROP, + BaseMetastoreTableOperations.ICEBERG_TABLE_TYPE_VALUE.toLowerCase(Locale.ENGLISH), + BaseMetastoreTableOperations.METADATA_LOCATION_PROP, + metadataFileLocation); + + String databaseName = + IcebergToGlueConverter.getDatabaseName( + identifier, awsProperties.glueCatalogSkipNameValidation()); + String tableName = + IcebergToGlueConverter.getTableName( + identifier, awsProperties.glueCatalogSkipNameValidation()); + + TableInput tableInput = + TableInput.builder() + .applyMutation( + builder -> IcebergToGlueConverter.setTableInputInformation(builder, metadata)) + .name(tableName) + .tableType(GlueTableOperations.GLUE_EXTERNAL_TABLE_TYPE) + .parameters(tableParameters) + .build(); try { - glue.createTable(CreateTableRequest.builder() - .databaseName(IcebergToGlueConverter.getDatabaseName(identifier, - awsProperties.glueCatalogSkipNameValidation())) - .tableInput(tableInput) - .build()); + glue.createTable( + CreateTableRequest.builder().databaseName(databaseName).tableInput(tableInput).build()); } catch (software.amazon.awssdk.services.glue.model.AlreadyExistsException e) { - glue.updateTable(UpdateTableRequest.builder() - .databaseName(IcebergToGlueConverter.getDatabaseName(identifier, - awsProperties.glueCatalogSkipNameValidation())) - .tableInput(tableInput) - .build()); + glue.updateTable( + UpdateTableRequest.builder().databaseName(databaseName).tableInput(tableInput).build()); } catch (EntityNotFoundException e) { - throw new NoSuchNamespaceException(e, "Namespace %s is not found in Glue", identifier.namespace()); + throw new NoSuchNamespaceException( + e, "Namespace %s is not found in Glue", identifier.namespace()); } return loadTable(identifier); diff --git a/aws/src/test/java/org/apache/iceberg/aws/glue/TestGlueCatalog.java b/aws/src/test/java/org/apache/iceberg/aws/glue/TestGlueCatalog.java index d0b9c0116d1f..e14ee8b6fd9e 100644 --- a/aws/src/test/java/org/apache/iceberg/aws/glue/TestGlueCatalog.java +++ b/aws/src/test/java/org/apache/iceberg/aws/glue/TestGlueCatalog.java @@ -389,12 +389,14 @@ public Object answer(InvocationOnMock invocation) throws Throwable { @Test public void testRegisterTableInvalidIdentifier() { - AssertHelpers.assertThrows("Should not allow registering table with multi-level namespace", + AssertHelpers.assertThrows( + "Should not allow registering table with multi-level namespace", IllegalArgumentException.class, "Table identifier to register is invalid", () -> glueCatalog.registerTable(TableIdentifier.of("a", "b", "name"), "s3://path")); - AssertHelpers.assertThrows("Should not allow registering table with unsupported table name", + AssertHelpers.assertThrows( + "Should not allow registering table with unsupported table name", IllegalArgumentException.class, "Table identifier to register is invalid", () -> glueCatalog.registerTable(TableIdentifier.of("a", "$name"), "s3://path")); @@ -402,12 +404,14 @@ public void testRegisterTableInvalidIdentifier() { @Test public void testRegisterTableWithBadLocation() { - AssertHelpers.assertThrows("Should not allow registering null location", + AssertHelpers.assertThrows( + "Should not allow registering null location", IllegalArgumentException.class, "Cannot register an empty metadata file location as a table", () -> glueCatalog.registerTable(TableIdentifier.of("a", "name"), null)); - AssertHelpers.assertThrows("Should not allow registering empty location", + AssertHelpers.assertThrows( + "Should not allow registering empty location", IllegalArgumentException.class, "Cannot register an empty metadata file location as a table", () -> glueCatalog.registerTable(TableIdentifier.of("a", "name"), "")); From 3c25e64089098f1af21cfe8d4948cf2ea2973756 Mon Sep 17 00:00:00 2001 From: Theo Xu Date: Tue, 7 Feb 2023 13:49:54 -0800 Subject: [PATCH 4/8] add feature flag to registerTable & specify versionId to ensure atomic transaction --- .../apache/iceberg/aws/glue/GlueTestBase.java | 12 ++++++++++ .../aws/glue/TestGlueCatalogTable.java | 19 +++++---------- .../org/apache/iceberg/aws/AwsProperties.java | 24 +++++++++++++++++++ .../apache/iceberg/aws/glue/GlueCatalog.java | 16 +++++++++++-- 4 files changed, 56 insertions(+), 15 deletions(-) diff --git a/aws/src/integration/java/org/apache/iceberg/aws/glue/GlueTestBase.java b/aws/src/integration/java/org/apache/iceberg/aws/glue/GlueTestBase.java index 359d06db0494..18175c9fcea6 100644 --- a/aws/src/integration/java/org/apache/iceberg/aws/glue/GlueTestBase.java +++ b/aws/src/integration/java/org/apache/iceberg/aws/glue/GlueTestBase.java @@ -63,6 +63,7 @@ public class GlueTestBase { static GlueCatalog glueCatalog; static GlueCatalog glueCatalogWithSkip; static GlueCatalog glueCatalogWithSkipNameValidation; + static GlueCatalog glueCatalogWithForceRegisterTable; static Schema schema = new Schema(Types.NestedField.required(1, "c1", Types.StringType.get(), "c1")); @@ -111,6 +112,17 @@ public static void beforeClass() { LockManagers.defaultLockManager(), fileIO, ImmutableMap.of()); + glueCatalogWithForceRegisterTable = new GlueCatalog(); + AwsProperties propertiesForceRegisterTable = new AwsProperties(); + propertiesForceRegisterTable.setGlueCatalogForceRegisterTable(true); + glueCatalogWithForceRegisterTable.initialize( + catalogName, + testBucketPath, + propertiesForceRegisterTable, + glue, + LockManagers.defaultLockManager(), + fileIO, + ImmutableMap.of()); } @AfterClass diff --git a/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java b/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java index b1865a0f8094..7ef0dee3224e 100644 --- a/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java +++ b/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java @@ -44,7 +44,6 @@ import org.apache.iceberg.exceptions.AlreadyExistsException; import org.apache.iceberg.exceptions.NoSuchNamespaceException; import org.apache.iceberg.exceptions.NoSuchTableException; -import org.apache.iceberg.exceptions.ValidationException; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Maps; @@ -270,14 +269,6 @@ public void testRenameTableFailsToDeleteOldTable() { .databaseName(namespace) .tableInput(TableInput.builder().name(tableName).parameters(Maps.newHashMap()).build()) .build()); - AssertHelpers.assertThrows( - "should fail to rename", - ValidationException.class, - "Input Glue table is not an iceberg table", - () -> - glueCatalog.renameTable( - TableIdentifier.of(namespace, tableName), - TableIdentifier.of(namespace, newTableName))); AssertHelpers.assertThrows( "renamed table should be deleted", EntityNotFoundException.class, @@ -604,10 +595,11 @@ public void testRegisterTableAlreadyExists() { String tableName = getRandomName(); createTable(namespace, tableName); TableIdentifier identifier = TableIdentifier.of(namespace, tableName); - Table table = glueCatalog.loadTable(identifier); + Table table = glueCatalogWithForceRegisterTable.loadTable(identifier); String metadataLocation = ((BaseTable) table).operations().current().metadataFileLocation(); - Assertions.assertThat(glueCatalog.dropTable(identifier, false)).isTrue(); - Table registeredTable = glueCatalog.registerTable(identifier, metadataLocation); + Assertions.assertThat(glueCatalogWithForceRegisterTable.dropTable(identifier, false)).isTrue(); + Table registeredTable = + glueCatalogWithForceRegisterTable.registerTable(identifier, metadataLocation); Assertions.assertThat(registeredTable).isNotNull(); GetTableResponse response = @@ -634,7 +626,7 @@ public void testRegisterTableAlreadyExists() { metadataLocation = ((BaseTable) table).operations().current().metadataFileLocation(); // update metadata location - glueCatalog.registerTable(identifier, metadataLocation); + glueCatalogWithForceRegisterTable.registerTable(identifier, metadataLocation); response = glue.getTable(GetTableRequest.builder().databaseName(namespace).name(tableName).build()); String updatedMetadataLocation = @@ -643,6 +635,7 @@ public void testRegisterTableAlreadyExists() { "metadata location should be updated with registerTable call", metadataLocation, updatedMetadataLocation); + Assert.assertEquals("Table Version should be updated", "2", response.table().versionId()); } @Test diff --git a/aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java b/aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java index 7cf46d7c3f3b..38a1885563f4 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java +++ b/aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java @@ -164,6 +164,15 @@ public class AwsProperties implements Serializable { */ public static final String GLUE_CATALOG_ENDPOINT = "glue.endpoint"; + /** + * If set, Glue will always update the catalog table if the table already exists in glue catalog. + * By default, Glue catalog will only be able to create new table and will throw + * AlreadyExistsException when register an existing table name. + */ + public static final String GLUE_CATALOG_FORCE_REGISTER_TABLE = "glue.force-register-table"; + + public static final boolean GLUE_CATALOG_FORCE_REGISTER_TABLE_DEFAULT = false; + /** * Number of threads to use for uploading parts to S3 (shared pool across all output streams), * default to {@link Runtime#availableProcessors()} @@ -684,6 +693,7 @@ public class AwsProperties implements Serializable { private boolean glueCatalogSkipArchive; private boolean glueCatalogSkipNameValidation; private boolean glueLakeFormationEnabled; + private boolean glueCatalogForceRegisterTable; private String dynamoDbTableName; private String dynamoDbEndpoint; @@ -744,6 +754,7 @@ public AwsProperties() { this.glueCatalogSkipArchive = GLUE_CATALOG_SKIP_ARCHIVE_DEFAULT; this.glueCatalogSkipNameValidation = GLUE_CATALOG_SKIP_NAME_VALIDATION_DEFAULT; this.glueLakeFormationEnabled = GLUE_LAKEFORMATION_ENABLED_DEFAULT; + this.glueCatalogForceRegisterTable = GLUE_CATALOG_FORCE_REGISTER_TABLE_DEFAULT; this.dynamoDbEndpoint = null; this.dynamoDbTableName = DYNAMODB_TABLE_NAME_DEFAULT; @@ -826,6 +837,11 @@ public AwsProperties(Map properties) { this.glueLakeFormationEnabled = PropertyUtil.propertyAsBoolean( properties, GLUE_LAKEFORMATION_ENABLED, GLUE_LAKEFORMATION_ENABLED_DEFAULT); + this.glueCatalogForceRegisterTable = + PropertyUtil.propertyAsBoolean( + properties, + GLUE_CATALOG_FORCE_REGISTER_TABLE, + GLUE_CATALOG_FORCE_REGISTER_TABLE_DEFAULT); this.s3FileIoMultipartUploadThreads = PropertyUtil.propertyAsInt( properties, @@ -1000,6 +1016,14 @@ public void setGlueLakeFormationEnabled(boolean glueLakeFormationEnabled) { this.glueLakeFormationEnabled = glueLakeFormationEnabled; } + public boolean glueCatalogForceRegisterTable() { + return glueCatalogForceRegisterTable; + } + + public void setGlueCatalogForceRegisterTable(boolean glueCatalogForceRegisterTable) { + this.glueCatalogForceRegisterTable = glueCatalogForceRegisterTable; + } + public int s3FileIoMultipartUploadThreads() { return s3FileIoMultipartUploadThreads; } diff --git a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java index 300707045dcf..3f6c5ffa465a 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java +++ b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java @@ -481,8 +481,20 @@ public org.apache.iceberg.Table registerTable( glue.createTable( CreateTableRequest.builder().databaseName(databaseName).tableInput(tableInput).build()); } catch (software.amazon.awssdk.services.glue.model.AlreadyExistsException e) { - glue.updateTable( - UpdateTableRequest.builder().databaseName(databaseName).tableInput(tableInput).build()); + if (awsProperties.glueCatalogForceRegisterTable()) { + GetTableResponse response = + glue.getTable( + GetTableRequest.builder().databaseName(databaseName).name(tableName).build()); + String versionId = response.table().versionId(); + glue.updateTable( + UpdateTableRequest.builder() + .databaseName(databaseName) + .tableInput(tableInput) + .versionId(versionId) + .build()); + } else { + throw new AlreadyExistsException("Table already exists: %s", identifier); + } } catch (EntityNotFoundException e) { throw new NoSuchNamespaceException( e, "Namespace %s is not found in Glue", identifier.namespace()); From 8e5349bbf53f9d15f485bd69e16fb1dcdf0ac625 Mon Sep 17 00:00:00 2001 From: Theo Xu Date: Tue, 7 Feb 2023 14:37:59 -0800 Subject: [PATCH 5/8] resolve comments to optimize the flag behavior in registerTable --- .../aws/glue/TestGlueCatalogTable.java | 5 ++-- .../apache/iceberg/aws/glue/GlueCatalog.java | 29 ++++++++++--------- 2 files changed, 18 insertions(+), 16 deletions(-) diff --git a/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java b/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java index 7ef0dee3224e..6ba897906786 100644 --- a/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java +++ b/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java @@ -578,14 +578,15 @@ public void testRegisterTableNamespaceNotFound() { String namespace = createNamespace(); String tableName = getRandomName(); createTable(namespace, tableName); - Table table = glueCatalog.loadTable(TableIdentifier.of(namespace, tableName)); + Table table = + glueCatalogWithForceRegisterTable.loadTable(TableIdentifier.of(namespace, tableName)); String metadataLocation = ((BaseTable) table).operations().current().metadataFileLocation(); AssertHelpers.assertThrows( "Should fail to register to an unknown namespace", NoSuchNamespaceException.class, "not found in Glue", () -> - glueCatalog.registerTable( + glueCatalogWithForceRegisterTable.registerTable( TableIdentifier.of(getRandomName(), getRandomName()), metadataLocation)); } diff --git a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java index 3f6c5ffa465a..9feb42a806ef 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java +++ b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java @@ -450,6 +450,11 @@ public org.apache.iceberg.Table registerTable( metadataFileLocation != null && !metadataFileLocation.isEmpty(), "Cannot register an empty metadata file location as a table"); + // keep the original behavior when force-register-table flag is off + if (!awsProperties.glueCatalogForceRegisterTable()) { + return super.registerTable(identifier, metadataFileLocation); + } + TableOperations ops = newTableOps(identifier); InputFile metadataFile = ops.io().newInputFile(metadataFileLocation); TableMetadata metadata = TableMetadataParser.read(ops.io(), metadataFile); @@ -481,20 +486,16 @@ public org.apache.iceberg.Table registerTable( glue.createTable( CreateTableRequest.builder().databaseName(databaseName).tableInput(tableInput).build()); } catch (software.amazon.awssdk.services.glue.model.AlreadyExistsException e) { - if (awsProperties.glueCatalogForceRegisterTable()) { - GetTableResponse response = - glue.getTable( - GetTableRequest.builder().databaseName(databaseName).name(tableName).build()); - String versionId = response.table().versionId(); - glue.updateTable( - UpdateTableRequest.builder() - .databaseName(databaseName) - .tableInput(tableInput) - .versionId(versionId) - .build()); - } else { - throw new AlreadyExistsException("Table already exists: %s", identifier); - } + GetTableResponse response = + glue.getTable( + GetTableRequest.builder().databaseName(databaseName).name(tableName).build()); + String versionId = response.table().versionId(); + glue.updateTable( + UpdateTableRequest.builder() + .databaseName(databaseName) + .tableInput(tableInput) + .versionId(versionId) + .build()); } catch (EntityNotFoundException e) { throw new NoSuchNamespaceException( e, "Namespace %s is not found in Glue", identifier.namespace()); From 043b8a6b9c40e9752676a21e1c69753f1221df7d Mon Sep 17 00:00:00 2001 From: Theo Xu Date: Fri, 7 Apr 2023 01:18:23 -0700 Subject: [PATCH 6/8] support glue endpoint for assumeRoleAwsClientFactory --- .../java/org/apache/iceberg/aws/AssumeRoleAwsClientFactory.java | 1 + 1 file changed, 1 insertion(+) diff --git a/aws/src/main/java/org/apache/iceberg/aws/AssumeRoleAwsClientFactory.java b/aws/src/main/java/org/apache/iceberg/aws/AssumeRoleAwsClientFactory.java index c7e93879921a..562b205f2e3f 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/AssumeRoleAwsClientFactory.java +++ b/aws/src/main/java/org/apache/iceberg/aws/AssumeRoleAwsClientFactory.java @@ -52,6 +52,7 @@ public GlueClient glue() { return GlueClient.builder() .applyMutation(this::applyAssumeRoleConfigurations) .applyMutation(awsProperties::applyHttpClientConfigurations) + .applyMutation(awsProperties::applyGlueEndpointConfigurations) .build(); } From 4bdb165dc88698b18972560c80d0dad0d3ffaf7e Mon Sep 17 00:00:00 2001 From: Theo Xu Date: Tue, 11 Apr 2023 13:37:43 -0700 Subject: [PATCH 7/8] support pass in table paramters to storagedescriptor --- .../java/org/apache/iceberg/aws/glue/GlueCatalog.java | 4 +++- .../apache/iceberg/aws/glue/GlueTableOperations.java | 6 ++++-- .../iceberg/aws/glue/IcebergToGlueConverter.java | 10 ++++++++-- .../iceberg/aws/glue/TestIcebergToGlueConverter.java | 7 +++++-- 4 files changed, 20 insertions(+), 7 deletions(-) diff --git a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java index 9feb42a806ef..afc3d736cb76 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java +++ b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java @@ -476,7 +476,9 @@ public org.apache.iceberg.Table registerTable( TableInput tableInput = TableInput.builder() .applyMutation( - builder -> IcebergToGlueConverter.setTableInputInformation(builder, metadata)) + builder -> + IcebergToGlueConverter.setTableInputInformation( + builder, metadata, tableParameters)) .name(tableName) .tableType(GlueTableOperations.GLUE_EXTERNAL_TABLE_TYPE) .parameters(tableParameters) diff --git a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueTableOperations.java b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueTableOperations.java index afb796d63a5d..b2f3a7164cae 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueTableOperations.java +++ b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueTableOperations.java @@ -310,7 +310,8 @@ void persistGlueTable(Table glueTable, Map parameters, TableMeta TableInput.builder() .applyMutation( builder -> - IcebergToGlueConverter.setTableInputInformation(builder, metadata)) + IcebergToGlueConverter.setTableInputInformation( + builder, metadata, parameters)) .name(tableName) .tableType(GLUE_EXTERNAL_TABLE_TYPE) .parameters(parameters) @@ -331,7 +332,8 @@ void persistGlueTable(Table glueTable, Map parameters, TableMeta TableInput.builder() .applyMutation( builder -> - IcebergToGlueConverter.setTableInputInformation(builder, metadata)) + IcebergToGlueConverter.setTableInputInformation( + builder, metadata, parameters)) .name(tableName) .tableType(GLUE_EXTERNAL_TABLE_TYPE) .parameters(parameters) diff --git a/aws/src/main/java/org/apache/iceberg/aws/glue/IcebergToGlueConverter.java b/aws/src/main/java/org/apache/iceberg/aws/glue/IcebergToGlueConverter.java index 241c0098d628..3e0358a7468a 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/glue/IcebergToGlueConverter.java +++ b/aws/src/main/java/org/apache/iceberg/aws/glue/IcebergToGlueConverter.java @@ -226,7 +226,9 @@ static void validateTableIdentifier(TableIdentifier tableIdentifier) { * @param metadata Iceberg table metadata */ static void setTableInputInformation( - TableInput.Builder tableInputBuilder, TableMetadata metadata) { + TableInput.Builder tableInputBuilder, + TableMetadata metadata, + Map parameters) { try { StorageDescriptor.Builder storageDescriptor = StorageDescriptor.builder(); if (!SET_ADDITIONAL_LOCATIONS.isNoop()) { @@ -239,7 +241,11 @@ static void setTableInputInformation( } tableInputBuilder.storageDescriptor( - storageDescriptor.location(metadata.location()).columns(toColumns(metadata)).build()); + storageDescriptor + .location(metadata.location()) + .columns(toColumns(metadata)) + .parameters(parameters) + .build()); } catch (RuntimeException e) { LOG.warn( "Encountered unexpected exception while converting Iceberg metadata to Glue table information", diff --git a/aws/src/test/java/org/apache/iceberg/aws/glue/TestIcebergToGlueConverter.java b/aws/src/test/java/org/apache/iceberg/aws/glue/TestIcebergToGlueConverter.java index 701416de2f8d..1e49c9673ddd 100644 --- a/aws/src/test/java/org/apache/iceberg/aws/glue/TestIcebergToGlueConverter.java +++ b/aws/src/test/java/org/apache/iceberg/aws/glue/TestIcebergToGlueConverter.java @@ -18,6 +18,7 @@ */ package org.apache.iceberg.aws.glue; +import java.util.Collections; import java.util.List; import java.util.Map; import org.apache.iceberg.AssertHelpers; @@ -162,7 +163,8 @@ public void testSetTableInputInformation() { PartitionSpec.builderFor(schema).identity("x").withSpecId(1000).build(); TableMetadata tableMetadata = TableMetadata.newTableMetadata(schema, partitionSpec, "s3://test", tableLocationProperties); - IcebergToGlueConverter.setTableInputInformation(actualTableInputBuilder, tableMetadata); + IcebergToGlueConverter.setTableInputInformation( + actualTableInputBuilder, tableMetadata, Collections.emptyMap()); TableInput actualTableInput = actualTableInputBuilder.build(); // Expected TableInput @@ -231,7 +233,8 @@ public void testSetTableInputInformationWithRemovedColumns() { Schema newSchema = new Schema(Types.NestedField.required(1, "x", Types.StringType.get(), "comment1")); tableMetadata = tableMetadata.updateSchema(newSchema, 3); - IcebergToGlueConverter.setTableInputInformation(actualTableInputBuilder, tableMetadata); + IcebergToGlueConverter.setTableInputInformation( + actualTableInputBuilder, tableMetadata, Collections.emptyMap()); TableInput actualTableInput = actualTableInputBuilder.build(); // Expected TableInput From b38a928686d6c2b3fd00a306b3f6c08aed8b9b4a Mon Sep 17 00:00:00 2001 From: Theo Xu Date: Fri, 28 Apr 2023 16:34:32 -0700 Subject: [PATCH 8/8] allow s3 file io cross region call for glue catalog register table --- .../aws/glue/TestGlueCatalogTable.java | 41 +++++++++++++++++-- .../org/apache/iceberg/aws/AwsProperties.java | 15 +++++++ .../apache/iceberg/aws/glue/GlueCatalog.java | 15 ++++++- 3 files changed, 67 insertions(+), 4 deletions(-) diff --git a/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java b/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java index 6ba897906786..4a9250f50a04 100644 --- a/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java +++ b/aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java @@ -555,7 +555,28 @@ public void testRegisterTable() { Table registeredTable = glueCatalog.registerTable(identifier, metadataLocation); Assertions.assertThat(registeredTable).isNotNull(); String expectedMetadataLocation = - ((BaseTable) registeredTable).operations().current().metadataFileLocation(); + ((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(); + } + + @Test + public void testRegisterTableForceRegister() { + String namespace = createNamespace(); + String tableName = getRandomName(); + createTable(namespace, tableName); + TableIdentifier identifier = TableIdentifier.of(namespace, tableName); + Table table = glueCatalogWithForceRegisterTable.loadTable(identifier); + String metadataLocation = ((BaseTable) table).operations().current().metadataFileLocation(); + Assertions.assertThat(glueCatalogWithForceRegisterTable.dropTable(identifier, false)).isTrue(); + Table registeredTable = + glueCatalogWithForceRegisterTable.registerTable(identifier, metadataLocation); + Assertions.assertThat(registeredTable).isNotNull(); + String expectedMetadataLocation = + ((BaseTable) table).operations().current().metadataFileLocation(); Assertions.assertThat(metadataLocation).isEqualTo(expectedMetadataLocation); GetTableResponse response = @@ -568,8 +589,8 @@ public void testRegisterTable() { expectedMetadataLocation, actualMetadataLocationGlue); - Assertions.assertThat(glueCatalog.loadTable(identifier)).isNotNull(); - Assertions.assertThat(glueCatalog.dropTable(identifier, true)).isTrue(); + Assertions.assertThat(glueCatalogWithForceRegisterTable.loadTable(identifier)).isNotNull(); + Assertions.assertThat(glueCatalogWithForceRegisterTable.dropTable(identifier, true)).isTrue(); Assertions.assertThat(glueCatalog.dropNamespace(Namespace.of(namespace))).isTrue(); } @@ -592,6 +613,20 @@ public void testRegisterTableNamespaceNotFound() { @Test public void testRegisterTableAlreadyExists() { + String namespace = createNamespace(); + String tableName = getRandomName(); + createTable(namespace, tableName); + TableIdentifier identifier = TableIdentifier.of(namespace, tableName); + Table table = glueCatalog.loadTable(identifier); + String metadataLocation = ((BaseTable) table).operations().current().metadataFileLocation(); + Assertions.assertThatThrownBy(() -> glueCatalog.registerTable(identifier, metadataLocation)) + .isInstanceOf(AlreadyExistsException.class); + Assertions.assertThat(glueCatalog.dropTable(identifier, true)).isTrue(); + Assertions.assertThat(glueCatalog.dropNamespace(Namespace.of(namespace))).isTrue(); + } + + @Test + public void testRegisterTableAlreadyExistsForceRegister() { String namespace = createNamespace(); String tableName = getRandomName(); createTable(namespace, tableName); diff --git a/aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java b/aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java index 38a1885563f4..1582aacc0fab 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java +++ b/aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java @@ -173,6 +173,9 @@ public class AwsProperties implements Serializable { public static final boolean GLUE_CATALOG_FORCE_REGISTER_TABLE_DEFAULT = false; + /** Configure the Glue Catalog S3 FileIO Region to allow cross region s3 access */ + public static final String GLUE_CATALOG_FILE_IO_REGION = "glue.catalog-file-io-region"; + /** * Number of threads to use for uploading parts to S3 (shared pool across all output streams), * default to {@link Runtime#availableProcessors()} @@ -694,6 +697,7 @@ public class AwsProperties implements Serializable { private boolean glueCatalogSkipNameValidation; private boolean glueLakeFormationEnabled; private boolean glueCatalogForceRegisterTable; + private String glueCatalogFileIORegion; private String dynamoDbTableName; private String dynamoDbEndpoint; @@ -755,6 +759,7 @@ public AwsProperties() { this.glueCatalogSkipNameValidation = GLUE_CATALOG_SKIP_NAME_VALIDATION_DEFAULT; this.glueLakeFormationEnabled = GLUE_LAKEFORMATION_ENABLED_DEFAULT; this.glueCatalogForceRegisterTable = GLUE_CATALOG_FORCE_REGISTER_TABLE_DEFAULT; + this.glueCatalogFileIORegion = null; this.dynamoDbEndpoint = null; this.dynamoDbTableName = DYNAMODB_TABLE_NAME_DEFAULT; @@ -842,6 +847,8 @@ public AwsProperties(Map properties) { properties, GLUE_CATALOG_FORCE_REGISTER_TABLE, GLUE_CATALOG_FORCE_REGISTER_TABLE_DEFAULT); + this.glueCatalogFileIORegion = properties.get(GLUE_CATALOG_FILE_IO_REGION); + this.s3FileIoMultipartUploadThreads = PropertyUtil.propertyAsInt( properties, @@ -1024,6 +1031,14 @@ public void setGlueCatalogForceRegisterTable(boolean glueCatalogForceRegisterTab this.glueCatalogForceRegisterTable = glueCatalogForceRegisterTable; } + public String getGlueCatalogFileIORegion() { + return glueCatalogFileIORegion; + } + + public void setGlueCatalogFileIORegion(String glueCatalogFileIORegion) { + this.glueCatalogFileIORegion = glueCatalogFileIORegion; + } + public int s3FileIoMultipartUploadThreads() { return s3FileIoMultipartUploadThreads; } diff --git a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java index afc3d736cb76..943bc15accdd 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java +++ b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java @@ -20,6 +20,7 @@ import java.io.Closeable; import java.io.IOException; +import java.util.HashMap; import java.util.List; import java.util.Locale; import java.util.Map; @@ -34,6 +35,7 @@ import org.apache.iceberg.TableMetadata; import org.apache.iceberg.TableMetadataParser; import org.apache.iceberg.TableOperations; +import org.apache.iceberg.aws.AssumeRoleAwsClientFactory; import org.apache.iceberg.aws.AwsClientFactories; import org.apache.iceberg.aws.AwsClientFactory; import org.apache.iceberg.aws.AwsProperties; @@ -115,7 +117,8 @@ public GlueCatalog() {} @Override public void initialize(String name, Map properties) { - this.catalogProperties = ImmutableMap.copyOf(properties); + this.catalogProperties = new HashMap<>(); + catalogProperties.putAll(properties); AwsClientFactory awsClientFactory; FileIO catalogFileIO; if (PropertyUtil.propertyAsBoolean( @@ -455,6 +458,16 @@ public org.apache.iceberg.Table registerTable( return super.registerTable(identifier, metadataFileLocation); } + String factoryImpl = + PropertyUtil.propertyAsString(catalogProperties, AwsProperties.CLIENT_FACTORY, null); + if (factoryImpl != null && factoryImpl.equals(AssumeRoleAwsClientFactory.class.getName())) { + // overwrite client assume_role_region for file IO to make cross region call + String catalogFileIORegion = awsProperties.getGlueCatalogFileIORegion(); + if (catalogFileIORegion != null) { + catalogProperties.put(AwsProperties.CLIENT_ASSUME_ROLE_REGION, catalogFileIORegion); + } + } + TableOperations ops = newTableOps(identifier); InputFile metadataFile = ops.io().newInputFile(metadataFileLocation); TableMetadata metadata = TableMetadataParser.read(ops.io(), metadataFile);