Skip to content

AWS: support force register table in GlueCatalog #6742

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
theoxu31 wants to merge 9 commits into from
Closed

AWS: support force register table in GlueCatalog #6742

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
73ecc5d
support registerTable in GlueCatalog
theoxu31 Feb 4, 2023
4ac803c
remove unnessary lines
theoxu31 Feb 4, 2023
3509f3a
resolve comments in pr
theoxu31 Feb 7, 2023
3c25e64
add feature flag to registerTable & specify versionId to ensure atomi…
theoxu31 Feb 7, 2023
8e5349b
resolve comments to optimize the flag behavior in registerTable
theoxu31 Feb 7, 2023
043b8a6
support glue endpoint for assumeRoleAwsClientFactory
theoxu31 Apr 7, 2023
4bdb165
support pass in table paramters to storagedescriptor
theoxu31 Apr 11, 2023
b38a928
allow s3 file io cross region call for glue catalog register table
theoxu31 Apr 28, 2023
dd9fdc4
Merge branch 'master' into glue-register-table
theoxu31 May 5, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 56 additions & 5 deletions aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand All @@ -578,10 +603,36 @@ 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);

// 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
Expand Down
40 changes: 40 additions & 0 deletions aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<Configuration> {
Expand Down Expand Up @@ -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) {

@jackye1995 jackye1995 Feb 4, 2023 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we might want to have a feature flag like glue.register-table.create-new-metadata (for the lack of a better name) in AwsProperties to distinguish 2 behaviors, to either create a new metadata file or not. If the flag is true (by default), it can call the base class method directly.

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<String, String> 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)

@jackye1995 jackye1995 Feb 4, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think here we need to still get the metadata and use the IcebergToGlueConverter to get the merged schema for display.

.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()

@jackye1995 jackye1995 Feb 4, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Similarly, I think we should at least make this a feature flag like glue.register-table.replace-if-exists.

We could further argue if we should make this an API feature or not, but that is subject to debate. Any thoughts?

.databaseName(IcebergToGlueConverter.getDatabaseName(identifier,
awsProperties.glueCatalogSkipNameValidation()))
.tableInput(tableInput)
.build());

@jackye1995 jackye1995 Feb 6, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this should first get the last table version to avoid commit conflict

@theoxu31 theoxu31 Feb 7, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think glue catalog already handle this internally?

If we call glue.getTable to get the table version first then call glue.updateTable with nextVersionId it will cause concurrentModificationException.

@jackye1995 jackye1995 Feb 7, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes this is what I mean. It needs to explicitly pass in the version number of the current version to ensure atomic update.

@theoxu31 theoxu31 Feb 7, 2023

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ack, will update

} catch (EntityNotFoundException e) {
Comment on lines +471 to +514

@ChristinaTech ChristinaTech May 7, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So this logic is a combination of a fork of the logic in BaseMetastoreCatalog.registerTable and GlueTableOperations.persistGlueTable. As GlueTableOperations also has access to AwsProperties, I would recommend refactoring the logic in GlueTableOperations so that persistGlueTable can conditionally fall back to its update mode as that improves code reuse.

If the concern is the creation of an extra metadata file, it looks like GlueTableOperations already checks whether it needs to during the writeNewMetadataIfRequired function and considering tableMetadata is always set for registerTable, it will never choose to write a new metadata file anyways.

throw new NoSuchNamespaceException(e, "Namespace %s is not found in Glue", identifier.namespace());
}

return loadTable(identifier);
}

@Override
public void createNamespace(Namespace namespace, Map<String, String> metadata) {
try {
Expand Down
2 changes: 1 addition & 1 deletion aws/src/main/java/org/apache/iceberg/aws/glue/GlueTableOperations.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
26 changes: 26 additions & 0 deletions aws/src/test/java/org/apache/iceberg/aws/glue/TestGlueCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand Down