yuqi1129 commented on code in PR #14561:
URL: https://github.com/apache/gravitino/pull/14561#discussion_r4464396537
##########
catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueCatalogOperations.java:
##########
@@ -918,11 +1315,94 @@ private TableInput buildTableInput(
private software.amazon.awssdk.services.glue.model.Column
toGlueColumn(Column col) {
return software.amazon.awssdk.services.glue.model.Column.builder()
.name(col.name())
- .type(typeConverter.fromGravitino(col.dataType()))
+ .type(toGlueType(col.name(), col.dataType()))
.comment(col.comment())
.build();
}
+ private String toGlueType(String columnName, Type type) {
+ try {
+ return typeConverter.fromGravitino(type);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ String.format("Column %s: %s", columnName, e.getMessage()), e);
+ }
+ }
+
+ /**
+ * Resolves the storage classes of a Hive-format table. A {@code format}
sets all three classes,
+ * and any class given next to it must match. Without a format, either all
three classes are
+ * given, or none and the table is a text table, as in Hive and Athena.
+ *
+ * @return the storage format to apply, or null when all three classes are
given explicitly
+ */
+ private static StorageFormat resolveStorageFormat(
+ String format, String inputFormat, String outputFormat, String serdeLib)
{
+ if (format == null) {
+ // Built-in input and output classes must come from the same storage
format; a custom SerDe
+ // over a built-in format, such as a JSON SerDe over text files, is
allowed.
+ Preconditions.checkArgument(
Review Comment:
[Important] This checks the built-in input/output pair, but a known
incompatible SerDe still passes when `format` is omitted. For example, ORC
input + ORC output + ParquetHiveSerDe satisfies this check and the
all-three-classes check, so an inconsistent StorageDescriptor is submitted to
Glue.
I added a temporary regression case using ORC_INPUT_FORMAT_CLASS,
ORC_OUTPUT_FORMAT_CLASS and PARQUET_SERDE_CLASS; the expected rejection did not
occur. Could we also reject clearly incompatible known built-in SerDes? We can
retain the intended support for custom SerDes (for example JSON over text)
without expanding this into validation of arbitrary custom classes. Athena
documents separate ORC and Parquet SerDes:
https://docs.aws.amazon.com/athena/latest/ug/supported-serdes.html
##########
catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueCatalogOperations.java:
##########
@@ -550,70 +591,311 @@ public GlueTable alterTable(NameIdentifier ident,
TableChange... changes)
return alterIcebergTable(ident, dbName, rawGlueTable, changes);
}
- GlueTable current = GlueTable.fromGlueTable(rawGlueTable, typeConverter);
- current.initOpsContext(glueClient, catalogId, dbName);
-
- String newName = current.name();
- String newComment = current.comment();
- Map<String, String> newProps = new HashMap<>(current.properties());
+ return alterHiveTable(ident, dbName, rawGlueTable, changes);
+ }
- // Separate data columns from partition columns
- int partCount = current.partitioning().length;
- List<Column> allCols = new ArrayList<>(Arrays.asList(current.columns()));
- List<Column> dataCols = new ArrayList<>(allCols.subList(0, allCols.size()
- partCount));
- List<Column> partCols =
- new ArrayList<>(allCols.subList(allCols.size() - partCount,
allCols.size()));
+ /**
+ * Alters a Hive-format table. Glue UpdateTable replaces the whole table
definition, so the new
+ * definition starts from the table as Glue holds it and only the requested
parts change; fields
+ * Gravitino does not model, and column types that are not changed, are
written back as read. The
+ * version read is sent along, so Glue refuses the update if the table
changed in between.
+ */
+ private GlueTable alterHiveTable(
+ NameIdentifier ident, String dbName, Table raw, TableChange... changes) {
+ StorageDescriptor rawSd = raw.storageDescriptor();
+ HiveColumns cols =
+ new HiveColumns(
+ rawSd != null && rawSd.hasColumns() ? rawSd.columns() :
Collections.emptyList(),
+ raw.hasPartitionKeys() ? raw.partitionKeys() :
Collections.emptyList(),
+ rawSd != null && rawSd.hasBucketColumns()
+ ? rawSd.bucketColumns()
+ : Collections.emptyList(),
+ rawSd != null && rawSd.hasSortColumns() ? rawSd.sortColumns() :
Collections.emptyList(),
+ rawSd != null && rawSd.skewedInfo() != null &&
rawSd.skewedInfo().hasSkewedColumnNames()
+ ? rawSd.skewedInfo().skewedColumnNames()
+ : Collections.emptyList());
+ String description = raw.description();
+ Map<String, String> params =
+ raw.hasParameters() ? new HashMap<>(raw.parameters()) : new
HashMap<>();
+ SerDeInfo rawSerde = rawSd != null ? rawSd.serdeInfo() : null;
+ String location = rawSd != null ? rawSd.location() : null;
+ String serdeName = rawSerde != null ? rawSerde.name() : null;
+ Map<String, String> serdeParams =
+ rawSerde != null && rawSerde.hasParameters()
+ ? new HashMap<>(rawSerde.parameters())
+ : new HashMap<>();
+ boolean storageChanged = false;
+ boolean changed = false;
for (TableChange change : changes) {
if (change instanceof TableChange.RenameTable) {
throw new UnsupportedOperationException("Glue does not support table
rename");
} else if (change instanceof TableChange.UpdateComment) {
- newComment = ((TableChange.UpdateComment) change).getNewComment();
+ description = ((TableChange.UpdateComment) change).getNewComment();
+ changed = true;
} else if (change instanceof TableChange.SetProperty) {
+ // Storage properties map to storage descriptor fields, the rest to
table parameters.
TableChange.SetProperty sp = (TableChange.SetProperty) change;
- newProps.put(sp.getProperty(), sp.getValue());
+ String property = sp.getProperty();
+ if (GlueConstants.LOCATION.equals(property)) {
+ location = sp.getValue();
+ storageChanged = true;
+ } else if (GlueConstants.SERDE_NAME.equals(property)) {
+ serdeName = sp.getValue();
+ storageChanged = true;
+ } else if (property.startsWith(GlueConstants.SERDE_PARAMETER_PREFIX)) {
+ serdeParams.put(
+
property.substring(GlueConstants.SERDE_PARAMETER_PREFIX.length()),
sp.getValue());
+ storageChanged = true;
+ } else {
+ checkTableParameterKey(property);
+ params.put(property, sp.getValue());
+ }
+ changed = true;
} else if (change instanceof TableChange.RemoveProperty) {
- newProps.remove(((TableChange.RemoveProperty) change).getProperty());
+ String property = ((TableChange.RemoveProperty) change).getProperty();
+ // A table parameter of the same name is shown when the storage
descriptor has no value, so
+ // it is removed as well.
+ if (isStorageProperty(property)) {
+ changed |= params.remove(property) != null;
+ }
+ if (GlueConstants.LOCATION.equals(property)) {
+ storageChanged |= location != null;
+ location = null;
+ } else if (GlueConstants.SERDE_NAME.equals(property)) {
+ storageChanged |= serdeName != null;
+ serdeName = null;
+ } else if (property.startsWith(GlueConstants.SERDE_PARAMETER_PREFIX)) {
+ storageChanged |=
+
serdeParams.remove(property.substring(GlueConstants.SERDE_PARAMETER_PREFIX.length()))
+ != null;
+ } else {
+ checkTableParameterKey(property);
+ changed |= params.remove(property) != null;
+ }
+ changed |= storageChanged;
} else if (change instanceof TableChange.ColumnChange) {
- applyColumnChange(dataCols, partCols, (TableChange.ColumnChange)
change);
+ changed |= cols.apply((TableChange.ColumnChange) change);
} else {
throw new IllegalArgumentException(
"Unsupported table change: " + change.getClass().getSimpleName());
}
}
- List<Column> newAllCols = new ArrayList<>(dataCols);
- newAllCols.addAll(partCols);
- Column[] newColumns = newAllCols.toArray(new Column[0]);
+ if (!changed) {
+ return loadTable(ident);
+ }
+
+ StorageDescriptor sd = rawSd;
+ if (rawSd != null || !cols.dataCols.isEmpty() || storageChanged) {
+ StorageDescriptor.Builder sdBuilder =
+ rawSd != null ? rawSd.toBuilder() : StorageDescriptor.builder();
+
sdBuilder.columns(cols.dataCols).bucketColumns(cols.bucketCols).sortColumns(cols.sortCols);
+ if (rawSd != null && rawSd.skewedInfo() != null) {
+ sdBuilder.skewedInfo(
+
rawSd.skewedInfo().toBuilder().skewedColumnNames(cols.skewCols).build());
+ }
+ if (storageChanged) {
+ SerDeInfo.Builder serdeBuilder =
+ rawSerde != null ? rawSerde.toBuilder() : SerDeInfo.builder();
+ sdBuilder
+ .location(location)
+
.serdeInfo(serdeBuilder.name(serdeName).parameters(serdeParams).build());
+ }
+ sd = sdBuilder.build();
+ }
TableInput input =
- buildTableInput(
- dbName,
- newName,
- newComment,
- newColumns,
- newProps,
- current.partitioning(),
- current.distribution(),
- current.sortOrder(),
- false);
+ TableInput.builder()
+ .name(raw.name())
+ .description(description)
+ .owner(raw.owner())
+ .lastAccessTime(raw.lastAccessTime())
+ .lastAnalyzedTime(raw.lastAnalyzedTime())
+ .retention(raw.retention())
+ .storageDescriptor(sd)
+ .partitionKeys(cols.partCols)
+ .viewOriginalText(raw.viewOriginalText())
+ .viewExpandedText(raw.viewExpandedText())
+ .tableType(raw.tableType())
+ .parameters(params)
+ .targetTable(raw.targetTable())
+ .build();
- executeUpdateTable(ident,
UpdateTableRequest.builder().databaseName(dbName).tableInput(input));
+ executeUpdateTable(
+ ident,
+ UpdateTableRequest.builder()
+ .databaseName(dbName)
+ .tableInput(input)
+ .versionId(raw.versionId()));
LOG.info("Altered Glue table {}.{}", dbName, ident.name());
+ return loadTable(ident);
+ }
- GlueTable altered =
- GlueTable.builder()
- .withName(newName)
- .withComment(newComment)
- .withColumns(newColumns)
- .withProperties(newProps)
- .withPartitioning(current.partitioning())
- .withDistribution(current.distribution())
- .withSortOrders(current.sortOrder())
- .withAuditInfo(current.auditInfo())
- .build();
- altered.initOpsContext(glueClient, catalogId, dbName);
- return altered;
+ private static boolean isStorageProperty(String property) {
+ return GlueConstants.LOCATION.equals(property)
+ || GlueConstants.SERDE_NAME.equals(property)
+ || property.startsWith(GlueConstants.SERDE_PARAMETER_PREFIX);
+ }
+
+ /**
+ * The storage format and classes are set at create only, and so is the
table format: changing it
+ * would make a Hive-format table read as another format, or hide it behind
the format filter.
+ */
+ private static void checkTableParameterKey(String property) {
+ Preconditions.checkArgument(
+ !SD_TABLE_PROPERTY_KEYS.contains(property)
+ && !GlueConstants.TABLE_FORMAT.equals(property)
Review Comment:
[Nit] Making `table-format` create-only for Hive tables seems reasonable,
but could we sync the property contract and docs with this restriction?
GlueTablePropertiesMetadata still declares it with stringOptionalPropertyEntry,
and the table in docs/aws-glue-catalog.md lists Immutable=No. A caller
following those declarations is told that the property can be changed, while
this branch rejects both set and remove.
If Iceberg has a different policy, documenting the Hive-specific restriction
would be enough; this does not need to become a global immutability change.
##########
catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueCatalogOperations.java:
##########
@@ -550,70 +591,311 @@ public GlueTable alterTable(NameIdentifier ident,
TableChange... changes)
return alterIcebergTable(ident, dbName, rawGlueTable, changes);
}
- GlueTable current = GlueTable.fromGlueTable(rawGlueTable, typeConverter);
- current.initOpsContext(glueClient, catalogId, dbName);
-
- String newName = current.name();
- String newComment = current.comment();
- Map<String, String> newProps = new HashMap<>(current.properties());
+ return alterHiveTable(ident, dbName, rawGlueTable, changes);
+ }
- // Separate data columns from partition columns
- int partCount = current.partitioning().length;
- List<Column> allCols = new ArrayList<>(Arrays.asList(current.columns()));
- List<Column> dataCols = new ArrayList<>(allCols.subList(0, allCols.size()
- partCount));
- List<Column> partCols =
- new ArrayList<>(allCols.subList(allCols.size() - partCount,
allCols.size()));
+ /**
+ * Alters a Hive-format table. Glue UpdateTable replaces the whole table
definition, so the new
+ * definition starts from the table as Glue holds it and only the requested
parts change; fields
+ * Gravitino does not model, and column types that are not changed, are
written back as read. The
+ * version read is sent along, so Glue refuses the update if the table
changed in between.
+ */
+ private GlueTable alterHiveTable(
+ NameIdentifier ident, String dbName, Table raw, TableChange... changes) {
+ StorageDescriptor rawSd = raw.storageDescriptor();
+ HiveColumns cols =
+ new HiveColumns(
+ rawSd != null && rawSd.hasColumns() ? rawSd.columns() :
Collections.emptyList(),
+ raw.hasPartitionKeys() ? raw.partitionKeys() :
Collections.emptyList(),
+ rawSd != null && rawSd.hasBucketColumns()
+ ? rawSd.bucketColumns()
+ : Collections.emptyList(),
+ rawSd != null && rawSd.hasSortColumns() ? rawSd.sortColumns() :
Collections.emptyList(),
+ rawSd != null && rawSd.skewedInfo() != null &&
rawSd.skewedInfo().hasSkewedColumnNames()
+ ? rawSd.skewedInfo().skewedColumnNames()
+ : Collections.emptyList());
+ String description = raw.description();
+ Map<String, String> params =
+ raw.hasParameters() ? new HashMap<>(raw.parameters()) : new
HashMap<>();
+ SerDeInfo rawSerde = rawSd != null ? rawSd.serdeInfo() : null;
+ String location = rawSd != null ? rawSd.location() : null;
+ String serdeName = rawSerde != null ? rawSerde.name() : null;
+ Map<String, String> serdeParams =
+ rawSerde != null && rawSerde.hasParameters()
+ ? new HashMap<>(rawSerde.parameters())
+ : new HashMap<>();
+ boolean storageChanged = false;
+ boolean changed = false;
for (TableChange change : changes) {
if (change instanceof TableChange.RenameTable) {
throw new UnsupportedOperationException("Glue does not support table
rename");
} else if (change instanceof TableChange.UpdateComment) {
- newComment = ((TableChange.UpdateComment) change).getNewComment();
+ description = ((TableChange.UpdateComment) change).getNewComment();
+ changed = true;
} else if (change instanceof TableChange.SetProperty) {
+ // Storage properties map to storage descriptor fields, the rest to
table parameters.
TableChange.SetProperty sp = (TableChange.SetProperty) change;
- newProps.put(sp.getProperty(), sp.getValue());
+ String property = sp.getProperty();
+ if (GlueConstants.LOCATION.equals(property)) {
+ location = sp.getValue();
+ storageChanged = true;
+ } else if (GlueConstants.SERDE_NAME.equals(property)) {
+ serdeName = sp.getValue();
+ storageChanged = true;
+ } else if (property.startsWith(GlueConstants.SERDE_PARAMETER_PREFIX)) {
+ serdeParams.put(
+
property.substring(GlueConstants.SERDE_PARAMETER_PREFIX.length()),
sp.getValue());
+ storageChanged = true;
+ } else {
+ checkTableParameterKey(property);
+ params.put(property, sp.getValue());
+ }
+ changed = true;
} else if (change instanceof TableChange.RemoveProperty) {
- newProps.remove(((TableChange.RemoveProperty) change).getProperty());
+ String property = ((TableChange.RemoveProperty) change).getProperty();
+ // A table parameter of the same name is shown when the storage
descriptor has no value, so
+ // it is removed as well.
+ if (isStorageProperty(property)) {
+ changed |= params.remove(property) != null;
+ }
+ if (GlueConstants.LOCATION.equals(property)) {
+ storageChanged |= location != null;
+ location = null;
+ } else if (GlueConstants.SERDE_NAME.equals(property)) {
+ storageChanged |= serdeName != null;
+ serdeName = null;
+ } else if (property.startsWith(GlueConstants.SERDE_PARAMETER_PREFIX)) {
+ storageChanged |=
+
serdeParams.remove(property.substring(GlueConstants.SERDE_PARAMETER_PREFIX.length()))
+ != null;
+ } else {
+ checkTableParameterKey(property);
+ changed |= params.remove(property) != null;
+ }
+ changed |= storageChanged;
} else if (change instanceof TableChange.ColumnChange) {
- applyColumnChange(dataCols, partCols, (TableChange.ColumnChange)
change);
+ changed |= cols.apply((TableChange.ColumnChange) change);
} else {
throw new IllegalArgumentException(
"Unsupported table change: " + change.getClass().getSimpleName());
}
}
- List<Column> newAllCols = new ArrayList<>(dataCols);
- newAllCols.addAll(partCols);
- Column[] newColumns = newAllCols.toArray(new Column[0]);
+ if (!changed) {
+ return loadTable(ident);
+ }
+
+ StorageDescriptor sd = rawSd;
+ if (rawSd != null || !cols.dataCols.isEmpty() || storageChanged) {
+ StorageDescriptor.Builder sdBuilder =
+ rawSd != null ? rawSd.toBuilder() : StorageDescriptor.builder();
+
sdBuilder.columns(cols.dataCols).bucketColumns(cols.bucketCols).sortColumns(cols.sortCols);
+ if (rawSd != null && rawSd.skewedInfo() != null) {
+ sdBuilder.skewedInfo(
+
rawSd.skewedInfo().toBuilder().skewedColumnNames(cols.skewCols).build());
+ }
+ if (storageChanged) {
+ SerDeInfo.Builder serdeBuilder =
+ rawSerde != null ? rawSerde.toBuilder() : SerDeInfo.builder();
+ sdBuilder
+ .location(location)
+
.serdeInfo(serdeBuilder.name(serdeName).parameters(serdeParams).build());
+ }
+ sd = sdBuilder.build();
+ }
TableInput input =
- buildTableInput(
- dbName,
- newName,
- newComment,
- newColumns,
- newProps,
- current.partitioning(),
- current.distribution(),
- current.sortOrder(),
- false);
+ TableInput.builder()
+ .name(raw.name())
+ .description(description)
+ .owner(raw.owner())
+ .lastAccessTime(raw.lastAccessTime())
+ .lastAnalyzedTime(raw.lastAnalyzedTime())
+ .retention(raw.retention())
+ .storageDescriptor(sd)
+ .partitionKeys(cols.partCols)
+ .viewOriginalText(raw.viewOriginalText())
+ .viewExpandedText(raw.viewExpandedText())
+ .tableType(raw.tableType())
+ .parameters(params)
+ .targetTable(raw.targetTable())
+ .build();
- executeUpdateTable(ident,
UpdateTableRequest.builder().databaseName(dbName).tableInput(input));
+ executeUpdateTable(
+ ident,
+ UpdateTableRequest.builder()
+ .databaseName(dbName)
+ .tableInput(input)
+ .versionId(raw.versionId()));
LOG.info("Altered Glue table {}.{}", dbName, ident.name());
+ return loadTable(ident);
Review Comment:
[Important] Could we make this read-back best effort, as `createTable` now
does? The same issue exists in `alterSchema` at line 326.
At this point the update has already succeeded. If the following
GetTable/GetDatabase is throttled or times out, we report the entire alter as
failed even though Glue has committed it. The core dispatcher then skips its
remaining result synchronization/success path. Retrying a non-idempotent
change, such as a column rename, can fail again because the source column has
already been renamed.
I reproduced both paths by letting the initial Get and Update succeed, then
throwing a ThrottlingException from the second Get: both alter methods throw
ServiceUnavailableException after the successful write. Could we return an
object built from the submitted TableInput/DatabaseInput, or fall back to it
with a warning when the read-back fails, and cover both paths with regression
tests?
--
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]
