This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new f6dfdb1cc6 [#11011] feat(catalog-glue): Add native Iceberg table
support via Glue SDK (#11012)
f6dfdb1cc6 is described below
commit f6dfdb1cc6b398f76427965cd9dc6c367ffeea68
Author: Yuhui <[email protected]>
AuthorDate: Tue May 12 21:17:09 2026 +0800
[#11011] feat(catalog-glue): Add native Iceberg table support via Glue SDK
(#11012)
### What changes were proposed in this pull request?
Route Iceberg `createTable`/`alterTable` through AWS Glue native
OpenTableFormat APIs,
add `GlueIcebergHelper` for type conversion and schema evolution, fix
`matchesFormatFilter`
for `default-table-format=iceberg` tables, and upgrade AWS SDK to
2.31.73.
### Why are the changes needed?
Without native Iceberg API routing, Glue skips writing `metadata.json`,
making tables
unreadable by Iceberg engines, and `alterTable` causes metadata drift.
Fix: #11011
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added `TestGlueIceberg` with 30 unit tests covering routing, type
conversion, schema evolution, and error paths.
---
.../catalog/glue/GlueCatalogOperations.java | 178 +++++-
.../gravitino/catalog/glue/GlueClientProvider.java | 2 +-
.../gravitino/catalog/glue/GlueConstants.java | 24 +
.../gravitino/catalog/glue/GlueIcebergHelper.java | 435 +++++++++++++++
.../gravitino/catalog/glue/GlueTypeConverter.java | 2 +
.../gravitino/catalog/glue/TestGlueIceberg.java | 597 +++++++++++++++++++++
gradle/libs.versions.toml | 2 +-
7 files changed, 1218 insertions(+), 22 deletions(-)
diff --git
a/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueCatalogOperations.java
b/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueCatalogOperations.java
index 70f1d5278d..b356b903ed 100644
---
a/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueCatalogOperations.java
+++
b/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueCatalogOperations.java
@@ -30,6 +30,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import java.util.function.Consumer;
import java.util.function.UnaryOperator;
@@ -79,12 +80,19 @@ import
software.amazon.awssdk.services.glue.model.GetTableRequest;
import software.amazon.awssdk.services.glue.model.GetTablesRequest;
import software.amazon.awssdk.services.glue.model.GetTablesResponse;
import software.amazon.awssdk.services.glue.model.GlueException;
+import software.amazon.awssdk.services.glue.model.IcebergInput;
+import software.amazon.awssdk.services.glue.model.IcebergTableUpdate;
+import software.amazon.awssdk.services.glue.model.MetadataOperation;
+import software.amazon.awssdk.services.glue.model.OpenTableFormatInput;
import software.amazon.awssdk.services.glue.model.Order;
import software.amazon.awssdk.services.glue.model.SerDeInfo;
import software.amazon.awssdk.services.glue.model.StorageDescriptor;
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.UpdateIcebergInput;
+import software.amazon.awssdk.services.glue.model.UpdateIcebergTableInput;
+import software.amazon.awssdk.services.glue.model.UpdateOpenTableFormatInput;
import software.amazon.awssdk.services.glue.model.UpdateTableRequest;
/**
@@ -117,6 +125,8 @@ public class GlueCatalogOperations implements
CatalogOperations, SupportsSchemas
/** Nullable — when null all table formats are exposed. */
@VisibleForTesting Set<String> tableFormatFilter;
+ @VisibleForTesting String defaultTableFormat;
+
private final GlueTypeConverter typeConverter = new GlueTypeConverter();
@Override
@@ -125,6 +135,9 @@ public class GlueCatalogOperations implements
CatalogOperations, SupportsSchemas
throws RuntimeException {
this.glueClient = GlueClientProvider.buildClient(config);
this.catalogId = config.get(GlueConstants.AWS_GLUE_CATALOG_ID);
+ this.defaultTableFormat =
+ config.getOrDefault(
+ GlueConstants.DEFAULT_TABLE_FORMAT,
GlueConstants.DEFAULT_TABLE_FORMAT_VALUE);
String filterProp =
config.getOrDefault(
GlueConstants.TABLE_FORMAT_FILTER,
GlueConstants.DEFAULT_TABLE_FORMAT_FILTER);
@@ -375,24 +388,52 @@ public class GlueCatalogOperations implements
CatalogOperations, SupportsSchemas
String dbName = schemaName(ident.namespace());
Map<String, String> props = properties != null ? properties :
Collections.emptyMap();
+ String tableFormat = props.getOrDefault(GlueConstants.TABLE_FORMAT,
defaultTableFormat);
+ boolean isIceberg =
GlueConstants.ICEBERG_TABLE_TYPE_VALUE.equalsIgnoreCase(tableFormat);
+
+ // For Iceberg tables, stamp table_type=ICEBERG into the Glue parameters
so that
+ // isIcebergTable() detection works consistently for future
alterTable/listTable calls.
+ Map<String, String> finalProps = props;
+ if (isIceberg) {
+ finalProps = new HashMap<>(props);
+ finalProps.put(GlueConstants.TABLE_TYPE_PARAM,
GlueConstants.ICEBERG_TABLE_TYPE_VALUE);
+ }
+
TableInput input =
buildTableInput(
- ident.name(), comment, columns, props, partitions, distribution,
sortOrders);
+ ident.name(), comment, columns, finalProps, partitions,
distribution, sortOrders);
CreateTableRequest.Builder req =
CreateTableRequest.builder().databaseName(dbName).tableInput(input);
- applyCatalogId(catalogId, req::catalogId);
- try {
- glueClient.createTable(req.build());
- } catch (EntityNotFoundException e) {
- throw new NoSuchSchemaException(e, "Schema %s does not exist", dbName);
- } catch (GlueException e) {
- throw GlueExceptionConverter.toTableException(e, "table " +
ident.name());
+ if (isIceberg) {
+ // Register mode: metadata_location points to existing Iceberg metadata.
+ // Create mode: new table; Glue writes metadata.json at the given
location.
+ boolean registerMode =
props.containsKey(GlueConstants.METADATA_LOCATION);
+ if (!registerMode) {
+ Preconditions.checkArgument(
+ props.containsKey(GlueConstants.LOCATION),
+ "Either '%s' (register existing table) or '%s' (create new table)
is required",
+ GlueConstants.METADATA_LOCATION,
+ GlueConstants.LOCATION);
+ req.openTableFormatInput(
+ OpenTableFormatInput.builder()
+ .icebergInput(
+ IcebergInput.builder()
+ .metadataOperation(MetadataOperation.CREATE)
+ .version(GlueConstants.ICEBERG_FORMAT_VERSION)
+ .build())
+ .build());
+ }
}
- LOG.info("Created Glue table {}.{}", dbName, ident.name());
+ executeCreateTable(dbName, ident, req);
+ LOG.info("Created {} table {}.{}", isIceberg ? "Iceberg" : "Glue", dbName,
ident.name());
+ if (isIceberg) {
+ // Load from Glue to pick up any server-set parameters (e.g.
current-schema-id).
+ return loadTable(ident);
+ }
GlueTable created =
GlueTable.builder()
.withName(ident.name())
@@ -416,9 +457,25 @@ public class GlueCatalogOperations implements
CatalogOperations, SupportsSchemas
public GlueTable alterTable(NameIdentifier ident, TableChange... changes)
throws NoSuchTableException, IllegalArgumentException {
- GlueTable current = loadTable(ident);
String dbName = schemaName(ident.namespace());
+ GetTableRequest.Builder rawReq =
+ GetTableRequest.builder().databaseName(dbName).name(ident.name());
+ applyCatalogId(catalogId, rawReq::catalogId);
+ Table rawGlueTable;
+ try {
+ rawGlueTable = glueClient.getTable(rawReq.build()).table();
+ } catch (GlueException e) {
+ throw GlueExceptionConverter.toTableException(e, "table " +
ident.name());
+ }
+
+ if (GlueIcebergHelper.isIcebergTable(rawGlueTable)) {
+ 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());
@@ -470,16 +527,7 @@ public class GlueCatalogOperations implements
CatalogOperations, SupportsSchemas
current.distribution(),
current.sortOrder());
- UpdateTableRequest.Builder req =
- UpdateTableRequest.builder().databaseName(dbName).tableInput(input);
- applyCatalogId(catalogId, req::catalogId);
-
- try {
- glueClient.updateTable(req.build());
- } catch (GlueException e) {
- throw GlueExceptionConverter.toTableException(e, "table " +
ident.name());
- }
-
+ executeUpdateTable(ident,
UpdateTableRequest.builder().databaseName(dbName).tableInput(input));
LOG.info("Altered Glue table {}.{}", dbName, ident.name());
GlueTable altered =
@@ -497,6 +545,55 @@ public class GlueCatalogOperations implements
CatalogOperations, SupportsSchemas
return altered;
}
+ private GlueTable alterIcebergTable(
+ NameIdentifier ident, String dbName, Table rawGlueTable, TableChange...
changes) {
+
+ GlueIcebergHelper.validateChanges(changes);
+
+ Optional<IcebergTableUpdate> schemaUpdate =
+ GlueIcebergHelper.buildSchemaUpdate(rawGlueTable, changes);
+ Map<String, String> propUpdates =
GlueIcebergHelper.extractSetProperties(changes);
+
+ if (schemaUpdate.isEmpty() && propUpdates.isEmpty()) {
+ LOG.debug("No-op alterIcebergTable for {}.{}", dbName, ident.name());
+ return loadTable(ident);
+ }
+
+ if (schemaUpdate.isPresent()) {
+ UpdateIcebergTableInput icebergTableInput =
+
UpdateIcebergTableInput.builder().updates(schemaUpdate.get()).build();
+ UpdateIcebergInput icebergInput =
+
UpdateIcebergInput.builder().updateIcebergTableInput(icebergTableInput).build();
+ UpdateOpenTableFormatInput openFormatInput =
+
UpdateOpenTableFormatInput.builder().updateIcebergInput(icebergInput).build();
+ executeUpdateTable(
+ ident,
+ UpdateTableRequest.builder()
+ .databaseName(dbName)
+ .updateOpenTableFormatInput(openFormatInput));
+ LOG.info("Altered Iceberg table {}.{} schema via Glue native API",
dbName, ident.name());
+ // Re-fetch to pick up server-side parameter changes (e.g.,
current-schema-id update)
+ // before using rawGlueTable.parameters() for the property update below.
+ GetTableRequest.Builder rawReq =
+ GetTableRequest.builder().databaseName(dbName).name(ident.name());
+ applyCatalogId(catalogId, rawReq::catalogId);
+ rawGlueTable = glueClient.getTable(rawReq.build()).table();
+ }
+
+ if (!propUpdates.isEmpty()) {
+ Map<String, String> newParams = new HashMap<>(rawGlueTable.parameters());
+ newParams.putAll(propUpdates);
+ executeUpdateTable(
+ ident,
+ UpdateTableRequest.builder()
+ .databaseName(dbName)
+ .tableInput(tableInputFromRaw(rawGlueTable, newParams)));
+ LOG.info("Altered Iceberg table {}.{} properties directly", dbName,
ident.name());
+ }
+
+ return loadTable(ident);
+ }
+
@Override
public boolean dropTable(NameIdentifier ident) {
String dbName = schemaName(ident.namespace());
@@ -527,11 +624,52 @@ public class GlueCatalogOperations implements
CatalogOperations, SupportsSchemas
private boolean matchesFormatFilter(Table table) {
if (tableFormatFilter == null) return true;
String fmt = table.hasParameters() ?
table.parameters().get(GlueConstants.TABLE_FORMAT) : null;
+ // Fall back to checking native Glue table_type for tables created without
the Gravitino
+ // table-format property (e.g. created via default-table-format config or
external tooling).
+ if (fmt == null && GlueIcebergHelper.isIcebergTable(table)) {
+ fmt = GlueConstants.ICEBERG_TABLE_TYPE_VALUE;
+ }
String normalized =
fmt != null ? fmt.toLowerCase(Locale.ROOT) :
GlueConstants.DEFAULT_TABLE_FORMAT_VALUE;
return tableFormatFilter.contains(normalized);
}
+ private void executeCreateTable(
+ String dbName, NameIdentifier ident, CreateTableRequest.Builder req) {
+ applyCatalogId(catalogId, req::catalogId);
+ try {
+ glueClient.createTable(req.build());
+ } catch (EntityNotFoundException e) {
+ throw new NoSuchSchemaException(e, "Schema %s does not exist", dbName);
+ } catch (GlueException e) {
+ throw GlueExceptionConverter.toTableException(e, "table " +
ident.name());
+ }
+ }
+
+ private void executeUpdateTable(NameIdentifier ident,
UpdateTableRequest.Builder req) {
+ applyCatalogId(catalogId, req::catalogId);
+ try {
+ glueClient.updateTable(req.build());
+ } catch (GlueException e) {
+ throw GlueExceptionConverter.toTableException(e, "table " +
ident.name());
+ }
+ }
+
+ /** Copies fields from {@code rawGlueTable} into a {@link TableInput} with
updated parameters. */
+ private static TableInput tableInputFromRaw(Table rawGlueTable, Map<String,
String> newParams) {
+ TableInput.Builder b =
+ TableInput.builder()
+ .name(rawGlueTable.name())
+ .description(rawGlueTable.description())
+ .tableType(rawGlueTable.tableType())
+ .parameters(newParams);
+ if (rawGlueTable.storageDescriptor() != null) {
+ b.storageDescriptor(rawGlueTable.storageDescriptor())
+ .partitionKeys(rawGlueTable.partitionKeys());
+ }
+ return b.build();
+ }
+
private TableInput buildTableInput(
String name,
String comment,
diff --git
a/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueClientProvider.java
b/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueClientProvider.java
index 536a3955bc..8f4aa666c2 100644
---
a/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueClientProvider.java
+++
b/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueClientProvider.java
@@ -88,7 +88,7 @@ public final class GlueClientProvider {
builder.credentialsProvider(
StaticCredentialsProvider.create(AwsBasicCredentials.create(accessKey,
secretKey)));
} else {
- builder.credentialsProvider(DefaultCredentialsProvider.create());
+
builder.credentialsProvider(DefaultCredentialsProvider.builder().build());
}
// Optional custom endpoint override for VPC endpoints or LocalStack
testing.
diff --git
a/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueConstants.java
b/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueConstants.java
index 2244f7552a..f919f27899 100644
---
a/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueConstants.java
+++
b/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueConstants.java
@@ -78,6 +78,30 @@ public final class GlueConstants {
/** Iceberg table metadata location stored in Glue {@code
Table.parameters()}. */
public static final String METADATA_LOCATION = "metadata_location";
+ /**
+ * Key of the {@code table_type} entry in {@code Table.parameters()} that
identifies the table
+ * format (e.g., {@code "ICEBERG"}).
+ */
+ public static final String TABLE_TYPE_PARAM = "table_type";
+
+ /** Value of {@link #TABLE_TYPE_PARAM} that identifies an Iceberg table. */
+ public static final String ICEBERG_TABLE_TYPE_VALUE = "ICEBERG";
+
+ /** Default Iceberg spec version used when creating Iceberg tables via Glue.
*/
+ public static final String ICEBERG_FORMAT_VERSION = "2";
+
+ /**
+ * Key of the {@code current-schema-id} entry in {@code Table.parameters()}
that holds the active
+ * Iceberg schema ID.
+ */
+ public static final String CURRENT_SCHEMA_ID_PARAM = "current-schema-id";
+
+ /** Glue column parameter key for the Iceberg field ID (assigned at table
creation). */
+ public static final String ICEBERG_FIELD_ID = "iceberg.field.id";
+
+ /** Glue column parameter key indicating whether the Iceberg field is
optional (nullable). */
+ public static final String ICEBERG_FIELD_OPTIONAL = "iceberg.field.optional";
+
// -------------------------------------------------------------------------
// StorageDescriptor-derived table properties (stored in Gravitino
properties map)
// -------------------------------------------------------------------------
diff --git
a/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueIcebergHelper.java
b/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueIcebergHelper.java
new file mode 100644
index 0000000000..2bc64e843a
--- /dev/null
+++
b/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueIcebergHelper.java
@@ -0,0 +1,435 @@
+/*
+ * 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 static
org.apache.gravitino.catalog.glue.GlueConstants.CURRENT_SCHEMA_ID_PARAM;
+import static org.apache.gravitino.catalog.glue.GlueConstants.ICEBERG_FIELD_ID;
+import static
org.apache.gravitino.catalog.glue.GlueConstants.ICEBERG_FIELD_OPTIONAL;
+import static org.apache.gravitino.catalog.glue.GlueConstants.TABLE_TYPE_PARAM;
+import static org.apache.gravitino.catalog.glue.GlueTypeConverter.BIGINT;
+import static org.apache.gravitino.catalog.glue.GlueTypeConverter.BINARY;
+import static org.apache.gravitino.catalog.glue.GlueTypeConverter.BOOLEAN;
+import static org.apache.gravitino.catalog.glue.GlueTypeConverter.CHAR;
+import static org.apache.gravitino.catalog.glue.GlueTypeConverter.DATE;
+import static org.apache.gravitino.catalog.glue.GlueTypeConverter.DECIMAL;
+import static org.apache.gravitino.catalog.glue.GlueTypeConverter.DOUBLE;
+import static
org.apache.gravitino.catalog.glue.GlueTypeConverter.DOUBLE_PRECISION;
+import static org.apache.gravitino.catalog.glue.GlueTypeConverter.FLOAT;
+import static org.apache.gravitino.catalog.glue.GlueTypeConverter.INT;
+import static org.apache.gravitino.catalog.glue.GlueTypeConverter.INTEGER;
+import static org.apache.gravitino.catalog.glue.GlueTypeConverter.LONG;
+import static org.apache.gravitino.catalog.glue.GlueTypeConverter.SMALLINT;
+import static org.apache.gravitino.catalog.glue.GlueTypeConverter.STRING;
+import static org.apache.gravitino.catalog.glue.GlueTypeConverter.TIMESTAMP;
+import static org.apache.gravitino.catalog.glue.GlueTypeConverter.TINYINT;
+import static org.apache.gravitino.catalog.glue.GlueTypeConverter.VARCHAR;
+
+import com.google.common.base.Preconditions;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.UnaryOperator;
+import org.apache.gravitino.rel.TableChange;
+import org.apache.gravitino.rel.types.Type;
+import org.apache.gravitino.rel.types.Types;
+import software.amazon.awssdk.core.document.Document;
+import software.amazon.awssdk.services.glue.model.Column;
+import software.amazon.awssdk.services.glue.model.IcebergSchema;
+import software.amazon.awssdk.services.glue.model.IcebergStructField;
+import software.amazon.awssdk.services.glue.model.IcebergStructTypeEnum;
+import software.amazon.awssdk.services.glue.model.IcebergTableUpdate;
+import software.amazon.awssdk.services.glue.model.StorageDescriptor;
+import software.amazon.awssdk.services.glue.model.Table;
+
+/**
+ * Utility methods for building Iceberg-specific Glue API structures used when
creating and
+ * modifying Iceberg-format tables via the AWS Glue Data Catalog.
+ */
+final class GlueIcebergHelper {
+
+ private GlueIcebergHelper() {}
+
+ /**
+ * Returns true if the Glue table is an Iceberg-format table.
+ *
+ * <p>Checks for {@code table_type=ICEBERG} in {@code Table.parameters()}.
+ */
+ static boolean isIcebergTable(Table glueTable) {
+ if (!glueTable.hasParameters()) return false;
+ return GlueConstants.ICEBERG_TABLE_TYPE_VALUE.equalsIgnoreCase(
+ glueTable.parameters().get(TABLE_TYPE_PARAM));
+ }
+
+ /**
+ * Validates that all {@code changes} are supported for Iceberg tables.
Throws {@link
+ * IllegalArgumentException} for unsupported change types, including {@link
+ * TableChange.RemoveProperty} (not supported via the Glue SDK) and
non-column/non-property
+ * changes (e.g., {@code RenameTable}, {@code UpdateComment}).
+ */
+ static void validateChanges(TableChange... changes) {
+ for (TableChange change : changes) {
+ if (change instanceof TableChange.ColumnChange || change instanceof
TableChange.SetProperty) {
+ continue;
+ }
+ if (change instanceof TableChange.RemoveProperty) {
+ throw new IllegalArgumentException(
+ "Removing properties from Iceberg tables is not supported via the
Glue SDK."
+ + " Use SetProperty to override values instead.");
+ }
+ throw new IllegalArgumentException(
+ "Unsupported table change for Iceberg table: " +
change.getClass().getSimpleName());
+ }
+ }
+
+ /**
+ * Filters {@code changes} to column-schema changes and builds a single
{@link IcebergTableUpdate}
+ * carrying the updated schema. Returns empty if there are no column changes.
+ */
+ static Optional<IcebergTableUpdate> buildSchemaUpdate(
+ Table rawGlueTable, TableChange... changes) {
+ List<TableChange> schemaChanges = new ArrayList<>();
+ for (TableChange c : changes) {
+ if (c instanceof TableChange.ColumnChange) schemaChanges.add(c);
+ }
+ if (schemaChanges.isEmpty()) return Optional.empty();
+ return Optional.of(
+ IcebergTableUpdate.builder()
+ .schema(buildUpdatedSchema(rawGlueTable, schemaChanges))
+ .build());
+ }
+
+ /**
+ * Extracts {@link TableChange.SetProperty} entries from {@code changes}
into a key-value map.
+ * Returns an empty map if there are no property-set changes.
+ */
+ static Map<String, String> extractSetProperties(TableChange... changes) {
+ Map<String, String> props = new HashMap<>();
+ for (TableChange change : changes) {
+ if (change instanceof TableChange.SetProperty) {
+ TableChange.SetProperty sp = (TableChange.SetProperty) change;
+ props.put(sp.getProperty(), sp.getValue());
+ }
+ }
+ return props;
+ }
+
+ /**
+ * Reads the current Iceberg schema from a Glue table's StorageDescriptor
columns (preserving
+ * {@code iceberg.field.id} from column parameters), applies the given
column changes, and returns
+ * the resulting {@link IcebergSchema} with an incremented schema ID.
+ */
+ static IcebergSchema buildUpdatedSchema(Table rawGlueTable,
List<TableChange> schemaChanges) {
+ List<IcebergStructField> fields = currentFields(rawGlueTable);
+ int maxId =
fields.stream().mapToInt(IcebergStructField::id).max().orElse(0);
+
+ for (TableChange change : schemaChanges) {
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ Preconditions.checkArgument(
+ add.fieldName().length == 1, "Nested column additions are not
supported");
+ fields.add(
+ IcebergStructField.builder()
+ .id(++maxId)
+ .name(add.fieldName()[0])
+ .type(gravitinoTypeToIcebergType(add.getDataType()))
+ .required(!add.isNullable())
+ .doc(add.getComment())
+ .build());
+
+ } else if (change instanceof TableChange.DeleteColumn) {
+ TableChange.DeleteColumn del = (TableChange.DeleteColumn) change;
+ Preconditions.checkArgument(
+ del.fieldName().length == 1, "Nested column deletions are not
supported");
+ String name = del.fieldName()[0];
+ boolean removed = fields.removeIf(f -> f.name().equals(name));
+ if (!removed && !del.getIfExists()) {
+ throw new IllegalArgumentException(
+ "Column '" + name + "' not found in Iceberg table schema");
+ }
+
+ } else if (change instanceof TableChange.RenameColumn) {
+ TableChange.RenameColumn rename = (TableChange.RenameColumn) change;
+ Preconditions.checkArgument(
+ rename.fieldName().length == 1, "Nested column renames are not
supported");
+ updateField(fields, rename.fieldName()[0], b ->
b.name(rename.getNewName()));
+
+ } else if (change instanceof TableChange.UpdateColumnType) {
+ TableChange.UpdateColumnType upd = (TableChange.UpdateColumnType)
change;
+ Preconditions.checkArgument(
+ upd.fieldName().length == 1, "Nested column type updates are not
supported");
+ Document newTypeDoc = gravitinoTypeToIcebergType(upd.getNewDataType());
+ updateField(fields, upd.fieldName()[0], b -> b.type(newTypeDoc));
+
+ } else if (change instanceof TableChange.UpdateColumnComment) {
+ TableChange.UpdateColumnComment upd =
(TableChange.UpdateColumnComment) change;
+ Preconditions.checkArgument(
+ upd.fieldName().length == 1, "Nested column comment updates are
not supported");
+ updateField(fields, upd.fieldName()[0], b ->
b.doc(upd.getNewComment()));
+
+ } else if (change instanceof TableChange.UpdateColumnNullability) {
+ TableChange.UpdateColumnNullability upd =
(TableChange.UpdateColumnNullability) change;
+ Preconditions.checkArgument(
+ upd.fieldName().length == 1, "Nested column nullability updates
are not supported");
+ updateField(fields, upd.fieldName()[0], b ->
b.required(!upd.nullable()));
+
+ } else {
+ throw new IllegalArgumentException(
+ "Unsupported column change: " + change.getClass().getSimpleName());
+ }
+ }
+
+ int currentSchemaId = parseSchemaId(rawGlueTable);
+ return IcebergSchema.builder()
+ .schemaId(currentSchemaId + 1)
+ .type(IcebergStructTypeEnum.STRUCT)
+ .fields(fields)
+ .build();
+ }
+
+ //
---------------------------------------------------------------------------
+ // Private helpers
+ //
---------------------------------------------------------------------------
+
+ /**
+ * Finds the first field with the given {@code name} in {@code fields} and
replaces it in-place
+ * with the result of applying {@code updater} to its builder. Throws if the
field is not found.
+ */
+ private static void updateField(
+ List<IcebergStructField> fields,
+ String name,
+ UnaryOperator<IcebergStructField.Builder> updater) {
+ for (int i = 0; i < fields.size(); i++) {
+ if (fields.get(i).name().equals(name)) {
+ fields.set(i, updater.apply(fields.get(i).toBuilder()).build());
+ return;
+ }
+ }
+ throw new IllegalArgumentException("Column '" + name + "' not found in
Iceberg table schema");
+ }
+
+ /** Reads current Iceberg fields from the Glue table's StorageDescriptor
columns. */
+ private static List<IcebergStructField> currentFields(Table rawGlueTable) {
+ List<IcebergStructField> fields = new ArrayList<>();
+ StorageDescriptor sd = rawGlueTable.storageDescriptor();
+ if (sd == null || !sd.hasColumns()) return fields;
+
+ int fallbackId = 0;
+ for (Column col : sd.columns()) {
+ int fieldId = parseFieldId(col, ++fallbackId);
+ boolean required = parseRequired(col);
+ fields.add(
+ IcebergStructField.builder()
+ .id(fieldId)
+ .name(col.name())
+ .type(hiveTypeToIcebergType(col.type()))
+ .required(required)
+ .doc(col.comment())
+ .build());
+ }
+
+ Set<Integer> seen = new HashSet<>();
+ for (IcebergStructField f : fields) {
+ if (!seen.add(f.id())) {
+ throw new IllegalStateException(
+ String.format(
+ "Iceberg table '%s' has duplicate field ID %d in column
metadata."
+ + " The Glue metadata may be corrupt. Aborting schema
update.",
+ rawGlueTable.name(), f.id()));
+ }
+ }
+ return fields;
+ }
+
+ /** Parses {@code iceberg.field.id} from column parameters; falls back to
{@code fallbackId}. */
+ private static int parseFieldId(Column col, int fallbackId) {
+ if (col.hasParameters()) {
+ String raw = col.parameters().get(ICEBERG_FIELD_ID);
+ if (raw != null) {
+ try {
+ return Integer.parseInt(raw);
+ } catch (NumberFormatException e) {
+ throw new IllegalStateException(
+ String.format(
+ "Column '%s' has non-numeric iceberg.field.id '%s'."
+ + " The Iceberg metadata may be corrupt. Aborting schema
update.",
+ col.name(), raw),
+ e);
+ }
+ }
+ }
+ return fallbackId;
+ }
+
+ /**
+ * Parses {@code iceberg.field.optional}; defaults to {@code false} (i.e.,
not required /
+ * optional).
+ */
+ private static boolean parseRequired(Column col) {
+ if (col.hasParameters()) {
+ String optional = col.parameters().get(ICEBERG_FIELD_OPTIONAL);
+ if (optional != null) {
+ return !Boolean.parseBoolean(optional);
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Reads the current schema ID from {@code
Table.parameters()["current-schema-id"]}; returns 0 if
+ * absent. Throws {@link IllegalStateException} if the value is present but
non-numeric, since
+ * proceeding with a fabricated schema ID risks corrupting the Iceberg
schema version chain.
+ */
+ private static int parseSchemaId(Table rawGlueTable) {
+ if (!rawGlueTable.hasParameters()) return 0;
+ String raw = rawGlueTable.parameters().get(CURRENT_SCHEMA_ID_PARAM);
+ if (raw == null) return 0;
+ try {
+ return Integer.parseInt(raw);
+ } catch (NumberFormatException e) {
+ throw new IllegalStateException(
+ String.format(
+ "Iceberg table '%s' has non-numeric current-schema-id '%s'."
+ + " The Iceberg metadata may be corrupt. Aborting schema
update.",
+ rawGlueTable.name(), raw),
+ e);
+ }
+ }
+
+ /**
+ * Converts a Glue/Hive column type string to an Iceberg type {@link
+ * software.amazon.awssdk.core.document.Document}.
+ *
+ * <p>Iceberg REST spec type names are used for primitives (e.g. {@code
"long"} for Hive {@code
+ * bigint}). Complex types (array, map, struct) are not supported and will
throw {@link
+ * IllegalStateException}.
+ */
+ static Document hiveTypeToIcebergType(String hiveType) {
+ if (hiveType == null) return Document.fromString(STRING);
+ String t = hiveType.toLowerCase(Locale.ROOT).trim();
+
+ switch (t) {
+ case STRING:
+ return Document.fromString(STRING);
+ case BIGINT:
+ case LONG: // Iceberg type name stored by Glue native Iceberg API
+ return Document.fromString(LONG);
+ case INT:
+ case INTEGER:
+ case SMALLINT:
+ case TINYINT:
+ return Document.fromString(INT);
+ case FLOAT:
+ return Document.fromString(FLOAT);
+ case DOUBLE:
+ case DOUBLE_PRECISION:
+ return Document.fromString(DOUBLE);
+ case BOOLEAN:
+ return Document.fromString(BOOLEAN);
+ case BINARY:
+ return Document.fromString(BINARY);
+ case DATE:
+ return Document.fromString(DATE);
+ case TIMESTAMP:
+ return Document.fromString(TIMESTAMP);
+ default:
+ break;
+ }
+
+ if (t.startsWith(DECIMAL + "(")) {
+ try {
+ String inner =
+ t.substring((DECIMAL + "(").length(), t.endsWith(")") ? t.length()
- 1 : t.length());
+ String[] parts = inner.split(",");
+ int precision = Integer.parseInt(parts[0].trim());
+ int scale = parts.length > 1 ? Integer.parseInt(parts[1].trim()) : 0;
+ return Document.fromMap(
+ Map.of(
+ "type", Document.fromString(DECIMAL),
+ "precision", Document.fromNumber(precision),
+ "scale", Document.fromNumber(scale)));
+ } catch (NumberFormatException e) {
+ throw new IllegalStateException(
+ "Malformed Hive decimal type '"
+ + hiveType
+ + "' in existing Iceberg table schema."
+ + " Cannot safely build schema update.",
+ e);
+ }
+ }
+ if (t.startsWith(VARCHAR + "(") || t.startsWith(CHAR + "(")) {
+ return Document.fromString(STRING);
+ }
+ throw new IllegalStateException(
+ "Unsupported Hive column type '"
+ + hiveType
+ + "' in existing Iceberg table schema."
+ + " Cannot safely build schema update."
+ + " Complex types (array, map, struct) are not supported by the
Glue Iceberg schema update API.");
+ }
+
+ /**
+ * Converts a Gravitino {@link Type} to an Iceberg type {@link
+ * software.amazon.awssdk.core.document.Document} ({@code "Doc"} in the
method name refers to the
+ * AWS SDK {@code Document} type used to represent Iceberg type descriptors
in Glue API payloads).
+ *
+ * <p>Supports primitive types. Complex types (list, map, struct) are not
supported and will throw
+ * {@link UnsupportedOperationException}.
+ */
+ static Document gravitinoTypeToIcebergType(Type type) {
+ if (type instanceof Types.StringType
+ || type instanceof Types.VarCharType
+ || type instanceof Types.FixedCharType) {
+ return Document.fromString("string");
+ }
+ if (type instanceof Types.LongType) return Document.fromString("long");
+ if (type instanceof Types.IntegerType
+ || type instanceof Types.ShortType
+ || type instanceof Types.ByteType) {
+ return Document.fromString("int");
+ }
+ if (type instanceof Types.FloatType) return Document.fromString("float");
+ if (type instanceof Types.DoubleType) return Document.fromString("double");
+ if (type instanceof Types.BooleanType) return
Document.fromString("boolean");
+ if (type instanceof Types.BinaryType || type instanceof Types.FixedType) {
+ return Document.fromString("binary");
+ }
+ if (type instanceof Types.DateType) return Document.fromString("date");
+ if (type instanceof Types.TimeType) return Document.fromString("time");
+ if (type instanceof Types.TimestampType) {
+ return Document.fromString(
+ ((Types.TimestampType) type).hasTimeZone() ? "timestamptz" :
"timestamp");
+ }
+ if (type instanceof Types.UUIDType) return Document.fromString("uuid");
+ if (type instanceof Types.DecimalType) {
+ Types.DecimalType dt = (Types.DecimalType) type;
+ return Document.fromMap(
+ Map.of(
+ "type", Document.fromString("decimal"),
+ "precision", Document.fromNumber(dt.precision()),
+ "scale", Document.fromNumber(dt.scale())));
+ }
+ throw new UnsupportedOperationException(
+ "Iceberg Glue catalog does not support type: " + type.simpleString());
+ }
+}
diff --git
a/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueTypeConverter.java
b/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueTypeConverter.java
index a35beb51d5..ff82962848 100644
---
a/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueTypeConverter.java
+++
b/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueTypeConverter.java
@@ -47,8 +47,10 @@ public class GlueTypeConverter implements
DataTypeConverter<String, String> {
static final String INT = "int";
static final String INTEGER = "integer";
static final String BIGINT = "bigint";
+ static final String LONG = "long";
static final String FLOAT = "float";
static final String DOUBLE = "double";
+ static final String DOUBLE_PRECISION = "double precision";
static final String STRING = "string";
static final String DATE = "date";
static final String TIMESTAMP = "timestamp";
diff --git
a/catalogs/catalog-glue/src/test/java/org/apache/gravitino/catalog/glue/TestGlueIceberg.java
b/catalogs/catalog-glue/src/test/java/org/apache/gravitino/catalog/glue/TestGlueIceberg.java
new file mode 100644
index 0000000000..89b5cd7e42
--- /dev/null
+++
b/catalogs/catalog-glue/src/test/java/org/apache/gravitino/catalog/glue/TestGlueIceberg.java
@@ -0,0 +1,597 @@
+/*
+ * 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 static
org.apache.gravitino.catalog.glue.GlueConstants.CURRENT_SCHEMA_ID_PARAM;
+import static
org.apache.gravitino.catalog.glue.GlueConstants.ICEBERG_TABLE_TYPE_VALUE;
+import static org.apache.gravitino.catalog.glue.GlueConstants.TABLE_TYPE_PARAM;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.rel.TableChange;
+import org.apache.gravitino.rel.expressions.distributions.Distributions;
+import org.apache.gravitino.rel.expressions.transforms.Transform;
+import org.apache.gravitino.rel.indexes.Indexes;
+import org.apache.gravitino.rel.types.Types;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import software.amazon.awssdk.services.glue.GlueClient;
+import software.amazon.awssdk.services.glue.model.Column;
+import software.amazon.awssdk.services.glue.model.CreateTableRequest;
+import software.amazon.awssdk.services.glue.model.CreateTableResponse;
+import software.amazon.awssdk.services.glue.model.GetTableRequest;
+import software.amazon.awssdk.services.glue.model.GetTableResponse;
+import software.amazon.awssdk.services.glue.model.IcebergSchema;
+import software.amazon.awssdk.services.glue.model.IcebergStructField;
+import software.amazon.awssdk.services.glue.model.StorageDescriptor;
+import software.amazon.awssdk.services.glue.model.Table;
+import software.amazon.awssdk.services.glue.model.UpdateTableRequest;
+import software.amazon.awssdk.services.glue.model.UpdateTableResponse;
+
+class TestGlueIceberg {
+
+ private static final String DB = "mydb";
+ private static final String TABLE = "ice1";
+ private static final String LOCATION = "s3://my-bucket/warehouse/ice1";
+
+ private GlueClient mockClient;
+ private GlueCatalogOperations ops;
+
+ @BeforeEach
+ void setup() {
+ mockClient = mock(GlueClient.class);
+ ops = new GlueCatalogOperations();
+ ops.glueClient = mockClient;
+ ops.catalogId = null;
+ ops.tableFormatFilter = null;
+ ops.defaultTableFormat = GlueConstants.DEFAULT_TABLE_FORMAT_VALUE;
+ }
+
+ //
---------------------------------------------------------------------------
+ // isIcebergTable
+ //
---------------------------------------------------------------------------
+
+ @Test
+ void testIsIcebergTable_withIcebergType() {
+ Table t =
+ Table.builder().parameters(Map.of(TABLE_TYPE_PARAM,
ICEBERG_TABLE_TYPE_VALUE)).build();
+ assertTrue(GlueIcebergHelper.isIcebergTable(t));
+ }
+
+ @Test
+ void testIsIcebergTable_caseInsensitive() {
+ Table t = Table.builder().parameters(Map.of("table_type",
"iceberg")).build();
+ assertTrue(GlueIcebergHelper.isIcebergTable(t));
+ }
+
+ @Test
+ void testIsIcebergTable_hiveTable() {
+ Table t = Table.builder().parameters(Map.of("table_type", "HIVE")).build();
+ assertFalse(GlueIcebergHelper.isIcebergTable(t));
+ }
+
+ @Test
+ void testIsIcebergTable_noParameters() {
+ Table t = Table.builder().build();
+ assertFalse(GlueIcebergHelper.isIcebergTable(t));
+ }
+
+ //
---------------------------------------------------------------------------
+ // hiveTypeToIcebergType
+ //
---------------------------------------------------------------------------
+
+ @Test
+ void testHiveTypeToDoc_primitives() {
+ assertEquals("long",
GlueIcebergHelper.hiveTypeToIcebergType("bigint").asString());
+ assertEquals("int",
GlueIcebergHelper.hiveTypeToIcebergType("int").asString());
+ assertEquals("int",
GlueIcebergHelper.hiveTypeToIcebergType("smallint").asString());
+ assertEquals("float",
GlueIcebergHelper.hiveTypeToIcebergType("float").asString());
+ assertEquals("double",
GlueIcebergHelper.hiveTypeToIcebergType("double").asString());
+ assertEquals("boolean",
GlueIcebergHelper.hiveTypeToIcebergType("boolean").asString());
+ assertEquals("binary",
GlueIcebergHelper.hiveTypeToIcebergType("binary").asString());
+ assertEquals("date",
GlueIcebergHelper.hiveTypeToIcebergType("date").asString());
+ assertEquals("timestamp",
GlueIcebergHelper.hiveTypeToIcebergType("timestamp").asString());
+ assertEquals("string",
GlueIcebergHelper.hiveTypeToIcebergType("string").asString());
+ }
+
+ @Test
+ void testHiveTypeToDoc_decimal() {
+ var doc = GlueIcebergHelper.hiveTypeToIcebergType("decimal(18,2)");
+ var m = doc.asMap();
+ assertEquals("decimal", m.get("type").asString());
+ assertEquals(18, m.get("precision").asNumber().intValue());
+ assertEquals(2, m.get("scale").asNumber().intValue());
+ }
+
+ @Test
+ void testHiveTypeToDoc_varchar() {
+ assertEquals("string",
GlueIcebergHelper.hiveTypeToIcebergType("varchar(255)").asString());
+ }
+
+ //
---------------------------------------------------------------------------
+ // gravitinoTypeToIcebergType
+ //
---------------------------------------------------------------------------
+
+ @Test
+ void testGravitinoTypeToDoc_primitives() {
+ assertEquals(
+ "long",
GlueIcebergHelper.gravitinoTypeToIcebergType(Types.LongType.get()).asString());
+ assertEquals(
+ "int",
GlueIcebergHelper.gravitinoTypeToIcebergType(Types.IntegerType.get()).asString());
+ assertEquals(
+ "float",
GlueIcebergHelper.gravitinoTypeToIcebergType(Types.FloatType.get()).asString());
+ assertEquals(
+ "double",
GlueIcebergHelper.gravitinoTypeToIcebergType(Types.DoubleType.get()).asString());
+ assertEquals(
+ "boolean",
+
GlueIcebergHelper.gravitinoTypeToIcebergType(Types.BooleanType.get()).asString());
+ assertEquals(
+ "string",
GlueIcebergHelper.gravitinoTypeToIcebergType(Types.StringType.get()).asString());
+ assertEquals(
+ "date",
GlueIcebergHelper.gravitinoTypeToIcebergType(Types.DateType.get()).asString());
+ assertEquals(
+ "timestamp",
+
GlueIcebergHelper.gravitinoTypeToIcebergType(Types.TimestampType.withoutTimeZone())
+ .asString());
+ assertEquals(
+ "timestamptz",
+
GlueIcebergHelper.gravitinoTypeToIcebergType(Types.TimestampType.withTimeZone())
+ .asString());
+ assertEquals(
+ "uuid",
GlueIcebergHelper.gravitinoTypeToIcebergType(Types.UUIDType.get()).asString());
+ }
+
+ @Test
+ void testGravitinoTypeToDoc_decimal() {
+ var doc =
GlueIcebergHelper.gravitinoTypeToIcebergType(Types.DecimalType.of(10, 3));
+ var m = doc.asMap();
+ assertEquals("decimal", m.get("type").asString());
+ assertEquals(10, m.get("precision").asNumber().intValue());
+ assertEquals(3, m.get("scale").asNumber().intValue());
+ }
+
+ @Test
+ void testGravitinoTypeToDoc_unsupported() {
+ assertThrows(
+ UnsupportedOperationException.class,
+ () ->
GlueIcebergHelper.gravitinoTypeToIcebergType(Types.NullType.get()));
+ }
+
+ //
---------------------------------------------------------------------------
+ // buildSchemaUpdate / extractSetProperties / validateChanges
+ //
---------------------------------------------------------------------------
+
+ @Test
+ void testBuildSchemaUpdate_addColumn() {
+ Table raw = icebergTable(icebergColumn("id", "long", 1, false));
+
+ TableChange add = TableChange.addColumn(new String[] {"score"},
Types.FloatType.get(), true);
+ IcebergSchema schema = GlueIcebergHelper.buildSchemaUpdate(raw,
add).orElseThrow().schema();
+
+ assertEquals(2, schema.fields().size());
+ IcebergStructField newField = schema.fields().get(1);
+ assertEquals("score", newField.name());
+ assertEquals("float", newField.type().asString());
+ assertEquals(2, newField.id());
+ assertFalse(newField.required());
+ }
+
+ @Test
+ void testBuildSchemaUpdate_deleteColumn() {
+ Table raw =
+ icebergTable(
+ icebergColumn("id", "long", 1, false), icebergColumn("name",
"string", 2, true));
+
+ TableChange delete = TableChange.deleteColumn(new String[] {"name"}, true);
+ IcebergSchema schema = GlueIcebergHelper.buildSchemaUpdate(raw,
delete).orElseThrow().schema();
+
+ assertEquals(1, schema.fields().size());
+ assertEquals("id", schema.fields().get(0).name());
+ }
+
+ @Test
+ void testBuildSchemaUpdate_renameColumn() {
+ Table raw = icebergTable(icebergColumn("old_name", "string", 1, true));
+
+ TableChange rename = TableChange.renameColumn(new String[] {"old_name"},
"new_name");
+ IcebergSchema schema = GlueIcebergHelper.buildSchemaUpdate(raw,
rename).orElseThrow().schema();
+
+ assertEquals("new_name", schema.fields().get(0).name());
+ assertEquals(1, schema.fields().get(0).id()); // ID preserved
+ }
+
+ @Test
+ void testExtractSetProperties_singleProperty() {
+ TableChange set = TableChange.setProperty("write.format.default",
"parquet");
+ Map<String, String> props = GlueIcebergHelper.extractSetProperties(set);
+ assertEquals(1, props.size());
+ assertEquals("parquet", props.get("write.format.default"));
+ }
+
+ @Test
+ void testExtractSetProperties_empty() {
+ Map<String, String> props =
+ GlueIcebergHelper.extractSetProperties(
+ TableChange.addColumn(new String[] {"col"}, Types.LongType.get()));
+ assertTrue(props.isEmpty());
+ }
+
+ @Test
+ void testValidateChanges_removePropertyThrows() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
GlueIcebergHelper.validateChanges(TableChange.removeProperty("some.prop")));
+ }
+
+ @Test
+ void testBuildSchemaUpdate_schemaIdIncrement() {
+ Table raw =
+ Table.builder()
+ .parameters(
+ Map.of(TABLE_TYPE_PARAM, ICEBERG_TABLE_TYPE_VALUE,
CURRENT_SCHEMA_ID_PARAM, "3"))
+ .storageDescriptor(StorageDescriptor.builder().build())
+ .build();
+
+ TableChange add = TableChange.addColumn(new String[] {"ts"},
Types.DateType.get(), true);
+ IcebergSchema schema = GlueIcebergHelper.buildSchemaUpdate(raw,
add).orElseThrow().schema();
+ assertEquals(4, schema.schemaId());
+ }
+
+ @Test
+ void testBuildSchemaUpdate_andExtractSetProperties_mixedChanges() {
+ Table raw = icebergTable(icebergColumn("id", "long", 1, false));
+ TableChange colChange = TableChange.addColumn(new String[] {"ts"},
Types.DateType.get(), true);
+ TableChange propChange =
TableChange.setProperty("write.target-file-size-bytes", "134217728");
+
+ IcebergSchema schema =
+ GlueIcebergHelper.buildSchemaUpdate(raw, colChange,
propChange).orElseThrow().schema();
+ assertEquals(2, schema.fields().size());
+ assertEquals("ts", schema.fields().get(1).name());
+
+ Map<String, String> props =
GlueIcebergHelper.extractSetProperties(colChange, propChange);
+ assertEquals("134217728", props.get("write.target-file-size-bytes"));
+ }
+
+ @Test
+ void testBuildSchemaUpdate_updateColumnType() {
+ Table raw = icebergTable(icebergColumn("id", "int", 1, false));
+
+ IcebergSchema schema =
+ GlueIcebergHelper.buildSchemaUpdate(
+ raw, TableChange.updateColumnType(new String[] {"id"},
Types.LongType.get()))
+ .orElseThrow()
+ .schema();
+ assertEquals("long", schema.fields().get(0).type().asString());
+ assertEquals(1, schema.fields().get(0).id()); // field ID preserved
+ }
+
+ @Test
+ void testBuildSchemaUpdate_emptyChanges() {
+ Table raw = icebergTable(icebergColumn("id", "long", 1, false));
+ assertFalse(GlueIcebergHelper.buildSchemaUpdate(raw).isPresent());
+ }
+
+ @Test
+ void testBuildSchemaUpdate_nonSequentialFieldIds() {
+ Table raw =
+ icebergTable(
+ icebergColumn("a", "long", 1, false),
+ icebergColumn("b", "string", 5, true),
+ icebergColumn("c", "date", 10, true));
+
+ IcebergSchema schema =
+ GlueIcebergHelper.buildSchemaUpdate(
+ raw, TableChange.addColumn(new String[] {"d"},
Types.IntegerType.get(), true))
+ .orElseThrow()
+ .schema();
+ IcebergStructField newField = schema.fields().get(3);
+ assertEquals("d", newField.name());
+ assertEquals(11, newField.id()); // max(1,5,10) + 1 = 11, not
fields.size()+1 = 4
+ }
+
+ @Test
+ void testBuildSchemaUpdate_columnNotFoundThrows() {
+ Table raw = icebergTable(icebergColumn("id", "long", 1, false));
+
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ GlueIcebergHelper.buildSchemaUpdate(
+ raw, TableChange.renameColumn(new String[] {"nonexistent"},
"new_name")));
+ }
+
+ @Test
+ void testHiveTypeToDoc_unknownTypeThrows() {
+ // Complex types (array, map, struct) in existing Iceberg schema must
abort, not silently retype
+ assertThrows(
+ IllegalStateException.class,
+ () -> GlueIcebergHelper.hiveTypeToIcebergType("array<string>"));
+ }
+
+ @Test
+ void testHiveTypeToDoc_malformedDecimalThrows() {
+ assertThrows(
+ IllegalStateException.class,
+ () -> GlueIcebergHelper.hiveTypeToIcebergType("decimal(abc,2)"));
+ }
+
+ //
---------------------------------------------------------------------------
+ // createTable Iceberg routing
+ //
---------------------------------------------------------------------------
+
+ @Test
+ void testCreateTable_icebergRoutesOpenTableFormatInput() {
+ Table created =
+ Table.builder()
+ .name(TABLE)
+ .parameters(
+ Map.of(
+ "table_type",
+ "ICEBERG",
+ "metadata_location",
+ LOCATION + "/metadata/00000.metadata.json"))
+ .storageDescriptor(StorageDescriptor.builder().build())
+ .build();
+
+ when(mockClient.createTable(any(CreateTableRequest.class)))
+ .thenReturn(CreateTableResponse.builder().build());
+ when(mockClient.getTable(any(GetTableRequest.class)))
+ .thenReturn(GetTableResponse.builder().table(created).build());
+
+ NameIdentifier ident = NameIdentifier.of("cat", "ns", DB, TABLE);
+ GlueColumn[] cols = {
+
GlueColumn.builder().withName("id").withType(Types.LongType.get()).withNullable(false).build()
+ };
+
+ ops.createTable(
+ ident,
+ cols,
+ "iceberg table",
+ Map.of(GlueConstants.TABLE_FORMAT, "iceberg", GlueConstants.LOCATION,
LOCATION),
+ new Transform[0],
+ Distributions.NONE,
+ null,
+ Indexes.EMPTY_INDEXES);
+
+ ArgumentCaptor<CreateTableRequest> captor =
ArgumentCaptor.forClass(CreateTableRequest.class);
+ verify(mockClient).createTable(captor.capture());
+ CreateTableRequest req = captor.getValue();
+ assertNotNull(
+ req.openTableFormatInput(), "openTableFormatInput must be set for
Iceberg tables");
+ assertNotNull(req.openTableFormatInput().icebergInput());
+ }
+
+ //
---------------------------------------------------------------------------
+ // alterTable Iceberg routing
+ //
---------------------------------------------------------------------------
+
+ @Test
+ void testAlterTable_icebergRoutesUpdateOpenTableFormatInput() {
+ Table rawTable = icebergTable(icebergColumn("id", "long", 1, false));
+
+ // getTable is called multiple times: initial load, re-fetch after schema
update, loadTable
+ when(mockClient.getTable(any(GetTableRequest.class)))
+ .thenReturn(GetTableResponse.builder().table(rawTable).build());
+ when(mockClient.updateTable(any(UpdateTableRequest.class)))
+ .thenReturn(UpdateTableResponse.builder().build());
+
+ NameIdentifier ident = NameIdentifier.of("cat", "ns", DB, TABLE);
+ ops.alterTable(
+ ident, TableChange.addColumn(new String[] {"score"},
Types.DoubleType.get(), true));
+
+ ArgumentCaptor<UpdateTableRequest> captor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ verify(mockClient).updateTable(captor.capture());
+ UpdateTableRequest req = captor.getValue();
+ assertNotNull(
+ req.updateOpenTableFormatInput(),
+ "updateOpenTableFormatInput must be set for Iceberg alter");
+ assertNotNull(req.updateOpenTableFormatInput().updateIcebergInput());
+ }
+
+ @Test
+ void testCreateTable_icebergMissingLocationThrows() {
+ NameIdentifier ident = NameIdentifier.of("cat", "ns", DB, TABLE);
+ GlueColumn[] cols = {
+
GlueColumn.builder().withName("id").withType(Types.LongType.get()).withNullable(false).build()
+ };
+
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ ops.createTable(
+ ident,
+ cols,
+ "iceberg table",
+ Map.of(GlueConstants.TABLE_FORMAT, "iceberg"), // missing
location
+ new Transform[0],
+ Distributions.NONE,
+ null,
+ Indexes.EMPTY_INDEXES));
+ }
+
+ @Test
+ void testAlterTable_icebergRenameThrows() {
+ Table rawTable = icebergTable(icebergColumn("id", "long", 1, false));
+
+ when(mockClient.getTable(any(GetTableRequest.class)))
+ .thenReturn(GetTableResponse.builder().table(rawTable).build());
+
+ NameIdentifier ident = NameIdentifier.of("cat", "ns", DB, TABLE);
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> ops.alterTable(ident, TableChange.rename("new_table_name")));
+ }
+
+ @Test
+ void testHiveTypeToDoc_nullFallsBackToString() {
+ assertEquals("string",
GlueIcebergHelper.hiveTypeToIcebergType(null).asString());
+ }
+
+ @Test
+ void testBuildSchemaUpdate_updateColumnNullability() {
+ Table raw = icebergTable(icebergColumn("id", "long", 1, true));
+
+ IcebergSchema schema =
+ GlueIcebergHelper.buildSchemaUpdate(
+ raw, TableChange.updateColumnNullability(new String[] {"id"},
false))
+ .orElseThrow()
+ .schema();
+ assertTrue(schema.fields().get(0).required());
+ }
+
+ @Test
+ void testBuildSchemaUpdate_updateColumnComment() {
+ Table raw = icebergTable(icebergColumn("id", "long", 1, false));
+
+ IcebergSchema schema =
+ GlueIcebergHelper.buildSchemaUpdate(
+ raw, TableChange.updateColumnComment(new String[] {"id"},
"primary key"))
+ .orElseThrow()
+ .schema();
+ assertEquals("primary key", schema.fields().get(0).doc());
+ }
+
+ @Test
+ void testCreateTable_defaultTableFormatIcebergRoutesOpenTableFormatInput() {
+ Table created =
+ Table.builder()
+ .name(TABLE)
+ .parameters(
+ Map.of(TABLE_TYPE_PARAM, ICEBERG_TABLE_TYPE_VALUE,
"metadata_location", LOCATION))
+ .storageDescriptor(StorageDescriptor.builder().build())
+ .build();
+ when(mockClient.createTable(any(CreateTableRequest.class)))
+ .thenReturn(CreateTableResponse.builder().build());
+ when(mockClient.getTable(any(GetTableRequest.class)))
+ .thenReturn(GetTableResponse.builder().table(created).build());
+
+ // Set defaultTableFormat to iceberg so createTable without table-format
property routes
+ // through OpenTableFormatInput
+ ops.defaultTableFormat = "iceberg";
+ NameIdentifier ident = NameIdentifier.of("cat", "ns", DB, TABLE);
+ GlueColumn[] cols = {
+
GlueColumn.builder().withName("id").withType(Types.LongType.get()).withNullable(false).build()
+ };
+ ops.createTable(
+ ident,
+ cols,
+ "iceberg table",
+ Map.of(GlueConstants.LOCATION, LOCATION), // no table-format; relies
on defaultTableFormat
+ new Transform[0],
+ Distributions.NONE,
+ null,
+ Indexes.EMPTY_INDEXES);
+
+ ArgumentCaptor<CreateTableRequest> captor =
ArgumentCaptor.forClass(CreateTableRequest.class);
+ verify(mockClient).createTable(captor.capture());
+ assertNotNull(
+ captor.getValue().openTableFormatInput(),
+ "defaultTableFormat=iceberg should route through
openTableFormatInput");
+ }
+
+ @Test
+ void testMatchesFormatFilter_icebergFallbackViaTableType() {
+ // Table has table_type=ICEBERG but no table-format property (e.g. created
by external tooling)
+ Table externalIceberg =
+ Table.builder()
+ .name(TABLE)
+ .parameters(Map.of(TABLE_TYPE_PARAM, ICEBERG_TABLE_TYPE_VALUE))
+ .build();
+
+ ops.tableFormatFilter = Set.of("iceberg");
+ // matchesFormatFilter is private; test via listTables by mocking getTable
+ // Instead verify the helper logic directly via isIcebergTable + manual
check
+ assertTrue(
+ GlueIcebergHelper.isIcebergTable(externalIceberg),
+ "Table with table_type=ICEBERG should be recognized as Iceberg");
+
+ // A hive table should not match iceberg filter
+ Table hiveTable =
+ Table.builder().name("hive_t").parameters(Map.of("table_type",
"HIVE")).build();
+ assertFalse(GlueIcebergHelper.isIcebergTable(hiveTable));
+ }
+
+ @Test
+ void testBuildSchemaUpdate_nestedFieldThrows() {
+ Table raw = icebergTable(icebergColumn("id", "long", 1, false));
+
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ GlueIcebergHelper.buildSchemaUpdate(
+ raw, TableChange.deleteColumn(new String[] {"nested",
"field"}, false)));
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ GlueIcebergHelper.buildSchemaUpdate(
+ raw, TableChange.renameColumn(new String[] {"nested",
"field"}, "new_name")));
+ }
+
+ @Test
+ void testCurrentFields_malformedFieldIdThrows() {
+ Column col =
+ Column.builder()
+ .name("id")
+ .type("long")
+ .parameters(Map.of(GlueConstants.ICEBERG_FIELD_ID, "not-a-number"))
+ .build();
+ Table raw = icebergTable(col);
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ GlueIcebergHelper.buildSchemaUpdate(
+ raw, TableChange.deleteColumn(new String[] {"id"}, false)));
+ }
+
+ //
---------------------------------------------------------------------------
+ // Helpers
+ //
---------------------------------------------------------------------------
+
+ private static Column icebergColumn(String name, String type, int fieldId,
boolean optional) {
+ return Column.builder()
+ .name(name)
+ .type(type)
+ .parameters(
+ Map.of(
+ GlueConstants.ICEBERG_FIELD_ID,
+ String.valueOf(fieldId),
+ GlueConstants.ICEBERG_FIELD_OPTIONAL,
+ String.valueOf(optional)))
+ .build();
+ }
+
+ private static Table icebergTable(Column... columns) {
+ return Table.builder()
+ .name(TABLE)
+ .parameters(Map.of(TABLE_TYPE_PARAM, ICEBERG_TABLE_TYPE_VALUE))
+
.storageDescriptor(StorageDescriptor.builder().columns(columns).build())
+ .build();
+ }
+}
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 94f6196eb8..d206525a61 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -17,7 +17,7 @@
# under the License.
#
[versions]
-awssdk = "2.29.52"
+awssdk = "2.31.73"
azure-identity = "1.13.1"
azure-storage-file-datalake = "12.20.0"
reactor-netty-http = "1.2.1"
