yuqi1129 commented on code in PR #14565: URL: https://github.com/apache/gravitino/pull/14565#discussion_r4478101132
########## catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueIcebergAwsClientFactory.java: ########## @@ -0,0 +1,89 @@ +/* + * 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.gravitino.catalog.glue; + +import java.util.HashMap; +import java.util.Map; +import org.apache.iceberg.aws.AwsClientFactories; +import org.apache.iceberg.aws.AwsClientFactory; +import org.apache.iceberg.aws.AwsProperties; +import org.apache.iceberg.util.PropertyUtil; +import software.amazon.awssdk.services.dynamodb.DynamoDbClient; +import software.amazon.awssdk.services.glue.GlueClient; +import software.amazon.awssdk.services.kms.KmsClient; +import software.amazon.awssdk.services.s3.S3AsyncClient; +import software.amazon.awssdk.services.s3.S3Client; + +/** + * AWS client factory of the Iceberg Glue catalog used for Iceberg tables. The Glue client is built + * the same way as the catalog's own Glue client, so the call timeout, attempt timeout and maximum + * attempts of the catalog also apply to Iceberg tables. The other clients come from the Iceberg + * default factory. + * + * <p>The catalog properties the Glue client needs are passed under {@link #GLUE_CONFIG_PREFIX}. + */ +public class GlueIcebergAwsClientFactory implements AwsClientFactory { + + /** Prefix of the Iceberg catalog properties that carry the Gravitino Glue catalog properties. */ + static final String GLUE_CONFIG_PREFIX = "gravitino.glue."; + + private static final long serialVersionUID = 1L; + + private HashMap<String, String> glueConfig; + + private AwsClientFactory delegate; + + /** Creates a factory; Iceberg calls {@link #initialize(Map)} before using it. */ + public GlueIcebergAwsClientFactory() {} + + @Override + public void initialize(Map<String, String> properties) { + this.glueConfig = + new HashMap<>(PropertyUtil.propertiesWithPrefix(properties, GLUE_CONFIG_PREFIX)); + Map<String, String> delegateProperties = new HashMap<>(properties); + delegateProperties.remove(AwsProperties.CLIENT_FACTORY); + this.delegate = AwsClientFactories.from(delegateProperties); + } + + @Override + public GlueClient glue() { + // The catalog's own Glue client has already checked that credentials can be resolved. + return GlueClientProvider.buildClient(glueConfig, false); Review Comment: [Nit] Please also update the duplicated descriptions in `GlueCatalogPropertiesMetadata`: the call-timeout and max-attempts entries still say `Not applied to Iceberg table I/O.` This factory now applies them to Iceberg's Glue calls, as the updated constants and docs say. It would help to distinguish those Glue calls from S3 I/O, where only the connection/socket timeout is configured here. ########## catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueCatalogOperations.java: ########## @@ -964,37 +978,182 @@ private GlueTable alterIcebergTable( return loadTable(finalIdent); } - private GlueTable alterRegisterModeIcebergTable( + /** + * Registers an existing Iceberg table from its metadata file. Iceberg reads the file before the + * Glue table is created, and takes the Glue columns and location from it. The Glue table is then + * marked as registered, so that later changes through Gravitino touch only its Glue parameters + * and never write into the files it shares with the table it was registered from. + */ + private GlueTable registerIcebergTable( + NameIdentifier ident, + String dbName, + String comment, + Column[] columns, + Map<String, String> props, + Transform[] partitions, + Distribution distribution, + SortOrder[] sortOrders) { + String metadataLocation = props.get(GlueConstants.METADATA_LOCATION); + Preconditions.checkArgument( + StringUtils.isNotBlank(metadataLocation), + "Property %s cannot be blank", + GlueConstants.METADATA_LOCATION); + Preconditions.checkArgument( + !props.containsKey(GlueConstants.LOCATION), + "A registered Iceberg table takes its location from %s; do not set %s", + GlueConstants.METADATA_LOCATION, + GlueConstants.LOCATION); + Preconditions.checkArgument( + (partitions == null || partitions.length == 0) + && (sortOrders == null || sortOrders.length == 0) + && (distribution == null || distribution.strategy() == Strategy.NONE), + "A registered Iceberg table takes its partitioning and sort order from %s; do not set" + + " them", + GlueConstants.METADATA_LOCATION); + + TableIdentifier tableId = TableIdentifier.of(dbName, ident.name()); + org.apache.iceberg.Table registered; + try { + registered = icebergGlueCatalog.registerTable(tableId, metadataLocation); + } catch (RuntimeException e) { + throw GlueExceptionConverter.toIcebergException( + e, String.format("registering table %s from %s", ident.name(), metadataLocation)); + } + + // Iceberg writes only its own Glue parameters during registration, so the marker takes a second + // update. A process stopped between the two leaves an unmarked table, which has to be marked by + // hand. + try { + checkRegisteredColumns(columns, registered.schema(), metadataLocation); + Table raw = getRawTable(dbName, ident.name()); + Map<String, String> params = new HashMap<>(raw.parameters()); + props.forEach(params::putIfAbsent); + params.put(GlueConstants.REGISTERED_ICEBERG_TABLE_PARAM, "true"); + TableInput input = + tableInputFrom(raw) + .description(comment != null ? comment : raw.description()) + .parameters(params) + .build(); + executeUpdateTable( + ident, + UpdateTableRequest.builder() + .databaseName(dbName) + .tableInput(input) + .versionId(raw.versionId())); + } catch (RuntimeException e) { + // Without the marker the table would be altered as its own, so do not leave it behind. + // Dropping the Glue entry keeps the metadata and data files. + try { + icebergGlueCatalog.dropTable(tableId, false); + } catch (RuntimeException dropFailure) { + e.addSuppressed(dropFailure); + } + throw e; + } + LOG.info("Registered Iceberg table {}.{} from {}", dbName, ident.name(), metadataLocation); + return loadTable(ident); + } + + /** Request columns of a registration are optional, and must match the metadata file if given. */ + private static void checkRegisteredColumns( + Column[] columns, org.apache.iceberg.Schema schema, String metadataLocation) { + if (columns == null || columns.length == 0) { + return; + } + List<String> requested = + Arrays.stream(columns) + .map(c -> c.name().toLowerCase(Locale.ROOT)) + .collect(Collectors.toList()); + List<String> stored = + schema.columns().stream() + .map(f -> f.name().toLowerCase(Locale.ROOT)) + .collect(Collectors.toList()); + Preconditions.checkArgument( + requested.equals(stored), + "Columns %s differ from the columns %s in %s; omit the columns to take them from the" + + " metadata file", + requested, + stored, + metadataLocation); + } + + /** + * Alters an Iceberg table registered through Gravitino. Only its Glue parameters change; the + * metadata file and the data files belong to the table it was registered from. + */ + private GlueTable alterRegisteredIcebergTable( NameIdentifier ident, String dbName, Table rawGlueTable, TableChange... changes) { Map<String, String> newParams = new HashMap<>(rawGlueTable.parameters()); for (TableChange change : changes) { if (change instanceof TableChange.SetProperty) { TableChange.SetProperty sp = (TableChange.SetProperty) change; + checkIcebergTableParameterKey(sp.getProperty()); newParams.put(sp.getProperty(), sp.getValue()); Review Comment: [Important] Property changes on registered tables can succeed without becoming visible. If the source metadata contains `write.format.default=parquet`, setting it to `orc` writes `orc` to Glue, but `GlueIcebergTableHelper.loadTable()` overlays the source metadata properties and returns `parquet` again. Removing that property also succeeds, then the load adds it back from metadata. I reproduced both through `ops.alterTable()`, with the mocked Glue read returning the parameters from the successful update. The existing tests use a new key (`k1`), so they miss this overlap. Please define the registered-table override/removal semantics: either honor Glue overrides and persist a way to hide removed inherited keys, or reject edits to inherited properties before writing. The fix should keep the source metadata unchanged. Please cover the returned table and a subsequent load for both set and remove. ########## docs/aws-glue-catalog.md: ########## @@ -244,59 +244,80 @@ The `warehouse` catalog property is required, but it is used only as a fallback ### Register an Existing Iceberg Table -To register an Iceberg table that already exists in S3, set `metadata_location` to the path of the existing `metadata.json` file during `createTable()`. In this mode, Gravitino registers the table in Glue without creating new metadata. +To register an Iceberg table that already exists in S3, set `metadata_location` to the path of the existing `metadata.json` file during `createTable()`. Gravitino reads the metadata file before it creates the Glue table, and refuses the request if the file does not exist or is not Iceberg table metadata. No new metadata file is written. + +The columns, location, partitioning and sort order of a registered table come from the metadata file. Columns in the request are optional; if given, their names must match the metadata file. Do not set `location`, partitioning or sort order. + +A registered table shares its metadata and data files with the table it was registered from, so Gravitino marks it with the Glue parameter `gravitino.iceberg.registered=true` and changes only its Glue parameters afterwards: property changes are stored in Glue, and column, partition and other schema changes are refused. A registered table cannot be purged. + +A table registered by an earlier Gravitino version, or by a registration that stopped before the parameter was written, has no marker. Gravitino treats it as the owner of its metadata, so an alter through Gravitino writes new metadata files and forks the history it shares with the source table. Gravitino cannot detect such tables, and the parameter cannot be set through Gravitino; to protect one, add the parameter to the table directly in AWS Glue. ### Iceberg Column Types Iceberg tables use the Iceberg type system, which differs from Hive types. The following table lists the Gravitino types supported for Iceberg tables and how they map to Iceberg types: -| Gravitino Data Type | Iceberg Data Type | Notes | -|--------------------------|-----------------------|-----------------------------------------------------| -| `boolean` | `boolean` | | -| `byte` | `int` | Widened to 32-bit integer | -| `short` | `int` | Widened to 32-bit integer | -| `integer` | `int` | | -| `long` | `long` | | -| `float` | `float` | | -| `double` | `double` | | -| `decimal(p, s)` | `decimal(p, s)` | | -| `string` | `string` | | -| `varchar` | `string` | Iceberg has no variable-length char types | -| `char` | `string` | Iceberg has no variable-length char types | -| `date` | `date` | | -| `time(6)` | `time` | Only microsecond precision (6) is supported | -| `timestamp(6)` | `timestamp` | Only microsecond precision (6) is supported | -| `timestamptz(6)` | `timestamptz` | Only microsecond precision (6) is supported | -| `binary` | `binary` | | -| `fixed(n)` | `binary` | Mapped to variable-length binary | -| `uuid` | `uuid` | | -| `list` | `list` | | -| `map` | `map` | | -| `struct` | `struct` | | +| Gravitino Data Type | Iceberg Data Type | Notes | +|-----------------------------|-----------------------------|---------------------------------------------| +| `boolean` | `boolean` | | +| `integer` | `int` | | +| `long` | `long` | | +| `float` | `float` | | +| `double` | `double` | | +| `decimal(p, s)` | `decimal(p, s)` | | +| `string` | `string` | | +| `varchar` | `string` | Iceberg has no variable-length char types | +| `char` | `string` | Iceberg has no variable-length char types | +| `date` | `date` | | +| `time(6)` | `time` | Only microsecond precision (6) is supported | +| `timestamp(6)` | `timestamp` | | +| `timestamptz(6)` | `timestamptz` | | +| `timestamp(9)` | `timestamp_ns` | Needs format version 3 | +| `timestamptz(9)` | `timestamptz_ns` | Needs format version 3 | +| `binary` | `binary` | | +| `fixed(n)` | `fixed(n)` | | +| `uuid` | `uuid` | | +| `variant` | `variant` | Needs format version 3 | +| `geometry(crs)` | `geometry(crs)` | Needs format version 3 | +| `geography(crs, algorithm)` | `geography(crs, algorithm)` | Needs format version 3 | +| `null` | `unknown` | Needs format version 3 | +| `list` | `list` | | +| `map` | `map` | | +| `struct` | `struct` | | + +:::caution +`byte`, `short` and unsigned types are not supported for Iceberg tables; use `integer` or `long`. +::: ### Iceberg Table Alter Operations For Iceberg tables, the following alter operations are supported: -| Operation | Gravitino Table Update Request | -|---------------------------------|-----------------------------------| -| Add column | `Add a column` | -| Delete column | `Delete a column` | -| Rename column | `Rename a column` | -| Update column type | `Update the type of a column` | -| Update column comment | `Update the column comment` | -| Update column nullability | `Update column nullability` | -| Set table property | `Set a table property` | -| Remove table property | `Remove a table property` | +| Operation | Gravitino Table Update Request | +|---------------------------|-----------------------------------| +| Add column | `Add a column` | +| Delete column | `Delete a column` | +| Rename column | `Rename a column` | +| Update column type | `Update the type of a column` | +| Update column position | `Update the position of a column` | +| Update column comment | `Update the column comment` | +| Update column nullability | `Update column nullability` | +| Set table property | `Set a table property` | +| Remove table property | `Remove a table property` | -:::caution -Schema changes and property changes are committed in two separate Iceberg transactions. If the schema commit succeeds but the property commit fails, the table is left in a partially altered state. -::: +All changes of one request are committed in one Iceberg transaction, so a request that fails changes nothing. Review Comment: [Nit] The transaction guarantees atomic schema/property changes, but not that every failed request changed nothing. For example, a registered-table property update can succeed in Glue and then fail in the following `loadTable()`. Iceberg can also report `CommitStateUnknownException`, which the converter correctly tells callers to check before retrying. Could this describe the atomic commit guarantee and retain those exceptions? Registered-table property updates do not use an Iceberg transaction either. ########## docs/aws-glue-catalog.md: ########## @@ -244,59 +244,80 @@ The `warehouse` catalog property is required, but it is used only as a fallback ### Register an Existing Iceberg Table -To register an Iceberg table that already exists in S3, set `metadata_location` to the path of the existing `metadata.json` file during `createTable()`. In this mode, Gravitino registers the table in Glue without creating new metadata. +To register an Iceberg table that already exists in S3, set `metadata_location` to the path of the existing `metadata.json` file during `createTable()`. Gravitino reads the metadata file before it creates the Glue table, and refuses the request if the file does not exist or is not Iceberg table metadata. No new metadata file is written. + +The columns, location, partitioning and sort order of a registered table come from the metadata file. Columns in the request are optional; if given, their names must match the metadata file. Do not set `location`, partitioning or sort order. Review Comment: [Nit] The REST request still requires the `columns` array: `TableCreateRequest.validate()` calls `WritableDataTypeValidator.validateColumns()`, which rejects null before reaching registration. Omitting the field therefore returns 400; the new IT passes an empty array. Could this explicitly say to send `"columns": []` to use the metadata schema, and align the `checkRegisteredColumns()` message that currently says to omit columns? Supporting an omitted field would also need request normalization and REST coverage. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
