This is an automated email from the ASF dual-hosted git repository.
roryqi 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 6126562ff8 [#14545] fix(iceberg-rest): assign missing partition field
ids, allow v3 date promotions, refuse decimals Gravitino cannot import (#14546)
6126562ff8 is described below
commit 6126562ff891b000d9a39e7032dcfd76280efcb6
Author: Mark Hoerth <[email protected]>
AuthorDate: Mon Nov 2 22:07:25 2026 -0800
[#14545] fix(iceberg-rest): assign missing partition field ids, allow v3
date promotions, refuse decimals Gravitino cannot import (#14546)
### What changes were proposed in this pull request?
- `IcebergCommitValidation.assignPartitionFieldIds`, called from
`CatalogWrapperForREST.updateTable`: an `add-spec` field without a
`field-id` gets the id of the same field (source and transform) in an
earlier spec, or the next id above `last-partition-id`, from the table's
current metadata. Creates and requests whose fields all carry ids are
unchanged and load no metadata. The commit validation still checks the
result on every retry, so a concurrent change that takes the id is
refused.
- The schema check allows `date` to `timestamp` and `timestamp_ns`
(without time zone) on v3 tables, as the v3 table spec does.
- `IcebergColumnFieldValidator` checks every decimal type, nested ones
included, with Gravitino's own `Types.DecimalType` rule before a create,
register or schema commit, so nothing is created that cannot be
imported.
### Why are the changes needed?
A valid `add-spec` was refused, valid v3 promotions were refused, and an
invalid decimal left a created, unowned table behind a 400.
Fixes #14545
### Does this PR introduce _any_ user-facing change?
Yes. `add-spec` without field ids succeeds with correctly numbered ids;
v3 date promotions are accepted; a create with a decimal Gravitino
cannot represent is refused with 400 before anything is created.
### How was this patch tested?
-
`TestIcebergCommitValidation#testAddSpecWithoutFieldIdsGetsTheTablesIds`,
`#testAddSpecWithFieldIdsIsUnchanged`,
`#testDatePromotionsAreAllowedOnlyFromV3`.
-
`TestIcebergTableOperationExecutor#testRejectsUnrepresentableDecimalBeforeCreate`.
- The issue's curl reproduction, which fails before this change.
`./gradlew :iceberg:iceberg-rest-server:test -PskipITs`
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_01Eu6fb9pLfmrSJzGQDVLi1g
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../iceberg/service/CatalogWrapperForREST.java | 15 +-
.../iceberg/service/IcebergCommitValidation.java | 151 ++++++++++++++++++++-
.../dispatcher/IcebergColumnFieldValidator.java | 25 ++++
.../service/TestIcebergCommitValidation.java | 71 ++++++++++
.../TestIcebergTableOperationExecutor.java | 31 +++++
5 files changed, 289 insertions(+), 4 deletions(-)
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
index 8ff9b6de69..921a6bc127 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
@@ -34,6 +34,7 @@ import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Stream;
+import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.credential.CatalogCredentialManager;
@@ -53,6 +54,7 @@ import org.apache.gravitino.utils.PrincipalUtils;
import org.apache.iceberg.BaseTable;
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.IncrementalAppendScan;
import org.apache.iceberg.Scan;
import org.apache.iceberg.Table;
@@ -171,10 +173,21 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
@Override
public LoadTableResponse updateTable(
TableIdentifier tableIdentifier, UpdateTableRequest updateTableRequest) {
+ UpdateTableRequest request =
+ IcebergCommitValidation.assignPartitionFieldIds(
+ updateTableRequest, () -> currentMetadata(tableIdentifier));
return super.updateTable(
tableIdentifier,
IcebergCommitValidation.withValidation(
- updateTableRequest, () ->
getCatalog().loadTable(tableIdentifier).io()));
+ request, () -> getCatalog().loadTable(tableIdentifier).io()));
+ }
+
+ @Nullable
+ private TableMetadata currentMetadata(TableIdentifier tableIdentifier) {
+ Table table = getCatalog().loadTable(tableIdentifier);
+ return table instanceof HasTableOperations
+ ? ((HasTableOperations) table).operations().current()
+ : null;
}
public LoadTableResponse registerTable(
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCommitValidation.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCommitValidation.java
index d59271d6ab..dd5eaf5c71 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCommitValidation.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCommitValidation.java
@@ -18,9 +18,13 @@
*/
package org.apache.gravitino.iceberg.service;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -34,6 +38,7 @@ import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.MetadataUpdate;
import org.apache.iceberg.PartitionField;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionSpecParser;
import org.apache.iceberg.RetryableValidationException;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
@@ -42,6 +47,7 @@ import org.apache.iceberg.SnapshotRefType;
import org.apache.iceberg.SortField;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.UnboundPartitionSpec;
import org.apache.iceberg.UpdateRequirement;
import org.apache.iceberg.exceptions.CommitFailedException;
import org.apache.iceberg.exceptions.ValidationException;
@@ -50,6 +56,7 @@ import org.apache.iceberg.rest.requests.UpdateTableRequest;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.JsonUtil;
/**
* Checks a table commit against the metadata it is applied to, before Iceberg
applies it.
@@ -125,6 +132,131 @@ public final class IcebergCommitValidation implements
UpdateRequirement {
return new UpdateTableRequest(requirements.build(), request.updates());
}
+ /**
+ * Gives partition fields that an add-spec sends without a {@code field-id}
the id the table
+ * assigns: the id of the same field (source and transform) in an earlier
spec, or the next id
+ * above {@code last-partition-id}. The REST spec makes {@code field-id}
optional, but Iceberg's
+ * metadata builder numbers such fields from 1000 again, so they would take
the ids of existing
+ * partition fields.
+ *
+ * @param request the table commit request
+ * @param current supplies the table's current metadata, or null if it
cannot be loaded; called
+ * only when a field needs an id
+ * @return the request with every added partition field carrying an id
+ */
+ public static UpdateTableRequest assignPartitionFieldIds(
+ UpdateTableRequest request, Supplier<TableMetadata> current) {
+ if (request == null
+ || request.updates() == null
+ || request.requirements().stream()
+
.anyMatch(UpdateRequirement.AssertTableDoesNotExist.class::isInstance)) {
+ // A create starts from no partition fields, so numbering from 1000 is
right.
+ return request;
+ }
+ boolean missingIds = false;
+ for (MetadataUpdate update : request.updates()) {
+ if (update instanceof MetadataUpdate.AddPartitionSpec) {
+ for (Object field : fields(((MetadataUpdate.AddPartitionSpec)
update).spec())) {
+ missingIds |= partitionId(field) == null;
+ }
+ }
+ }
+ if (!missingIds) {
+ return request;
+ }
+ TableMetadata base = current.get();
+ if (base == null || base.formatVersion() < 2) {
+ return request;
+ }
+
+ Map<String, Integer> knownIds = new HashMap<>();
+ for (PartitionSpec spec : base.specs()) {
+ for (PartitionField field : spec.fields()) {
+ knownIds.putIfAbsent(
+ partitionFieldKey(field.sourceId(), field.transform().toString()),
field.fieldId());
+ }
+ }
+ int lastAssigned = Math.max(base.lastAssignedPartitionId(),
PARTITION_DATA_ID_START - 1);
+ for (MetadataUpdate update : request.updates()) {
+ if (update instanceof MetadataUpdate.AddPartitionSpec) {
+ for (Object field : fields(((MetadataUpdate.AddPartitionSpec)
update).spec())) {
+ Integer id = partitionId(field);
+ if (id != null) {
+ knownIds.putIfAbsent(partitionFieldKey(field), id);
+ lastAssigned = Math.max(lastAssigned, id);
+ }
+ }
+ }
+ }
+
+ List<MetadataUpdate> updates = new ArrayList<>();
+ for (MetadataUpdate update : request.updates()) {
+ if (!(update instanceof MetadataUpdate.AddPartitionSpec)) {
+ updates.add(update);
+ continue;
+ }
+ UnboundPartitionSpec spec = ((MetadataUpdate.AddPartitionSpec)
update).spec();
+ ObjectNode specNode = JsonUtil.mapper().createObjectNode();
+ specNode.put("spec-id", spec.specId());
+ ArrayNode fields = specNode.putArray("fields");
+ for (Object field : fields(spec)) {
+ Integer id = partitionId(field);
+ if (id == null) {
+ String key = partitionFieldKey(field);
+ id = knownIds.get(key);
+ if (id == null) {
+ id = ++lastAssigned;
+ knownIds.put(key, id);
+ }
+ }
+ fields
+ .addObject()
+ .put("name", (String) fieldValue(field, "name"))
+ .put("transform", (String) fieldValue(field, "transformAsString"))
+ .put("source-id", (Integer) fieldValue(field, "sourceId"))
+ .put("field-id", id);
+ }
+ updates.add(new
MetadataUpdate.AddPartitionSpec(PartitionSpecParser.fromJson(specNode)));
+ }
+ return request.identifier() == null
+ ? new UpdateTableRequest(request.requirements(), updates)
+ : UpdateTableRequest.create(request.identifier(),
request.requirements(), updates);
+ }
+
+ private static List<?> fields(UnboundPartitionSpec spec) {
+ return spec.fields();
+ }
+
+ @Nullable
+ private static Integer partitionId(Object unboundField) {
+ return (Integer) fieldValue(unboundField, "partitionId");
+ }
+
+ private static String partitionFieldKey(Object unboundField) {
+ return partitionFieldKey(
+ (Integer) fieldValue(unboundField, "sourceId"),
+ (String) fieldValue(unboundField, "transformAsString"));
+ }
+
+ /**
+ * Reads an accessor of {@code UnboundPartitionSpec.UnboundPartitionField}.
Its accessors are
+ * public but the class is package-private in Iceberg 1.11, and no public
API exposes whether a
+ * field was sent without a {@code field-id}.
+ */
+ private static Object fieldValue(Object unboundField, String accessor) {
+ try {
+ Method method = unboundField.getClass().getMethod(accessor);
+ method.setAccessible(true);
+ return method.invoke(unboundField);
+ } catch (ReflectiveOperationException e) {
+ throw new IllegalStateException("Cannot read partition field " +
accessor, e);
+ }
+ }
+
+ private static String partitionFieldKey(int sourceId, String transform) {
+ return sourceId + ":" + transform;
+ }
+
@Override
public void validate(@Nullable TableMetadata base) {
TableMetadata updated = validate(base, updates);
@@ -253,7 +385,7 @@ public final class IcebergCommitValidation implements
UpdateRequirement {
if (existing == null) {
checkNewField(base, field, hasData && !replacesTable);
} else if (hasData && !replacesTable) {
- checkTypeChange(existing, field);
+ checkTypeChange(existing, field, updated.formatVersion());
}
}
}
@@ -282,6 +414,17 @@ public final class IcebergCommitValidation implements
UpdateRequirement {
}
}
+ /**
+ * The v3 table spec also allows date to timestamp and date to timestamp_ns
(without time zone),
+ * which Iceberg's {@link TypeUtil#isPromotionAllowed} does not implement.
+ */
+ private static boolean isV3DatePromotion(Type from, Type to, int
formatVersion) {
+ return formatVersion >= 3
+ && from.typeId() == Type.TypeID.DATE
+ && (Types.TimestampType.withoutZone().equals(to)
+ || Types.TimestampNanoType.withoutZone().equals(to));
+ }
+
private static Types.NestedField previousField(TableMetadata base, int
fieldId) {
for (Schema schema : base.schemas()) {
Types.NestedField field = schema.findField(fieldId);
@@ -292,14 +435,16 @@ public final class IcebergCommitValidation implements
UpdateRequirement {
return null;
}
- private static void checkTypeChange(Types.NestedField existing,
Types.NestedField field) {
+ private static void checkTypeChange(
+ Types.NestedField existing, Types.NestedField field, int formatVersion) {
Type from = existing.type();
Type to = field.type();
if (from.equals(to)) {
return;
}
if (from.isPrimitiveType() && to.isPrimitiveType()) {
- if (!TypeUtil.isPromotionAllowed(from, to.asPrimitiveType())) {
+ if (!TypeUtil.isPromotionAllowed(from, to.asPrimitiveType())
+ && !isV3DatePromotion(from, to, formatVersion)) {
throw new ValidationException(
"Invalid schema: field %s cannot change type from %s to %s",
field.name(), from, to);
}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergColumnFieldValidator.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergColumnFieldValidator.java
index 5c4d78a4ef..70206c9f21 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergColumnFieldValidator.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergColumnFieldValidator.java
@@ -21,9 +21,13 @@ package org.apache.gravitino.iceberg.service.dispatcher;
import static org.apache.gravitino.Entity.EntityType.COLUMN;
import org.apache.gravitino.meta.ColumnEntity;
+import org.apache.gravitino.rel.types.Types;
import org.apache.iceberg.MetadataUpdate;
import org.apache.iceberg.Schema;
import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types.DecimalType;
+import org.apache.iceberg.types.Types.NestedField;
final class IcebergColumnFieldValidator {
@@ -35,6 +39,27 @@ final class IcebergColumnFieldValidator {
ColumnEntity.NAME.validate(column.name(), COLUMN);
ColumnEntity.COMMENT.validate(column.doc(), COLUMN);
});
+ TypeUtil.indexById(schema.asStruct())
+ .values()
+ .forEach(IcebergColumnFieldValidator::validateType);
+ }
+
+ /**
+ * Refuses a column type that Iceberg accepts but Gravitino cannot
represent, before the table is
+ * created. Otherwise the table is created and then fails to import into
Gravitino.
+ */
+ private static void validateType(NestedField field) {
+ if (field.type() instanceof DecimalType) {
+ DecimalType decimal = (DecimalType) field.type();
+ try {
+ Types.DecimalType.of(decimal.precision(), decimal.scale());
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Invalid type %s for column %s: %s", decimal, field.name(),
e.getMessage()),
+ e);
+ }
+ }
}
static void validateUpdate(UpdateTableRequest request) {
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCommitValidation.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCommitValidation.java
index 238941d98f..398e9ce679 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCommitValidation.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCommitValidation.java
@@ -23,6 +23,7 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import java.util.List;
import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.MetadataUpdate;
import org.apache.iceberg.MetadataUpdateParser;
import org.apache.iceberg.PartitionSpec;
@@ -316,6 +317,39 @@ public class TestIcebergCommitValidation {
Assertions.assertEquals(2, catalog.loadTable(IDENT).schemas().size());
}
+ @Test
+ public void testDatePromotionsAreAllowedOnlyFromV3() {
+ // The v3 table spec allows date to timestamp and timestamp_ns; v2 does
not.
+ datePromotion("3", "timestamp", true);
+ datePromotion("3", "timestamp_ns", true);
+ datePromotion("2", "timestamp", false);
+ // A promotion to a type with a time zone is never allowed.
+ datePromotion("3", "timestamptz", false);
+ }
+
+ private void datePromotion(String version, String type, boolean allowed) {
+ TableIdentifier ident = TableIdentifier.of(Namespace.of("db"), "d_v" +
version + "_" + type);
+ append(
+ catalog.createTable(
+ ident,
+ new Schema(Types.NestedField.optional(1, "d",
Types.DateType.get())),
+ PartitionSpec.unpartitioned(),
+ ImmutableMap.of("format-version", version)),
+ "/data/" + ident.name() + ".parquet");
+ UpdateTableRequest request =
+ request(
+
addSchema("{\"id\":1,\"name\":\"d\",\"required\":false,\"type\":\"" + type +
"\"}"));
+ if (allowed) {
+ CatalogHandlers.updateTable(catalog, ident, request);
+ Assertions.assertEquals(2, catalog.loadTable(ident).schemas().size(),
type);
+ } else {
+ Assertions.assertThrows(
+ ValidationException.class,
+ () -> CatalogHandlers.updateTable(catalog, ident, request),
+ version + " " + type);
+ }
+ }
+
@Test
public void testReusingADroppedFieldIdIsRefused() {
// Schema 1 adds b (id 2); schema 2 drops it; a later schema may not give
id 2 to another field.
@@ -380,6 +414,43 @@ public class TestIcebergCommitValidation {
Assertions.assertNotNull(partitioned);
}
+ @Test
+ public void testAddSpecWithoutFieldIdsGetsTheTablesIds() {
+ createPartitionedTable("noid");
+ TableIdentifier ident = TableIdentifier.of(Namespace.of("db"), "noid");
+ // Spec 0 has 1000: s identity(2). Neither field below carries a field-id.
+ UpdateTableRequest request =
+ IcebergCommitValidation.assignPartitionFieldIds(
+ request(
+ addSpec(
+
"{\"name\":\"s\",\"transform\":\"identity\",\"source-id\":2},"
+ +
"{\"name\":\"id_b\",\"transform\":\"bucket[4]\",\"source-id\":1}"),
+ new MetadataUpdate.SetDefaultPartitionSpec(-1)),
+ () -> ((HasTableOperations)
catalog.loadTable(ident)).operations().current());
+
+ CatalogHandlers.updateTable(catalog, ident,
IcebergCommitValidation.withValidation(request));
+
+ PartitionSpec spec = catalog.loadTable(ident).spec();
+ // The field already partitioned on keeps its id; the new field gets the
next one.
+ Assertions.assertEquals(1000, spec.fields().get(0).fieldId());
+ Assertions.assertEquals(1001, spec.fields().get(1).fieldId());
+ }
+
+ @Test
+ public void testAddSpecWithFieldIdsIsUnchanged() {
+ UpdateTableRequest request =
+ request(
+ addSpec(
+
"{\"name\":\"id\",\"transform\":\"identity\",\"source-id\":1,\"field-id\":1000}"));
+ Assertions.assertSame(
+ request,
+ IcebergCommitValidation.assignPartitionFieldIds(
+ request,
+ () -> {
+ throw new AssertionError("metadata must not be loaded");
+ }));
+ }
+
@Test
public void testRemovingASpecWithLiveManifestsIsRefused() {
Table partitioned = createPartitionedTable("rm");
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergTableOperationExecutor.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergTableOperationExecutor.java
index 1e7f0f2a70..cbfd05dde4 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergTableOperationExecutor.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergTableOperationExecutor.java
@@ -44,6 +44,7 @@ import org.apache.iceberg.rest.requests.CreateTableRequest;
import org.apache.iceberg.rest.requests.RenameTableRequest;
import org.apache.iceberg.rest.requests.UpdateTableRequest;
import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.iceberg.types.Types;
import org.apache.iceberg.types.Types.NestedField;
import org.apache.iceberg.types.Types.StringType;
import org.junit.jupiter.api.Assertions;
@@ -269,6 +270,36 @@ public class TestIcebergTableOperationExecutor {
verifyNoInteractions(mockCatalogWrapper);
}
+ @Test
+ public void testRejectsUnrepresentableDecimalBeforeCreate() {
+ // Iceberg accepts decimal(10, 11); Gravitino cannot import it, so it must
be refused before the
+ // table is created rather than after.
+ Schema schema =
+ new Schema(
+ NestedField.required(1, "id", StringType.get()),
+ NestedField.optional(
+ 2,
+ "s",
+ Types.StructType.of(
+ NestedField.optional(3, "amount", Types.DecimalType.of(10,
11)))));
+ CreateTableRequest request =
+
CreateTableRequest.builder().withName("test_table").withSchema(schema).build();
+
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> executor.createTable(mockContext,
Namespace.of("test_namespace"), request));
+ Assertions.assertTrue(exception.getMessage().contains("amount"),
exception.getMessage());
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ executor.updateTable(
+ mockContext,
+ TableIdentifier.of("test_namespace", "test_table"),
+ updateRequest(schema)));
+ verifyNoInteractions(mockCatalogWrapper);
+ }
+
@Test
public void testRejectsOversizedColumnFieldsBeforeUpdate() {
String oversizedName = "a".repeat(EntityFieldLimits.MAX_NAME_LENGTH + 1);
