This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new b94de1e1e1 [Cherry-pick to branch-1.3] [#14545] fix(iceberg-rest):
assign missing partition field ids and refuse decimals Gravitino cannot import
(#14547)
b94de1e1e1 is described below
commit b94de1e1e1d04a0b46f1feba935ea69d459052c4
Author: Mark Hoerth <[email protected]>
AuthorDate: Mon Nov 2 22:07:47 2026 -0800
[Cherry-pick to branch-1.3] [#14545] fix(iceberg-rest): assign missing
partition field ids and refuse decimals Gravitino cannot import (#14547)
Backport to branch-1.3 of the fix for #14545, without the v3 date to
timestamp and timestamp_ns promotions, which are 2.0 only.
### 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 | 132 +++++++++++++++++++++
.../dispatcher/IcebergColumnFieldValidator.java | 25 ++++
.../service/TestIcebergCommitValidation.java | 38 ++++++
.../TestIcebergTableOperationExecutor.java | 31 +++++
5 files changed, 240 insertions(+), 1 deletion(-)
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 b9f7be6fa3..723ab46a45 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;
@@ -51,6 +52,7 @@ import org.apache.gravitino.utils.MapUtils;
import org.apache.gravitino.utils.PrincipalUtils;
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;
@@ -166,10 +168,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..6d72c96d20 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);
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..b53c0b12bc 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;
@@ -380,6 +381,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 a8c7f7b230..043121f2c3 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);
