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 cf9678e54d83..d5eeba2db46b 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/AssumeRoleAwsClientFactory.java +++ b/aws/src/main/java/org/apache/iceberg/aws/AssumeRoleAwsClientFactory.java @@ -76,6 +76,7 @@ public GlueClient glue() { return GlueClient.builder() .applyMutation(this::applyAssumeRoleConfigurations) .applyMutation(httpClientProperties::applyHttpClientConfigurations) + .applyMutation(awsProperties::applyGlueCatalogIdConfigurations) .applyMutation(awsClientProperties::applyRetryConfigurations) .build(); } diff --git a/aws/src/main/java/org/apache/iceberg/aws/AwsClientFactories.java b/aws/src/main/java/org/apache/iceberg/aws/AwsClientFactories.java index 9697ed6c7cae..3f7aed7554b9 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/AwsClientFactories.java +++ b/aws/src/main/java/org/apache/iceberg/aws/AwsClientFactories.java @@ -147,6 +147,7 @@ public GlueClient glue() { .applyMutation(awsClientProperties::applyClientRegionConfiguration) .applyMutation(httpClientProperties::applyHttpClientConfigurations) .applyMutation(awsProperties::applyGlueEndpointConfigurations) + .applyMutation(awsProperties::applyGlueCatalogIdConfigurations) .applyMutation(awsClientProperties::applyClientCredentialConfigurations) .applyMutation(awsClientProperties::applyRetryConfigurations) .build(); 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 abe6f1e9ba0a..8fee7fbdf4b1 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java +++ b/aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java @@ -427,6 +427,22 @@ public void applyGlueEndpointConfigurations(T buil configureEndpoint(builder, glueEndpoint); } + /** + * Add the interceptor to assign a catalog id to a glue client. + * + *

Sample usage: + * + * 

+   *     GlueClient.builder().applyMutation(awsProperties::applyGlueCatalogIdConfigurations)
+   *
+ */ + public void applyGlueCatalogIdConfigurations(T builder) { + if (!Strings.isNullOrEmpty(glueCatalogId)) { + builder.overrideConfiguration( + c -> c.addExecutionInterceptor(new GlueCatalogIdInterceptor(glueCatalogId))); + } + } + /** * Override the endpoint for a dynamoDb client. * diff --git a/aws/src/main/java/org/apache/iceberg/aws/GlueCatalogIdInterceptor.java b/aws/src/main/java/org/apache/iceberg/aws/GlueCatalogIdInterceptor.java new file mode 100644 index 000000000000..f678aa6b6167 --- /dev/null +++ b/aws/src/main/java/org/apache/iceberg/aws/GlueCatalogIdInterceptor.java @@ -0,0 +1,80 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.aws; + +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.base.Strings; +import software.amazon.awssdk.core.SdkRequest; +import software.amazon.awssdk.core.interceptor.Context.ModifyRequest; +import software.amazon.awssdk.core.interceptor.ExecutionAttributes; +import software.amazon.awssdk.core.interceptor.ExecutionInterceptor; +import software.amazon.awssdk.services.glue.model.CreateDatabaseRequest; +import software.amazon.awssdk.services.glue.model.CreateTableRequest; +import software.amazon.awssdk.services.glue.model.DeleteDatabaseRequest; +import software.amazon.awssdk.services.glue.model.DeleteTableRequest; +import software.amazon.awssdk.services.glue.model.GetDatabaseRequest; +import software.amazon.awssdk.services.glue.model.GetDatabasesRequest; +import software.amazon.awssdk.services.glue.model.GetTableRequest; +import software.amazon.awssdk.services.glue.model.GetTablesRequest; +import software.amazon.awssdk.services.glue.model.GlueRequest; +import software.amazon.awssdk.services.glue.model.UpdateDatabaseRequest; +import software.amazon.awssdk.services.glue.model.UpdateTableRequest; + +class GlueCatalogIdInterceptor implements ExecutionInterceptor { + + private final String catalogId; + + GlueCatalogIdInterceptor(String catalogId) { + Preconditions.checkArgument( + !Strings.isNullOrEmpty(catalogId), "Invalid catalog id: null or empty"); + this.catalogId = catalogId; + } + + @Override + public SdkRequest modifyRequest(ModifyRequest context, ExecutionAttributes executionAttributes) { + SdkRequest request = context.request(); + if (!(request instanceof GlueRequest)) { + return request; + } + + if (request instanceof GetDatabaseRequest) { + return ((GetDatabaseRequest) request).toBuilder().catalogId(catalogId).build(); + } else if (request instanceof GetDatabasesRequest) { + return ((GetDatabasesRequest) request).toBuilder().catalogId(catalogId).build(); + } else if (request instanceof CreateDatabaseRequest) { + return ((CreateDatabaseRequest) request).toBuilder().catalogId(catalogId).build(); + } else if (request instanceof DeleteDatabaseRequest) { + return ((DeleteDatabaseRequest) request).toBuilder().catalogId(catalogId).build(); + } else if (request instanceof UpdateDatabaseRequest) { + return ((UpdateDatabaseRequest) request).toBuilder().catalogId(catalogId).build(); + } else if (request instanceof GetTableRequest) { + return ((GetTableRequest) request).toBuilder().catalogId(catalogId).build(); + } else if (request instanceof GetTablesRequest) { + return ((GetTablesRequest) request).toBuilder().catalogId(catalogId).build(); + } else if (request instanceof CreateTableRequest) { + return ((CreateTableRequest) request).toBuilder().catalogId(catalogId).build(); + } else if (request instanceof UpdateTableRequest) { + return ((UpdateTableRequest) request).toBuilder().catalogId(catalogId).build(); + } else if (request instanceof DeleteTableRequest) { + return ((DeleteTableRequest) request).toBuilder().catalogId(catalogId).build(); + } else { + throw new IllegalArgumentException("Unexpected request: " + request.getClass().getName()); + } + } +} 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 adbf0a03e6b2..7b480a3ff8c5 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 @@ -289,7 +289,6 @@ protected String defaultWarehouseLocation(TableIdentifier tableIdentifier) { GetDatabaseResponse response = glue.getDatabase( GetDatabaseRequest.builder() - .catalogId(awsProperties.glueCatalogId()) .name( IcebergToGlueConverter.getDatabaseName( tableIdentifier, awsProperties.glueCatalogSkipNameValidation())) @@ -323,7 +322,6 @@ public List listTables(Namespace namespace) { GetTablesResponse response = glue.getTables( GetTablesRequest.builder() - .catalogId(awsProperties.glueCatalogId()) .databaseName( IcebergToGlueConverter.toDatabaseName( namespace, awsProperties.glueCatalogSkipNameValidation())) @@ -366,7 +364,6 @@ public boolean dropTable(TableIdentifier identifier, boolean purge) { } glue.deleteTable( DeleteTableRequest.builder() - .catalogId(awsProperties.glueCatalogId()) .databaseName( IcebergToGlueConverter.getDatabaseName( identifier, awsProperties.glueCatalogSkipNameValidation())) @@ -415,11 +412,7 @@ public void renameTable(TableIdentifier from, TableIdentifier to) { try { GetTableResponse response = glue.getTable( - GetTableRequest.builder() - .catalogId(awsProperties.glueCatalogId()) - .databaseName(fromTableDbName) - .name(fromTableName) - .build()); + GetTableRequest.builder().databaseName(fromTableDbName).name(fromTableName).build()); fromTable = response.table(); } catch (EntityNotFoundException e) { throw new NoSuchTableException( @@ -436,7 +429,6 @@ public void renameTable(TableIdentifier from, TableIdentifier to) { glue.createTable( CreateTableRequest.builder() - .catalogId(awsProperties.glueCatalogId()) .databaseName(toTableDbName) .tableInput(tableInputBuilder.name(toTableName).build()) .build()); @@ -452,11 +444,7 @@ public void renameTable(TableIdentifier from, TableIdentifier to) { to, e); glue.deleteTable( - DeleteTableRequest.builder() - .catalogId(awsProperties.glueCatalogId()) - .databaseName(toTableDbName) - .name(toTableName) - .build()); + DeleteTableRequest.builder().databaseName(toTableDbName).name(toTableName).build()); throw e; } @@ -468,7 +456,6 @@ public void createNamespace(Namespace namespace, Map metadata) { try { glue.createDatabase( CreateDatabaseRequest.builder() - .catalogId(awsProperties.glueCatalogId()) .databaseInput( IcebergToGlueConverter.toDatabaseInput( namespace, metadata, awsProperties.glueCatalogSkipNameValidation())) @@ -496,11 +483,7 @@ public List listNamespaces(Namespace namespace) throws NoSuchNamespac List results = Lists.newArrayList(); do { GetDatabasesResponse response = - glue.getDatabases( - GetDatabasesRequest.builder() - .catalogId(awsProperties.glueCatalogId()) - .nextToken(nextToken) - .build()); + glue.getDatabases(GetDatabasesRequest.builder().nextToken(nextToken).build()); nextToken = response.nextToken(); if (response.hasDatabaseList()) { results.addAll( @@ -522,12 +505,7 @@ public Map loadNamespaceMetadata(Namespace namespace) namespace, awsProperties.glueCatalogSkipNameValidation()); try { Database database = - glue.getDatabase( - GetDatabaseRequest.builder() - .catalogId(awsProperties.glueCatalogId()) - .name(databaseName) - .build()) - .database(); + glue.getDatabase(GetDatabaseRequest.builder().name(databaseName).build()).database(); Map result = Maps.newHashMap(database.parameters()); if (database.locationUri() != null) { @@ -559,7 +537,6 @@ public boolean dropNamespace(Namespace namespace) throws NamespaceNotEmptyExcept GetTablesResponse response = glue.getTables( GetTablesRequest.builder() - .catalogId(awsProperties.glueCatalogId()) .databaseName( IcebergToGlueConverter.toDatabaseName( namespace, awsProperties.glueCatalogSkipNameValidation())) @@ -578,7 +555,6 @@ public boolean dropNamespace(Namespace namespace) throws NamespaceNotEmptyExcept glue.deleteDatabase( DeleteDatabaseRequest.builder() - .catalogId(awsProperties.glueCatalogId()) .name( IcebergToGlueConverter.toDatabaseName( namespace, awsProperties.glueCatalogSkipNameValidation())) @@ -596,7 +572,6 @@ public boolean setProperties(Namespace namespace, Map properties newProperties.putAll(properties); glue.updateDatabase( UpdateDatabaseRequest.builder() - .catalogId(awsProperties.glueCatalogId()) .name( IcebergToGlueConverter.toDatabaseName( namespace, awsProperties.glueCatalogSkipNameValidation())) @@ -619,7 +594,6 @@ public boolean removeProperties(Namespace namespace, Set properties) glue.updateDatabase( UpdateDatabaseRequest.builder() - .catalogId(awsProperties.glueCatalogId()) .name( IcebergToGlueConverter.toDatabaseName( namespace, awsProperties.glueCatalogSkipNameValidation())) 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 f38ee8da8bca..f7814c49ee93 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 @@ -277,11 +277,7 @@ private Table getGlueTable() { try { GetTableResponse response = glue.getTable( - GetTableRequest.builder() - .catalogId(awsProperties.glueCatalogId()) - .databaseName(databaseName) - .name(tableName) - .build()); + GetTableRequest.builder().databaseName(databaseName).name(tableName).build()); return response.table(); } catch (EntityNotFoundException e) { return null; @@ -311,7 +307,6 @@ void persistGlueTable( UpdateTableRequest.Builder updateTableRequest = UpdateTableRequest.builder() .overrideConfiguration(c -> c.addMetricPublisher(retryDetector)) - .catalogId(awsProperties.glueCatalogId()) .databaseName(databaseName) .skipArchive(awsProperties.glueCatalogSkipArchive()) .tableInput( @@ -335,7 +330,6 @@ void persistGlueTable( glue.createTable( CreateTableRequest.builder() .overrideConfiguration(c -> c.addMetricPublisher(retryDetector)) - .catalogId(awsProperties.glueCatalogId()) .databaseName(databaseName) .tableInput( TableInput.builder() diff --git a/aws/src/main/java/org/apache/iceberg/aws/lakeformation/LakeFormationAwsClientFactory.java b/aws/src/main/java/org/apache/iceberg/aws/lakeformation/LakeFormationAwsClientFactory.java index 51c579ca1741..7a0e10f828f3 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/lakeformation/LakeFormationAwsClientFactory.java +++ b/aws/src/main/java/org/apache/iceberg/aws/lakeformation/LakeFormationAwsClientFactory.java @@ -54,7 +54,6 @@ public class LakeFormationAwsClientFactory extends AssumeRoleAwsClientFactory { private String dbName; private String tableName; - private String glueCatalogId; private String glueAccountId; public LakeFormationAwsClientFactory() {} @@ -70,7 +69,6 @@ public void initialize(Map catalogProperties) { AwsProperties.CLIENT_ASSUME_ROLE_TAGS_PREFIX); this.dbName = catalogProperties.get(AwsProperties.LAKE_FORMATION_DB_NAME); this.tableName = catalogProperties.get(AwsProperties.LAKE_FORMATION_TABLE_NAME); - this.glueCatalogId = catalogProperties.get(AwsProperties.GLUE_CATALOG_ID); this.glueAccountId = catalogProperties.get(AwsProperties.GLUE_ACCOUNT_ID); } @@ -114,13 +112,7 @@ private boolean isTableRegisteredWithLakeFormation() { tableName != null && !tableName.isEmpty(), "Table name can not be empty"); GetTableResponse response = - glue() - .getTable( - GetTableRequest.builder() - .catalogId(glueCatalogId) - .databaseName(dbName) - .name(tableName) - .build()); + glue().getTable(GetTableRequest.builder().databaseName(dbName).name(tableName).build()); return response.table().isRegisteredWithLakeFormation(); } diff --git a/aws/src/test/java/org/apache/iceberg/aws/TestGlueCatalogIdInterceptor.java b/aws/src/test/java/org/apache/iceberg/aws/TestGlueCatalogIdInterceptor.java new file mode 100644 index 000000000000..dc5e37ba8e68 --- /dev/null +++ b/aws/src/test/java/org/apache/iceberg/aws/TestGlueCatalogIdInterceptor.java @@ -0,0 +1,133 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.aws; + +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; + +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.core.SdkRequest; +import software.amazon.awssdk.core.interceptor.Context; +import software.amazon.awssdk.core.interceptor.ExecutionAttributes; +import software.amazon.awssdk.services.glue.model.CreateDatabaseRequest; +import software.amazon.awssdk.services.glue.model.CreateTableRequest; +import software.amazon.awssdk.services.glue.model.DeleteDatabaseRequest; +import software.amazon.awssdk.services.glue.model.DeleteTableRequest; +import software.amazon.awssdk.services.glue.model.GetDatabaseRequest; +import software.amazon.awssdk.services.glue.model.GetDatabasesRequest; +import software.amazon.awssdk.services.glue.model.GetTableRequest; +import software.amazon.awssdk.services.glue.model.GetTablesRequest; +import software.amazon.awssdk.services.glue.model.UpdateDatabaseRequest; +import software.amazon.awssdk.services.glue.model.UpdateTableRequest; + +class TestGlueCatalogIdInterceptor { + + private static final String CATALOG_ID = "testCatalogId"; + private static final ExecutionAttributes EMPTY_ATTRS = ExecutionAttributes.builder().build(); + + @Test + void getDatabaseRequest() { + SdkRequest result = intercept(GetDatabaseRequest.builder().name("db").build()); + assertThat(((GetDatabaseRequest) result).catalogId()).isEqualTo(CATALOG_ID); + } + + @Test + void getDatabasesRequest() { + SdkRequest result = intercept(GetDatabasesRequest.builder().build()); + assertThat(((GetDatabasesRequest) result).catalogId()).isEqualTo(CATALOG_ID); + } + + @Test + void createDatabaseRequest() { + SdkRequest result = intercept(CreateDatabaseRequest.builder().build()); + assertThat(((CreateDatabaseRequest) result).catalogId()).isEqualTo(CATALOG_ID); + } + + @Test + void deleteDatabaseRequest() { + SdkRequest result = intercept(DeleteDatabaseRequest.builder().name("db").build()); + assertThat(((DeleteDatabaseRequest) result).catalogId()).isEqualTo(CATALOG_ID); + } + + @Test + void updateDatabaseRequest() { + SdkRequest result = intercept(UpdateDatabaseRequest.builder().name("db").build()); + assertThat(((UpdateDatabaseRequest) result).catalogId()).isEqualTo(CATALOG_ID); + } + + @Test + void getTableRequest() { + SdkRequest result = intercept(GetTableRequest.builder().databaseName("db").name("t").build()); + assertThat(((GetTableRequest) result).catalogId()).isEqualTo(CATALOG_ID); + } + + @Test + void getTablesRequest() { + SdkRequest result = intercept(GetTablesRequest.builder().databaseName("db").build()); + assertThat(((GetTablesRequest) result).catalogId()).isEqualTo(CATALOG_ID); + } + + @Test + void createTableRequest() { + SdkRequest result = intercept(CreateTableRequest.builder().databaseName("db").build()); + assertThat(((CreateTableRequest) result).catalogId()).isEqualTo(CATALOG_ID); + } + + @Test + void updateTableRequest() { + SdkRequest result = intercept(UpdateTableRequest.builder().databaseName("db").build()); + assertThat(((UpdateTableRequest) result).catalogId()).isEqualTo(CATALOG_ID); + } + + @Test + void deleteTableRequest() { + SdkRequest result = + intercept(DeleteTableRequest.builder().databaseName("db").name("t").build()); + assertThat(((DeleteTableRequest) result).catalogId()).isEqualTo(CATALOG_ID); + } + + @Test + void nonGlueRequestPassedThrough() { + SdkRequest nonGlueRequest = mock(SdkRequest.class); + GlueCatalogIdInterceptor interceptor = new GlueCatalogIdInterceptor(CATALOG_ID); + Context.ModifyRequest ctx = mock(Context.ModifyRequest.class); + when(ctx.request()).thenReturn(nonGlueRequest); + assertThat(interceptor.modifyRequest(ctx, EMPTY_ATTRS)).isSameAs(nonGlueRequest); + } + + @Test + void invalidCatalogIdRejected() { + assertThatThrownBy(() -> new GlueCatalogIdInterceptor(null)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Invalid catalog id"); + + assertThatThrownBy(() -> new GlueCatalogIdInterceptor("")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Invalid catalog id"); + } + + private static SdkRequest intercept(SdkRequest request) { + GlueCatalogIdInterceptor interceptor = new GlueCatalogIdInterceptor(CATALOG_ID); + Context.ModifyRequest context = mock(Context.ModifyRequest.class); + when(context.request()).thenReturn(request); + return interceptor.modifyRequest(context, EMPTY_ATTRS); + } +} 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 82f7e84d563b..74f430e34965 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 @@ -166,34 +166,6 @@ public void testDefaultWarehouseLocationDbUriTrailingSlash() { assertThat(location).isEqualTo("s3://bucket2/db/table"); } - @Test - public void testDefaultWarehouseLocationCustomCatalogId() { - GlueCatalog catalogWithCustomCatalogId = new GlueCatalog(); - String catalogId = "myCatalogId"; - AwsProperties awsProperties = new AwsProperties(); - S3FileIOProperties s3FileIOProperties = new S3FileIOProperties(); - awsProperties.setGlueCatalogId(catalogId); - catalogWithCustomCatalogId.initialize( - CATALOG_NAME, - WAREHOUSE_PATH + "/", - awsProperties, - s3FileIOProperties, - glue, - LockManagers.defaultLockManager(), - ImmutableMap.of()); - - Mockito.doReturn( - GetDatabaseResponse.builder() - .database(Database.builder().name("db").locationUri("s3://bucket2/db").build()) - .build()) - .when(glue) - .getDatabase(Mockito.any(GetDatabaseRequest.class)); - catalogWithCustomCatalogId.defaultWarehouseLocation(TableIdentifier.of("db", "table")); - Mockito.verify(glue) - .getDatabase( - Mockito.argThat((GetDatabaseRequest req) -> req.catalogId().equals(catalogId))); - } - @Test public void testDefaultWarehouseLocationUnique() { GlueCatalog catalog = new GlueCatalog();