This is an automated email from the ASF dual-hosted git repository.
suxiaogang223 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 49f3cf87d8f [fix](iceberg) Preserve assigned field IDs for older REST
catalogs (#69757)
49f3cf87d8f is described below
commit 49f3cf87d8f4667555f3999e3de28ebb1e05f8ce
Author: Socrates <[email protected]>
AuthorDate: Thu Oct 29 10:22:07 2026 +0800
[fix](iceberg) Preserve assigned field IDs for older REST catalogs (#69757)
### What problem does this PR solve?
Issue Number: N/A
Related issue: https://github.com/apache/iceberg/issues/13850
Problem Summary:
With Iceberg SDK 1.9+ (including master's 1.12.0), dropping the
highest-ID column sends the surviving schema's maximum field ID as
`add-schema.last-column-id`. Iceberg REST 1.6.0 validates this
deprecated field against the table's historical assigned-ID high-water
mark, so a table with IDs 1, 2, 3 rejects dropping ID 3 with `Invalid
last column ID: 2 < 3`. Dropping ID 2 from a two-column table fails in
the same way.
Configure the standard REST HTTP client with SDK-compatible Jackson
codecs that preserve the base assigned-ID requirement in this legacy
wire field. Track the maximum across successive schema updates,
independently for each table in a multi-table commit. Keep field/schema
IDs, commit requirements, authentication, retry handling, and existing
DROP COLUMN partition-spec fencing unchanged. Non-schema commits use the
upstream request parser directly.
### Release note
Fix dropping the highest-ID Iceberg column through older REST catalogs,
including Iceberg REST 1.6.0.
### Check List (For Author)
- Test:
- [ ] Unit Test: 10 added covering both drops, historical IDs, repeated
schema updates, create/replace, multi-table isolation, unchanged
payloads, concurrent schema changes, and existing response codecs. Not
run locally.
- [ ] Regression test: extend the REST 1.6.0 HDFS fixture suite with
both drops, repeated drops, add/drop, and subsequent inserts. Not run
locally.
- Validation performed: repository Checkstyle 9.3 on all three changed
Java files; `git diff --check`. Compilation and test execution are left
to CI as requested; no local build dependencies or artifacts were
created.
- Behavior changed: Yes. Legacy REST servers receive a nondecreasing
historical `last-column-id` for schema commits.
- Does this need documentation: No.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../doris/connector/iceberg/IcebergConnector.java | 8 +-
.../iceberg/IcebergRestSchemaCompatibility.java | 115 ++++++++++++++
.../IcebergRestSchemaCompatibilityTest.java | 169 +++++++++++++++++++++
.../iceberg_rest_on_hdfs.groovy | 18 +++
4 files changed, 309 insertions(+), 1 deletion(-)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java
index 12c88ced2f8..3a9dd213dcf 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java
@@ -65,8 +65,10 @@ import org.apache.iceberg.hive.HiveCatalog;
import org.apache.iceberg.hive.HiveHadoopUtil;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.SupportsStorageCredentials;
+import org.apache.iceberg.rest.HTTPClient;
import org.apache.iceberg.rest.RESTCatalog;
import org.apache.iceberg.rest.RESTSessionCatalog;
+import org.apache.iceberg.rest.RESTUtil;
import org.apache.iceberg.util.ThreadPools;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -1209,7 +1211,11 @@ public class IcebergConnector implements Connector {
/** Builds and initializes a bare Iceberg {@link RESTSessionCatalog}
(fresh REST client + OAuth2 token fetch). */
private RESTSessionCatalog newRestSessionCatalog(String catalogName,
Map<String, String> props,
Configuration conf) {
- RESTSessionCatalog sessionCatalog = new RESTSessionCatalog();
+ RESTSessionCatalog sessionCatalog = new RESTSessionCatalog(config ->
HTTPClient.builder(config)
+ .uri(config.get(CatalogProperties.URI))
+ .withHeaders(RESTUtil.configHeaders(config))
+
.withObjectMapper(IcebergRestSchemaCompatibility.objectMapper())
+ .build(), null);
CatalogUtil.configureHadoopConf(sessionCatalog, conf);
sessionCatalog.initialize(catalogName, props);
return sessionCatalog;
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergRestSchemaCompatibility.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergRestSchemaCompatibility.java
new file mode 100644
index 00000000000..bf281b9c6ab
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergRestSchemaCompatibility.java
@@ -0,0 +1,115 @@
+// 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.doris.connector.iceberg;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.PropertyAccessor;
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.core.JsonFactoryBuilder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.PropertyNamingStrategies;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.iceberg.MetadataUpdate;
+import org.apache.iceberg.UpdateRequirement;
+import org.apache.iceberg.rest.RESTSerializers;
+import org.apache.iceberg.rest.requests.CommitTransactionRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequestParser;
+import org.apache.iceberg.util.JsonUtil;
+
+import java.io.IOException;
+import java.util.List;
+
+/** Preserves the assigned field ID high-water mark for Iceberg 1.6 REST
servers. */
+final class IcebergRestSchemaCompatibility {
+ private IcebergRestSchemaCompatibility() {
+ }
+
+ static ObjectMapper objectMapper() {
+ // Match the SDK's package-private RESTObjectMapper, including all
request/response codecs.
+ ObjectMapper mapper = new ObjectMapper(new JsonFactoryBuilder()
+ .configure(JsonFactory.Feature.INTERN_FIELD_NAMES, false)
+ .configure(JsonFactory.Feature.FAIL_ON_SYMBOL_HASH_OVERFLOW,
false).build());
+ mapper.setVisibility(PropertyAccessor.FIELD,
JsonAutoDetect.Visibility.ANY);
+ mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,
false);
+ mapper.setPropertyNamingStrategy(new
PropertyNamingStrategies.KebabCaseStrategy());
+ RESTSerializers.registerAll(mapper);
+ SimpleModule module = new
SimpleModule("doris-iceberg-rest-schema-compatibility");
+ module.addSerializer(UpdateTableRequest.class, new
UpdateTableSerializer());
+ // The SDK transaction serializer invokes UpdateTableRequestParser
directly, bypassing Jackson codecs.
+ module.addSerializer(CommitTransactionRequest.class, new
CommitTransactionSerializer());
+ mapper.registerModule(module);
+ return mapper;
+ }
+
+ private static void writeUpdate(UpdateTableRequest request, JsonGenerator
generator) throws IOException {
+ List<MetadataUpdate> updates = request.updates();
+ boolean addsSchema = updates.stream().anyMatch(update -> update
instanceof MetadataUpdate.AddSchema);
+ if (!addsSchema) {
+ UpdateTableRequestParser.toJson(request, generator);
+ return;
+ }
+
+ int lastColumnId = 0;
+ for (UpdateRequirement requirement : request.requirements()) {
+ if (requirement instanceof
UpdateRequirement.AssertLastAssignedFieldId) {
+ lastColumnId = ((UpdateRequirement.AssertLastAssignedFieldId)
requirement).lastAssignedFieldId();
+ }
+ }
+ JsonNode json =
JsonUtil.mapper().readTree(UpdateTableRequestParser.toJson(request));
+ for (int index = 0; index < updates.size(); index++) {
+ MetadataUpdate update = updates.get(index);
+ if (update instanceof MetadataUpdate.AddSchema) {
+ // SDK 1.9+ writes schema.highestFieldId(), which can decrease
after DROP COLUMN.
+ // REST 1.6 validates this deprecated field against the
table's historical last-column-id.
+ // Keep it monotonic across every schema change in a single
commit as well.
+ lastColumnId = Math.max(lastColumnId,
((MetadataUpdate.AddSchema) update).schema().highestFieldId());
+ ((ObjectNode)
json.get("updates").get(index)).put("last-column-id", lastColumnId);
+ }
+ }
+ generator.writeTree(json);
+ }
+
+ private static class UpdateTableSerializer extends
JsonSerializer<UpdateTableRequest> {
+ @Override
+ public void serialize(UpdateTableRequest request, JsonGenerator
generator, SerializerProvider provider)
+ throws IOException {
+ writeUpdate(request, generator);
+ }
+ }
+
+ private static class CommitTransactionSerializer extends
JsonSerializer<CommitTransactionRequest> {
+ @Override
+ public void serialize(CommitTransactionRequest request, JsonGenerator
generator, SerializerProvider provider)
+ throws IOException {
+ generator.writeStartObject();
+ generator.writeArrayFieldStart("table-changes");
+ for (UpdateTableRequest tableChange : request.tableChanges()) {
+ writeUpdate(tableChange, generator);
+ }
+ generator.writeEndArray();
+ generator.writeEndObject();
+ }
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergRestSchemaCompatibilityTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergRestSchemaCompatibilityTest.java
new file mode 100644
index 00000000000..58b9222498d
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergRestSchemaCompatibilityTest.java
@@ -0,0 +1,169 @@
+// 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.doris.connector.iceberg;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.iceberg.MetadataUpdate;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.UpdateRequirement;
+import org.apache.iceberg.UpdateRequirements;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.rest.requests.CommitTransactionRequest;
+import org.apache.iceberg.rest.requests.CommitTransactionRequestParser;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequestParser;
+import org.apache.iceberg.rest.responses.ConfigResponse;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+class IcebergRestSchemaCompatibilityTest {
+ private final ObjectMapper mapper =
IcebergRestSchemaCompatibility.objectMapper();
+
+ @Test
+ void testDropHighestOfThreeFields() throws Exception {
+ assertDropHighestField(3);
+ }
+
+ @Test
+ void testDropHighestOfTwoFields() throws Exception {
+ assertDropHighestField(2);
+ }
+
+ private void assertDropHighestField(int count) throws Exception {
+ TableMetadata base = metadata(count);
+ UpdateTableRequest request = update(base, schema(count - 1));
+ String original = UpdateTableRequestParser.toJson(request);
+ JsonNode expected = mapper.readTree(original);
+ // This is the only wire change: preserve the historical high-water
mark, not the surviving maximum ID.
+ ((ObjectNode) expected.get("updates").get(0)).put("last-column-id",
count);
+ JsonNode actual = wire(request);
+ Assertions.assertEquals(expected, actual);
+ Assertions.assertEquals(original,
UpdateTableRequestParser.toJson(request));
+ Assertions.assertEquals(count - 1,
actual.at("/updates/0/schema/fields").size());
+ for (int index = 0; index < count - 1; index++) {
+ Assertions.assertEquals(index + 1,
actual.at("/updates/0/schema/fields/" + index + "/id").asInt());
+ }
+ }
+
+ @Test
+ void testRetainHistoricalIdAfterEarlierDrop() throws Exception {
+ TableMetadata base =
TableMetadata.buildFrom(metadata(3)).addSchema(schema(2)).build();
+ Assertions.assertEquals(3, base.lastColumnId());
+ Assertions.assertEquals(3, wire(update(base,
schema(1))).at("/updates/0/last-column-id").asInt());
+ }
+
+ @Test
+ void testAddThenDropWithinOneCommitDoesNotLowerId() throws Exception {
+ JsonNode json = wire(update(metadata(3), schema(4), schema(2)));
+ Assertions.assertEquals(4,
json.at("/updates/0/last-column-id").asInt());
+ Assertions.assertEquals(4,
json.at("/updates/1/last-column-id").asInt());
+ }
+
+ @Test
+ void testCreateTracksIdsAcrossSchemaUpdates() throws Exception {
+ List<MetadataUpdate> updates = Arrays.asList(new
MetadataUpdate.AddSchema(schema(4)),
+ new MetadataUpdate.AddSchema(schema(2)));
+ JsonNode json = wire(new
UpdateTableRequest(UpdateRequirements.forCreateTable(updates), updates));
+ Assertions.assertEquals(4,
json.at("/updates/0/last-column-id").asInt());
+ Assertions.assertEquals(4,
json.at("/updates/1/last-column-id").asInt());
+ }
+
+ @Test
+ void testReplaceRetainsBaseAssignedId() throws Exception {
+ TableMetadata base = metadata(3);
+ List<MetadataUpdate> updates = Collections.singletonList(new
MetadataUpdate.AddSchema(schema(2)));
+ UpdateTableRequest request = new
UpdateTableRequest(UpdateRequirements.forReplaceTable(base, updates), updates);
+ Assertions.assertEquals(3,
wire(request).at("/updates/0/last-column-id").asInt());
+ }
+
+ @Test
+ void testMultiTableCommitKeepsIndependentHighWaterMarks() throws Exception
{
+ UpdateTableRequest first = update(metadata(3), schema(4), schema(2));
+ UpdateTableRequest second = update(metadata(2), schema(1));
+ second = UpdateTableRequest.create(TableIdentifier.of("db", "other"),
second.requirements(), second.updates());
+ CommitTransactionRequest request = new
CommitTransactionRequest(Arrays.asList(first, second));
+ JsonNode expected =
mapper.readTree(CommitTransactionRequestParser.toJson(request));
+ ((ObjectNode)
expected.at("/table-changes/0/updates/0")).put("last-column-id", 4);
+ ((ObjectNode)
expected.at("/table-changes/0/updates/1")).put("last-column-id", 4);
+ ((ObjectNode)
expected.at("/table-changes/1/updates/0")).put("last-column-id", 2);
+ Assertions.assertEquals(expected,
mapper.readTree(mapper.writeValueAsString(request)));
+ }
+
+ @Test
+ void testNonSchemaCommitIsUnchanged() throws Exception {
+ UpdateTableRequest request =
UpdateTableRequest.create(TableIdentifier.of("db", "table"),
+ Collections.singletonList(new
UpdateRequirement.AssertTableUUID(metadata(3).uuid())),
+ Collections.singletonList(new
MetadataUpdate.SetProperties(Collections.singletonMap("key", "value"))));
+
Assertions.assertEquals(mapper.readTree(UpdateTableRequestParser.toJson(request)),
wire(request));
+ }
+
+ @Test
+ void testConcurrentSchemaChangeStillFailsRequirements() throws Exception {
+ UpdateTableRequest request = update(metadata(3), schema(2));
+ UpdateTableRequest decoded =
UpdateTableRequestParser.fromJson(mapper.writeValueAsString(request));
+ TableMetadata concurrent =
TableMetadata.buildFrom(metadata(3)).addSchema(schema(4)).build();
+ UpdateRequirement fieldIdRequirement = decoded.requirements().stream()
+ .filter(requirement -> requirement instanceof
UpdateRequirement.AssertLastAssignedFieldId)
+ .findFirst().orElseThrow(AssertionError::new);
+ Assertions.assertThrows(CommitFailedException.class, () ->
fieldIdRequirement.validate(concurrent));
+ }
+
+ @Test
+ void testOtherRestResponsesKeepSdkCodecsAndUnknownFieldTolerance() throws
Exception {
+ ConfigResponse config =
mapper.readValue("{\"defaults\":{\"key\":\"value\"},"
+ + "\"overrides\":{},\"future-field\":true}",
ConfigResponse.class);
+ Assertions.assertEquals("value", config.defaults().get("key"));
+ }
+
+ private JsonNode wire(UpdateTableRequest request) throws Exception {
+ return mapper.readTree(mapper.writeValueAsString(request));
+ }
+
+ private static UpdateTableRequest update(TableMetadata base, Schema...
schemas) {
+ List<MetadataUpdate> updates = new ArrayList<>();
+ for (Schema schema : schemas) {
+ updates.add(new MetadataUpdate.AddSchema(schema));
+ }
+ return UpdateTableRequest.create(TableIdentifier.of("db", "table"),
+ UpdateRequirements.forUpdateTable(base, updates), updates);
+ }
+
+ private static TableMetadata metadata(int count) {
+ return TableMetadata.newTableMetadata(schema(count),
PartitionSpec.unpartitioned(), "s3://bucket/table",
+ Collections.emptyMap());
+ }
+
+ private static Schema schema(int count) {
+ List<Types.NestedField> fields = new ArrayList<>();
+ for (int id = 1; id <= count; id++) {
+ fields.add(Types.NestedField.optional(id, "c" + id,
Types.IntegerType.get()));
+ }
+ return new Schema(fields);
+ }
+}
diff --git
a/regression-test/suites/external_table_p2/refactor_catalog_param/iceberg_rest_on_hdfs.groovy
b/regression-test/suites/external_table_p2/refactor_catalog_param/iceberg_rest_on_hdfs.groovy
index abb02093cbd..e78f52a5851 100644
---
a/regression-test/suites/external_table_p2/refactor_catalog_param/iceberg_rest_on_hdfs.groovy
+++
b/regression-test/suites/external_table_p2/refactor_catalog_param/iceberg_rest_on_hdfs.groovy
@@ -49,6 +49,24 @@ suite("iceberg_rest_on_hdfs", "p2,external") {
sql """
use ${db_name};
"""
+ // The HDFS REST fixture runs tabulario/iceberg-rest:1.6.0. Dropping
the highest field ID must
+ // preserve the table's historical last-column-id in add-schema
commits to this older server.
+ sql """DROP TABLE IF EXISTS rest_drop_highest_three"""
+ sql """CREATE TABLE rest_drop_highest_three (c1 INT, c2 INT, c3 INT)"""
+ sql """INSERT INTO rest_drop_highest_three VALUES (1, 2, 3)"""
+ sql """ALTER TABLE rest_drop_highest_three DROP COLUMN c3"""
+ sql """INSERT INTO rest_drop_highest_three VALUES (4, 5)"""
+ sql """ALTER TABLE rest_drop_highest_three DROP COLUMN c2"""
+ sql """ALTER TABLE rest_drop_highest_three ADD COLUMN c4 INT"""
+ sql """ALTER TABLE rest_drop_highest_three DROP COLUMN c4"""
+ sql """INSERT INTO rest_drop_highest_three VALUES (6)"""
+
+ sql """DROP TABLE IF EXISTS rest_drop_highest_two"""
+ sql """CREATE TABLE rest_drop_highest_two (c1 INT, c2 INT)"""
+ sql """INSERT INTO rest_drop_highest_two VALUES (1, 2)"""
+ sql """ALTER TABLE rest_drop_highest_two DROP COLUMN c2"""
+ sql """INSERT INTO rest_drop_highest_two VALUES (3)"""
+
// ======= BASIC TABLE OPERATIONS TEST =======
def table_name = prefix + "_table"
sql """
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
