This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 0023c3512f [iceberg] Fix schema drift crash in
IcebergRestMetadataCommitter (#8335)
0023c3512f is described below
commit 0023c3512f29c743d2a554e2f3d92cee8e6426f5
Author: Oleksandr Nitavskyi <[email protected]>
AuthorDate: Wed Jun 24 03:23:28 2026 +0200
[iceberg] Fix schema drift crash in IcebergRestMetadataCommitter (#8335)
Option-only alterTable calls (e.g. changing metadata.iceberg.*
properties) increment Paimon's schema ID without modifying columns,
producing multiple IcebergSchema objects with identical fields but
different IDs.
Root cause: TableMetadata.Builder.addSchema() uses sameSchema() to
deduplicate identical schemas internally, collapsing them to the first
ID. When setCurrentSchema(N) is called with the higher (non-surviving)
ID, it fails:
"Cannot set current schema to unknown schema: N"
---
.../iceberg/IcebergRestMetadataCommitter.java | 223 +++++++++++++++++----
.../iceberg/IcebergRestMetadataCommitterTest.java | 183 +++++++++++++++++
2 files changed, 364 insertions(+), 42 deletions(-)
diff --git
a/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
b/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
index f144ea96de..820fb2ff94 100644
---
a/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
+++
b/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
@@ -50,8 +50,10 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -155,20 +157,33 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
icebergTable = getTable();
TableMetadata metadata = ((BaseTable)
icebergTable).operations().current();
- boolean withBase = checkBase(metadata, newMetadata,
baseIcebergMetadata);
- if (withBase) {
- LOG.info("create updates with base metadata.");
- updateBuilder = updatesForCorrectBase(metadata,
newMetadata, false);
- } else {
+
+ if (metadata.currentSnapshot() == null) {
+ // Table exists in the REST catalog but has no snapshots
yet. This happens
+ // when a previous createTable() or recreateTable()
succeeded but the
+ // subsequent commit() failed (e.g. network error, REST
server timeout).
+ // Treat it as a new table — populate schemas, partition
spec, and the
+ // current snapshot from scratch WITHOUT dropping and
recreating the table.
+ // Routing through updatesForIncorrectBase would call
recreateTable(), which
+ // on a repeated commit failure would create an infinite
drop+create loop.
LOG.info(
- "create updates without base metadata.
currentSnapshotId for base metadata: {}, for new metadata:{}",
- metadata.currentSnapshot() != null
- ? metadata.currentSnapshot().snapshotId()
- : "No snapshot",
- newMetadata.currentSnapshot() != null
- ?
newMetadata.currentSnapshot().snapshotId()
- : "No snapshot");
- updateBuilder = updatesForIncorrectBase(newMetadata);
+ "Iceberg table {} exists but has no snapshots,
treating as new table.",
+ icebergTableIdentifier);
+ updateBuilder = updatesForCorrectBase(metadata,
newMetadata, true);
+ } else {
+ boolean withBase = checkBase(metadata, newMetadata,
baseIcebergMetadata);
+ if (withBase) {
+ LOG.info("create updates with base metadata.");
+ updateBuilder = updatesForCorrectBase(metadata,
newMetadata, false);
+ } else {
+ LOG.info(
+ "create updates without base metadata.
currentSnapshotId for base metadata: {}, for new metadata:{}",
+ metadata.currentSnapshot().snapshotId(),
+ newMetadata.currentSnapshot() != null
+ ?
newMetadata.currentSnapshot().snapshotId()
+ : "No snapshot");
+ updateBuilder = updatesForIncorrectBase(newMetadata);
+ }
}
}
} catch (Exception e) {
@@ -243,6 +258,7 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
removeSnapshots(snapshotIdsToRemove, updateBuilder);
}
+ updateProperties(updateBuilder);
return updateBuilder;
}
@@ -398,16 +414,27 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
update.addSchema(schema);
}
update.setCurrentSchema(currentSchemaId);
+ }
- // update properties
- Map<String, String> properties = new HashMap<>();
- properties.put(
- METADATA_PREVIOUS_VERSIONS_MAX,
- String.valueOf(icebergOptions.previousVersionsMax()));
- properties.put(
- METADATA_DELETE_AFTER_COMMIT_ENABLED,
- String.valueOf(icebergOptions.deleteAfterCommitEnabled()));
- update.setProperties(properties);
+ // Update Iceberg REST table properties from current IcebergOptions, but
only
+ // if the values differ from what the REST catalog already has. This avoids
+ // emitting a redundant SetProperties update on every commit.
+ private void updateProperties(TableMetadata.Builder update) {
+ String desiredMax =
String.valueOf(icebergOptions.previousVersionsMax());
+ String desiredDeleteAfter =
String.valueOf(icebergOptions.deleteAfterCommitEnabled());
+
+ Map<String, String> current = icebergTable.properties();
+ boolean changed =
+ !desiredMax.equals(current.get(METADATA_PREVIOUS_VERSIONS_MAX))
+ || !desiredDeleteAfter.equals(
+
current.get(METADATA_DELETE_AFTER_COMMIT_ENABLED));
+
+ if (changed) {
+ Map<String, String> properties = new HashMap<>();
+ properties.put(METADATA_PREVIOUS_VERSIONS_MAX, desiredMax);
+ properties.put(METADATA_DELETE_AFTER_COMMIT_ENABLED,
desiredDeleteAfter);
+ update.setProperties(properties);
+ }
}
//
-------------------------------------------------------------------------------------
@@ -432,37 +459,149 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
}
// if the iceberg table is existed, check whether the current metadata
of the table is the
- // base of the new table metadata, we use current snapshot id to check
+ // base of the new table metadata, we use current snapshot id to check.
+ // Note: callers must ensure currentMetadata.currentSnapshot() is
non-null before calling
+ // this method (guarded in commitMetadataImpl).
return currentMetadata.currentSnapshot().snapshotId()
== newMetadata.currentSnapshot().snapshotId() - 1;
}
private IcebergMetadata adjustMetadataForRest(IcebergMetadata
newIcebergMetadata) {
- // why need this:
- // Since we will use an empty schema to create iceberg table in rest
catalog and id-0 will
- // be occupied by the empty schema, there will be 1-unit offset
between the schema-id in
- // metadata stored in rest catalog and the schema-id in paimon.
-
- List<IcebergSchema> schemas =
+ // --- Why we shift schema IDs by +1 ---
+ // When we create an Iceberg table via the REST catalog, we register
it with
+ // an empty schema that occupies schema ID 0. Paimon's own schema IDs
start
+ // at 0 as well, so we shift every Paimon schema ID by +1 to avoid
colliding
+ // with that placeholder.
+ //
+ // --- Why we deduplicate schemas ---
+ // Option-only alterTable calls in Paimon (e.g. changing table
properties
+ // like metadata.iceberg.* settings) increment Paimon's internal schema
+ // version without modifying the column definitions. This means Paimon
can
+ // accumulate multiple schema versions (e.g. IDs 0, 1, 2, 3) that all
have
+ // identical field lists.
+ //
+ // When these schemas are forwarded to Iceberg,
addAndSetCurrentSchema()
+ // calls TableMetadata.Builder.addSchema() for each one. Iceberg
internally
+ // deduplicates identical schemas via its sameSchema() check — it
keeps only
+ // the first occurrence and silently drops duplicates. So if schemas
1, 2, 3
+ // all have the same fields, Iceberg keeps only schema 1.
+ //
+ // By deduplicating here (using IcebergDataField.equals() which
compares id,
+ // name, required, type, and doc), we keep only unique schemas and
build a
+ // remap table so that currentSchemaId and snapshot schemaId
references point
+ // to the surviving schema's ID. The downstream metadata is then
internally
+ // consistent with what Iceberg catalog will actually store.
+
+ // Step 1 — Shift and convert: produce Iceberg Schema objects so we
can call
+ // sameSchema() in the dedup step.
+ List<IcebergSchema> shiftedIcebergSchemas =
newIcebergMetadata.schemas().stream()
- .map(schema -> new IcebergSchema(schema.schemaId() +
1, schema.fields()))
+ .map(s -> new IcebergSchema(s.schemaId() + 1,
s.fields()))
.collect(Collectors.toList());
- int currentSchemaId = newIcebergMetadata.currentSchemaId() + 1;
+ int shiftedCurrentSchemaId = newIcebergMetadata.currentSchemaId() + 1;
+ // Build a temporary IcebergMetadata with shifted IDs to obtain
Iceberg Schema objects.
+ IcebergMetadata shiftedForConversion =
+ new IcebergMetadata(
+ newIcebergMetadata.formatVersion(),
+ newIcebergMetadata.tableUuid(),
+ newIcebergMetadata.location(),
+ newIcebergMetadata.currentSnapshotId(),
+ newIcebergMetadata.lastColumnId(),
+ shiftedIcebergSchemas,
+ shiftedCurrentSchemaId,
+ newIcebergMetadata.partitionSpecs(),
+ newIcebergMetadata.lastPartitionId(),
+ newIcebergMetadata.snapshots().stream()
+ .map(
+ s ->
+ new IcebergSnapshot(
+ s.sequenceNumber(),
+ s.snapshotId(),
+ s.parentSnapshotId(),
+ s.timestampMs(),
+ s.summary(),
+ s.manifestList(),
+ s.schemaId() + 1,
+ s.firstRowId(),
+ s.addedRows()))
+ .collect(Collectors.toList()),
+ newIcebergMetadata.currentSnapshotId(),
+ newIcebergMetadata.refs());
+ TableMetadata shiftedTableMetadata =
+ TableMetadataParser.fromJson(shiftedForConversion.toJson());
+ Map<Integer, IcebergSchema> shiftedById =
+ shiftedIcebergSchemas.stream()
+ .collect(Collectors.toMap(IcebergSchema::schemaId, s
-> s));
+
+ // Step 2 — Deduplicate using sameSchema(): keeps first occurrence
(insertion order
+ // preserved by LinkedHashMap), remaps duplicates to the surviving
schema ID.
+ // Maps each shifted schema ID → the surviving (deduped) schema ID
(before renumbering).
+ LinkedHashMap<Integer, Schema> survivingById = new LinkedHashMap<>();
+ Map<Integer, Integer> schemaIdRemap = new HashMap<>();
+
+ for (Schema schema : shiftedTableMetadata.schemas()) {
+ int shiftedId = schema.schemaId();
+ Integer survivingId =
+ survivingById.keySet().stream()
+ .filter(sid ->
survivingById.get(sid).sameSchema(schema))
+ .findFirst()
+ .orElse(null);
+ if (survivingId != null) {
+ // Duplicate: remap this ID to the first schema with the same
structure
+ schemaIdRemap.put(shiftedId, survivingId);
+ } else {
+ // First occurrence: keep it
+ survivingById.put(shiftedId, schema);
+ schemaIdRemap.put(shiftedId, shiftedId);
+ }
+ }
+
+ // Step 2 — Renumber: after dedup, there may be gaps in the schema ID
sequence
+ // (e.g. [1, 2, 4] when schema 3 was deduped into 2). Iceberg's
addSchema() assigns
+ // IDs sequentially as max(existing) + 1, so it would produce [1, 2,
3] — not [1, 2, 4].
+ // Calling setCurrentSchema(4) would then fail with "unknown schema:
4".
+ // We renumber the surviving schemas to 1, 2, 3, ... and update the
remap table so
+ // that currentSchemaId and snapshot schemaId references stay
consistent.
+ Map<Integer, Integer> renumberMap = new HashMap<>(); // old deduped ID
→ new sequential ID
+ int nextId = 1;
+ List<IcebergSchema> schemas = new ArrayList<>();
+ for (int survivingId : survivingById.keySet()) {
+ renumberMap.put(survivingId, nextId);
+ schemas.add(new IcebergSchema(nextId,
shiftedById.get(survivingId).fields()));
+ nextId++;
+ }
+ // Apply renumbering on top of the dedup remap
+ schemaIdRemap.replaceAll((k, dedupedId) -> renumberMap.get(dedupedId));
+
+ // 3. Remap currentSchemaId through dedup + renumber
+ int currentSchemaId =
+ schemaIdRemap.getOrDefault(
+ newIcebergMetadata.currentSchemaId() + 1,
+ newIcebergMetadata.currentSchemaId() + 1);
+
+ // Remap snapshot schema references so they point to surviving schema
IDs
List<IcebergSnapshot> snapshots =
newIcebergMetadata.snapshots().stream()
.map(
- snapshot ->
- new IcebergSnapshot(
- snapshot.sequenceNumber(),
- snapshot.snapshotId(),
- snapshot.parentSnapshotId(),
- snapshot.timestampMs(),
- snapshot.summary(),
- snapshot.manifestList(),
- snapshot.schemaId() + 1,
- snapshot.firstRowId(),
- snapshot.addedRows()))
+ snapshot -> {
+ int shiftedSnapshotSchemaId =
snapshot.schemaId() + 1;
+ int remappedSchemaId =
+ schemaIdRemap.getOrDefault(
+ shiftedSnapshotSchemaId,
+ shiftedSnapshotSchemaId);
+ return new IcebergSnapshot(
+ snapshot.sequenceNumber(),
+ snapshot.snapshotId(),
+ snapshot.parentSnapshotId(),
+ snapshot.timestampMs(),
+ snapshot.summary(),
+ snapshot.manifestList(),
+ remappedSchemaId,
+ snapshot.firstRowId(),
+ snapshot.addedRows());
+ })
.collect(Collectors.toList());
+
return new IcebergMetadata(
newIcebergMetadata.formatVersion(),
newIcebergMetadata.tableUuid(),
diff --git
a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
index de39ea6f6d..a3ae135686 100644
---
a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
+++
b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
@@ -414,6 +414,189 @@ public class IcebergRestMetadataCommitterTest {
commit.close();
}
+ @Test
+ public void testOptionOnlyAlterTableDoesNotCrashIcebergSync() throws
Exception {
+ // The fix deduplicates schemas in adjustMetadataForRest() and remaps
+ // currentSchemaId + snapshot schemaId references to the surviving ID.
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new
String[] {"k", "v"});
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.singletonList("k"),
+ 1,
+ randomFormat(),
+ Collections.emptyMap());
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl<?> write = table.newWrite(commitUser);
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ // Initial write — establishes schema-0 in Paimon, schema-1 in Iceberg
+ write.write(GenericRow.of(1, 10));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(1, write.prepareCommit(false, 1));
+ assertThat(getIcebergResult()).containsExactlyInAnyOrder("Record(1,
10)", "Record(2, 20)");
+
+ // Perform 3 option-only schema changes — each increments Paimon
schema ID
+ // but does NOT change columns, creating the dedup scenario
+ SchemaManager schemaManager = new SchemaManager(table.fileIO(),
table.location());
+
schemaManager.commitChanges(SchemaChange.setOption("my.custom.option.1",
"value1"));
+
schemaManager.commitChanges(SchemaChange.setOption("my.custom.option.2",
"value2"));
+
schemaManager.commitChanges(SchemaChange.setOption("my.custom.option.3",
"value3"));
+ table = table.copy(table.schemaManager().latest().get());
+ write.close();
+ write = table.newWrite(commitUser);
+ commit.close();
+ commit = table.newCommit(commitUser);
+
+ // Write more data — this triggers IcebergRestMetadataCommitter with
+ // multiple schemas that have identical fields but different IDs.
+ // Without the fix, this crashes with "Cannot set current schema to
+ // unknown schema: N".
+ write.write(GenericRow.of(1, 11));
+ write.write(GenericRow.of(3, 30));
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ commit.commit(2, write.prepareCommit(true, 2));
+
+ // Verify data is readable through Iceberg
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder("Record(1, 11)", "Record(2, 20)",
"Record(3, 30)");
+
+ // Verify schema dedup happened: Iceberg should have fewer schemas than
+ // Paimon (Paimon has 4 schemas: original + 3 option changes; Iceberg
+ // should have just 1 unique schema after dedup, plus the empty
placeholder)
+ Table icebergTable = restCatalog.loadTable(TableIdentifier.of("mydb",
"t"));
+ int icebergSchemaCount = icebergTable.schemas().size();
+ int paimonSchemaCount = table.schemaManager().listAllIds().size();
+ assertThat(icebergSchemaCount).isLessThan(paimonSchemaCount);
+
+ // Perform another write + commit to verify no drift on subsequent
commits
+ // (the deduped currentSchemaId should match what Iceberg stored, so we
+ // should NOT re-enter the schema update path unnecessarily)
+ write.write(GenericRow.of(2, 21));
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ commit.commit(3, write.prepareCommit(true, 3));
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder("Record(1, 11)", "Record(2, 21)",
"Record(3, 30)");
+
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ public void testSchemaEvolutionWithInterleavedOptionAlter() throws
Exception {
+ // Verifies correct behaviour across a full schema-evolution cycle:
+ //
+ // 1. Initial schema: [k, v1, v2]
+ // 2. Drop column v2: [k, v1] — genuine schema change
+ // 3. Option-only alter on [k, v1] — same fields; must be
deduped with step 2
+ // 4. Re-add column v2 (BIGINT): [k, v1, v2_new] — new field ID;
must NOT be
+ // deduped with step 1 even though the column name matches
+ //
+ // Without the fix, step 3 crashes with "Cannot set current schema to
unknown schema: N"
+ // and step 4 would expose a renumbering bug where Iceberg assigns
sequential IDs
+ // (1, 2, 3) while setCurrentSchema references the gap-containing ID
(4).
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT(),
DataTypes.STRING()},
+ new String[] {"k", "v1", "v2"});
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.singletonList("k"),
+ 1,
+ randomFormat(),
+ Collections.emptyMap());
+ String commitUser = UUID.randomUUID().toString();
+ EvolveContext ctx = new EvolveContext(table, commitUser);
+
+ // Step 1: baseline
+ ctx.writeAndCommit(
+ GenericRow.of(1, 10, BinaryString.fromString("a")),
+ GenericRow.of(2, 20, BinaryString.fromString("b")));
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder("Record(1, 10, a)", "Record(2, 20,
b)");
+
+ // Step 2: drop v2 → [k, v1]
+ ctx.evolve(SchemaChange.dropColumn("v2"));
+ ctx.writeAndCommit(GenericRow.of(1, 11), GenericRow.of(3, 30));
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder("Record(1, 11)", "Record(2, 20)",
"Record(3, 30)");
+
+ // Step 3: option-only alter — same fields [k, v1], increments Paimon
schema ID
+ ctx.evolve(SchemaChange.setOption("my.test.option", "trigger-dedup"));
+ ctx.writeAndCommit(GenericRow.of(2, 21));
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder("Record(1, 11)", "Record(2, 21)",
"Record(3, 30)");
+
+ // Step 4: re-add v2 as BIGINT — new field ID, distinct from original
v2 (STRING)
+ ctx.evolve(SchemaChange.addColumn("v2", DataTypes.BIGINT()));
+ ctx.writeAndCommit(GenericRow.of(1, 11, 100L), GenericRow.of(4, 40,
400L));
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder(
+ "Record(1, 11, 100)",
+ "Record(2, 21, null)",
+ "Record(3, 30, null)",
+ "Record(4, 40, 400)");
+
+ // Final Iceberg schema must reflect [k, v1, v2_new(BIGINT)], not the
old v2(STRING)
+ Table icebergTable = restCatalog.loadTable(TableIdentifier.of("mydb",
"t"));
+ assertThat(icebergTable.schema().columns()).hasSize(3);
+
assertThat(icebergTable.schema().findField("v2").type().toString()).isEqualTo("long");
+
+ // Paimon has 4 schema versions (0-3); step 3 is a duplicate of step 2
(same fields).
+ // Iceberg should have exactly 4 schemas: the empty placeholder (id=0)
plus 3 unique
+ // field-sets. Without dedup it would have 5 (empty +
one-per-Paimon-schema).
+ int paimonSchemaCount = ctx.table.schemaManager().listAllIds().size();
+ assertThat(paimonSchemaCount).isEqualTo(4);
+ assertThat(icebergTable.schemas().size()).isEqualTo(4); // 5 without
dedup
+
+ ctx.close();
+ }
+
+ /** Helper context that keeps write/commit in sync with table schema
evolution. */
+ private class EvolveContext implements AutoCloseable {
+ FileStoreTable table;
+ private final String commitUser;
+ private TableWriteImpl<?> write;
+ private TableCommitImpl commit;
+ private int seq = 0;
+
+ EvolveContext(FileStoreTable table, String commitUser) throws
Exception {
+ this.table = table;
+ this.commitUser = commitUser;
+ this.write = table.newWrite(commitUser);
+ this.commit = table.newCommit(commitUser);
+ }
+
+ void evolve(SchemaChange change) throws Exception {
+ new SchemaManager(table.fileIO(),
table.location()).commitChanges(change);
+ table = table.copy(table.schemaManager().latest().get());
+ write.close();
+ write = table.newWrite(commitUser);
+ commit.close();
+ commit = table.newCommit(commitUser);
+ }
+
+ void writeAndCommit(GenericRow... rows) throws Exception {
+ for (GenericRow row : rows) {
+ write.write(row);
+ }
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ commit.commit(++seq, write.prepareCommit(true, seq));
+ }
+
+ @Override
+ public void close() throws Exception {
+ write.close();
+ commit.close();
+ }
+ }
+
@Test
public void testSchemaChangeBeforeSync() throws Exception {
RowType rowType =
