manifests, long assignedRows) {
+ this.manifests = manifests;
+ this.assignedRows = assignedRows;
+ }
}
/**
* Iceberg v3: assign first_row_id (field 520) to data manifests that do not have one yet.
- * Manifests carried over from base metadata keep their value; delete manifests stay null. The
- * watermark starts at the snapshot's first-row-id and advances by each assigned manifest's
- * addedRowsCount: ADDED entries are the only ones with a null per-file first_row_id (EXISTING
- * and DELETED entries are materialized on rewrite), so a manifest's inheriting rows equal its
- * added rows.
+ * Manifests carried over from base metadata that are already assigned keep their value; delete
+ * manifests stay null. The watermark starts at the snapshot's first-row-id and advances by each
+ * newly-assigned manifest's TRUE inheriting-rows count (see {@link #trueInheritingRowsCount}),
+ * returned as {@link ManifestRowIdAssignment#assignedRows}.
+ *
+ * A manifest written entirely by Layer 2 (this commit or a later one) satisfies "null-142
+ * rows == ADDED rows", so {@code addedRowsCount()} is exact for it. But a manifest carried over
+ * from before manifest-level assignment existed (a "Layer-1" manifest) may reach here
+ * unassigned with existing/deleted entries whose per-file field 142 is also still null; for
+ * those, {@code addedRowsCount()} alone would undercount the rows this assignment must cover,
+ * silently shrinking the range handed out and colliding with the next commit's ids. Callers
+ * MUST use {@code assignedRows} (not this commit's added-records count) to advance the
+ * snapshot's added-rows / table next-row-id, precisely because of that mismatch.
*/
- private List assignManifestFirstRowIds(
+ private ManifestRowIdAssignment assignManifestFirstRowIds(
List manifests, @Nullable Long snapshotFirstRowId) {
if (snapshotFirstRowId == null) {
- return manifests;
+ return new ManifestRowIdAssignment(manifests, 0L);
}
List result = new ArrayList<>();
long watermark = snapshotFirstRowId;
@@ -2050,12 +2078,39 @@ private List assignManifestFirstRowIds(
if (meta.content() == IcebergManifestFileMeta.Content.DATA
&& meta.firstRowId() == null) {
result.add(meta.withFirstRowId(watermark));
- watermark += meta.addedRowsCount();
+ watermark += trueInheritingRowsCount(meta);
} else {
result.add(meta);
}
}
- return result;
+ return new ManifestRowIdAssignment(result, watermark - snapshotFirstRowId);
+ }
+
+ /**
+ * The true number of rows an unassigned manifest needs from the row-id space: the sum of {@code
+ * recordCount()} over entries whose per-file first_row_id (field 142) is null.
+ *
+ * Fast path: when the manifest has no existing/deleted entries ({@code existingFilesCount()
+ * + deletedFilesCount() == 0}), every entry is ADDED and, by the Layer-2 invariant, has a null
+ * field 142, so {@code addedRowsCount()} already equals this sum without having to read the
+ * manifest file.
+ *
+ *
Otherwise (a manifest that may carry Layer-1-era existing/deleted entries whose field 142
+ * was never materialized) the manifest is actually read and entries are inspected one by one,
+ * since {@code addedRowsCount()} alone would not include those entries' rows.
+ */
+ private long trueInheritingRowsCount(IcebergManifestFileMeta meta) {
+ if (meta.existingFilesCount() + meta.deletedFilesCount() == 0) {
+ return meta.addedRowsCount();
+ }
+ long sum = 0;
+ for (IcebergManifestEntry entry :
+ manifestFile.read(new Path(meta.manifestPath()).getName())) {
+ if (entry.file().firstRowId() == null) {
+ sum += entry.file().recordCount();
+ }
+ }
+ return sum;
}
/**
diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMetaSerializer.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMetaSerializer.java
index bb38d8107aea..14b6bbf91bed 100644
--- a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMetaSerializer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMetaSerializer.java
@@ -41,6 +41,10 @@ public class IcebergManifestFileMetaSerializer extends ObjectSerializer
3.4.3
diff --git a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java
index 9e2fcf6b968d..08f41e28b1c7 100644
--- a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java
+++ b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java
@@ -69,6 +69,7 @@
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -504,21 +505,66 @@ public void testLayer1TableUpgradesOnNextCommit() throws Exception {
LocalFileIO.create()
.rename(paths.toManifestListPath(rewritten), paths.toManifestListPath(listName));
assertThat(manifestList.read(listName).get(0).firstRowId()).isNull();
+ // the stripped Layer-1 manifest has no existing/deleted entries, so its addedRowsCount()
+ // is an exact count of the legacy rows it carries and still needs a real id for
+ long legacyManifestRows = stripped.get(0).addedRowsCount();
+ assertThat(legacyManifestRows).isEqualTo(1L);
// next commit re-lists carried-over manifests: unassigned metas get assigned now
write.write(GenericRow.of(2, 20));
commit.commit(2, write.prepareCommit(false, 2));
- write.close();
- commit.close();
+ IcebergMetadata metadataAfterCommit2 = readIcebergMetadata(table, 2);
+ IcebergSnapshot snapshotAfterCommit2 = metadataAfterCommit2.currentSnapshot();
+ List dataManifestsAfterCommit2 = new ArrayList<>();
for (IcebergManifestFileMeta meta :
- manifestList.read(
- new Path(readIcebergMetadata(table, 2).currentSnapshot().manifestList())
- .getName())) {
+ manifestList.read(new Path(snapshotAfterCommit2.manifestList()).getName())) {
if (meta.content() == IcebergManifestFileMeta.Content.DATA) {
assertThat(meta.firstRowId()).isNotNull();
+ dataManifestsAfterCommit2.add(meta);
}
}
+ // the re-assigned legacy manifest (M1) and this commit's freshly written manifest (M2)
+ // stay separate: only 2 data manifests, well under the metadata-compaction threshold
+ assertThat(dataManifestsAfterCommit2).hasSize(2);
+
+ // this is the corruption scenario from the review: added-rows/next-row-id must count
+ // the legacy manifest's re-assigned rows in addition to this commit's own new rows, not
+ // just this commit's `metrics.addedRecords` (which is only the new row)
+ long newRowsThisCommit = 1L;
+ long expectedAddedRows = legacyManifestRows + newRowsThisCommit;
+ assertThat(snapshotAfterCommit2.addedRows())
+ .as("snapshot added-rows must include the re-assigned legacy manifest's rows")
+ .isEqualTo(expectedAddedRows);
+ assertThat(metadataAfterCommit2.nextRowId())
+ .as("next-row-id must equal first-row-id plus the true assigned-rows total")
+ .isEqualTo(snapshotAfterCommit2.firstRowId() + snapshotAfterCommit2.addedRows());
+ for (IcebergManifestFileMeta meta : dataManifestsAfterCommit2) {
+ assertThat(metadataAfterCommit2.nextRowId())
+ .as(
+ "next-row-id must be at or beyond every manifest's assigned range "
+ + "(manifest %s)",
+ meta.manifestPath())
+ .isGreaterThanOrEqualTo(meta.firstRowId() + meta.addedRowsCount());
+ }
+
+ // a third commit: its first-row-id must continue exactly where commit 2 left off, and no
+ // manifest's assigned id range may overlap another's (i.e. no file gets a duplicate
+ // effective first-row-id)
+ write.write(GenericRow.of(3, 30));
+ commit.commit(3, write.prepareCommit(false, 3));
+ write.close();
+ commit.close();
+
+ IcebergMetadata metadataAfterCommit3 = readIcebergMetadata(table, 3);
+ IcebergSnapshot snapshotAfterCommit3 = metadataAfterCommit3.currentSnapshot();
+ assertThat(snapshotAfterCommit3.firstRowId()).isEqualTo(metadataAfterCommit2.nextRowId());
+
+ Map effectiveIdsAfterCommit3 = effectiveFileFirstRowIds(table, 3);
+ assertThat(effectiveIdsAfterCommit3).hasSize(3);
+ assertThat(new HashSet<>(effectiveIdsAfterCommit3.values()))
+ .as("no two files may share an effective first-row-id")
+ .hasSameSizeAs(effectiveIdsAfterCommit3.values());
}
@Test
From 83fc5dd2d1143d634a14b299da62fedb29a250c5 Mon Sep 17 00:00:00 2001
From: Victor Babenko <37556649+vbabenkoru@users.noreply.github.com>
Date: Sat, 15 Aug 2026 22:01:43 -0700
Subject: [PATCH 09/15] [iceberg] Expire Iceberg manifests by path, not value
equality
Assigning first_row_id to a carried-over v3 manifest re-lists the same
physical avro file with a different list-level field. expireManifestList
compared IcebergManifestFileMeta values, so expiring the pre-assignment
manifest list deleted the shared file while newer manifest lists still
referenced it, breaking every subsequent Iceberg read of the retained
snapshots. Decide liveness by manifest path instead, and cover the
upgrade boundary with an expiration regression test.
---
.../paimon/iceberg/IcebergCommitCallback.java | 9 ++-
.../IcebergRowLineageCompatibilityTest.java | 61 +++++++++++++++++++
2 files changed, 68 insertions(+), 2 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
index b9dd1593f43d..6a0054bb8bae 100644
--- a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
+++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
@@ -1575,9 +1575,14 @@ private boolean shouldExpire(IcebergSnapshot snapshot, long currentSnapshotId) {
}
private void expireManifestList(String toExpire, String next) {
- Set metaInUse = new HashSet<>(manifestList.read(next));
+ // compare by physical path: a carried-over manifest may be re-listed with different
+ // list-level fields (e.g. an assigned first_row_id) while sharing the same file
+ Set pathsInUse = new HashSet<>();
+ for (IcebergManifestFileMeta meta : manifestList.read(next)) {
+ pathsInUse.add(meta.manifestPath());
+ }
for (IcebergManifestFileMeta meta : manifestList.read(toExpire)) {
- if (metaInUse.contains(meta)) {
+ if (pathsInUse.contains(meta.manifestPath())) {
continue;
}
table.fileIO().deleteQuietly(new Path(meta.manifestPath()));
diff --git a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java
index 08f41e28b1c7..2d3460db33fd 100644
--- a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java
+++ b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java
@@ -349,6 +349,67 @@ public void testReadsManifestListWrittenWithoutFirstRowIdColumn() throws Excepti
}
}
+ @Test
+ public void testExpireKeepsManifestSharedAcrossRowIdAssignment() throws Exception {
+ // A manifest written before manifest-level lineage is re-listed with an assigned
+ // first_row_id but the same physical path. Expiring the pre-assignment manifest list
+ // must not delete the shared file, so liveness is decided by path, not value equality.
+ Map options = formatVersionOptions(3);
+ options.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "1");
+ options.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "2");
+ FileStoreTable table = createPaimonTable(defaultRowType(), options, "avro");
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl write =
+ table.newWrite(commitUser)
+ .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp"));
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(1, 10));
+ commit.commit(1, write.prepareCommit(false, 1));
+
+ IcebergPathFactory paths = new IcebergPathFactory(new Path(table.location(), "metadata"));
+ String listName =
+ new Path(readIcebergMetadata(table, 1).currentSnapshot().manifestList()).getName();
+
+ // strip column 520 from snapshot 1's manifest list to simulate a pre-assignment writer
+ FileStoreTable v2SchemaView =
+ table.copy(Collections.singletonMap(IcebergOptions.FORMAT_VERSION.key(), "2"));
+ IcebergManifestList oldWriter = IcebergManifestList.create(v2SchemaView, paths);
+ IcebergManifestList reader = IcebergManifestList.create(table, paths);
+ String rewritten = oldWriter.writeWithoutRolling(reader.read(listName));
+ LocalFileIO.create().deleteQuietly(paths.toManifestListPath(listName));
+ LocalFileIO.create()
+ .rename(paths.toManifestListPath(rewritten), paths.toManifestListPath(listName));
+ List sharedManifestPaths = new ArrayList<>();
+ for (IcebergManifestFileMeta meta : reader.read(listName)) {
+ sharedManifestPaths.add(meta.manifestPath());
+ }
+ assertThat(sharedManifestPaths).isNotEmpty();
+
+ // commit 2 carries the manifest over, assigning first_row_id at the same path;
+ // commit 3 expires snapshot 1's manifest list against snapshot 2's
+ write.write(GenericRow.of(2, 20));
+ commit.commit(2, write.prepareCommit(false, 2));
+ write.write(GenericRow.of(3, 30));
+ commit.commit(3, write.prepareCommit(false, 3));
+ write.close();
+ commit.close();
+
+ IcebergMetadata metadata = readIcebergMetadata(table, 3);
+ assertThat(metadata.snapshots()).hasSize(2);
+ for (String manifestPath : sharedManifestPaths) {
+ assertThat(LocalFileIO.create().exists(new Path(manifestPath))).isTrue();
+ }
+ // every retained snapshot must stay readable end to end
+ IcebergManifestFile manifestFile = IcebergManifestFile.create(table, paths);
+ for (IcebergSnapshot snapshot : metadata.snapshots()) {
+ for (IcebergManifestFileMeta meta :
+ reader.read(new Path(snapshot.manifestList()).getName())) {
+ assertThat(manifestFile.read(meta)).isNotEmpty();
+ }
+ }
+ }
+
@Test
public void testManifestEntriesCarryFirstRowIdColumn() throws Exception {
FileStoreTable table = createPaimonTable(defaultRowType(), formatVersionOptions(3), "avro");
From bf92a9af71584e8ac230fa0f9bba4f8e082fd840 Mon Sep 17 00:00:00 2001
From: Victor Babenko <37556649+vbabenkoru@users.noreply.github.com>
Date: Sun, 16 Aug 2026 12:24:49 -0700
Subject: [PATCH 10/15] [iceberg] Keep Iceberg 1.8.1 default, gate GA
validation behind iceberg-ga profile
Iceberg 1.10+ ships Java-17 bytecode, so bumping paimon-iceberg to 1.11
outright broke the JDK 11 CI workflows that build this module. Restore
the 1.8.1 default so the module compiles and tests on JDK 11: the few
assertions that need the GA reader API (per-manifest / per-file
firstRowId) resolve it reflectively and skip themselves when absent, the
pre-GA enableRowLineage opt-in in IcebergMetadataTest is called
reflectively so the test passes on both lines, and the update logging in
the REST committer no longer names the RemoveSnapshot(s) class that was
renamed between the two lines. The full GA validation - Iceberg 1.11
with the Hadoop/Jetty/Jackson pins its REST test fixtures need - moves
behind the opt-in iceberg-ga profile (JDK 17):
mvn test -pl paimon-iceberg -Ppaimon-iceberg,iceberg-ga
Verified: default suite on JDK 11 (50 tests, 2 GA-only skips) and
-Piceberg-ga on JDK 17 (50 tests, no skips).
---
paimon-iceberg/pom.xml | 130 +++++++++++-------
.../iceberg/IcebergRestMetadataCommitter.java | 3 -
.../IcebergRowLineageCompatibilityTest.java | 45 +++++-
.../paimon/iceberg/IcebergMetadataTest.java | 18 ++-
.../IcebergRestMetadataCommitterTest.java | 35 ++++-
5 files changed, 168 insertions(+), 63 deletions(-)
diff --git a/paimon-iceberg/pom.xml b/paimon-iceberg/pom.xml
index 18116d5715c4..a4e926979e49 100644
--- a/paimon-iceberg/pom.xml
+++ b/paimon-iceberg/pom.xml
@@ -33,19 +33,12 @@ under the License.
11
- 1.11.0
+ 1.8.1
${paimon-flink-common.flink.version}
- 1.20
+ 1.19
2.3.10
- 1.20.0
-
- 3.4.3
+ 1.19.0
+ 11.0.24
@@ -119,19 +112,6 @@ under the License.
provided
-
-
- com.fasterxml.jackson.core
- jackson-annotations
- 2.21
- provided
-
-
org.apache.iceberg
iceberg-core
@@ -214,10 +194,7 @@ under the License.
commons-io
commons-io
-
-
+
org.eclipse.jetty
*
@@ -262,8 +239,7 @@ under the License.
commons-io
commons-io
-
-
+
org.eclipse.jetty
*
@@ -312,36 +288,25 @@ under the License.
-
+
org.eclipse.jetty
jetty-server
- 12.1.8
- test
-
-
-
- org.eclipse.jetty.ee10
- jetty-ee10-servlet
- 12.1.8
+ ${jetty.server.version}
test
- org.eclipse.jetty.compression
- jetty-compression-server
- 12.1.8
+ org.eclipse.jetty
+ jetty-servlet
+ 11.0.24
test
- org.eclipse.jetty.compression
- jetty-compression-gzip
- 12.1.8
+ jakarta.servlet
+ jakarta.servlet-api
+ 6.1.0
test
@@ -408,4 +373,71 @@ under the License.
+
+
+
+
+ iceberg-ga
+
+ 1.11.0
+ 1.20
+ 1.20.0
+
+ 12.1.8
+
+ 3.4.3
+
+
+
+
+ com.fasterxml.jackson.core
+ jackson-annotations
+ 2.21
+ provided
+
+
+
+ org.eclipse.jetty.ee10
+ jetty-ee10-servlet
+ 12.1.8
+ test
+
+
+
+
+ org.eclipse.jetty
+ jetty-security
+ 12.1.8
+ test
+
+
+
+ org.eclipse.jetty.compression
+ jetty-compression-server
+ 12.1.8
+ test
+
+
+
+ org.eclipse.jetty.compression
+ jetty-compression-gzip
+ 12.1.8
+ test
+
+
+
+
\ No newline at end of file
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 8fe8b8413532..708d14b025e1 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
@@ -682,9 +682,6 @@ private static String updateToString(MetadataUpdate update) {
return String.format(
"AddSnapshot(%s)",
((MetadataUpdate.AddSnapshot) update).snapshot().snapshotId());
- } else if (update instanceof MetadataUpdate.RemoveSnapshots) {
- return String.format(
- "RemoveSnapshots(%s)", ((MetadataUpdate.RemoveSnapshots) update).snapshotIds());
} else if (update instanceof MetadataUpdate.SetSnapshotRef) {
return String.format(
"SetSnapshotRef(%s, %s, %s)",
diff --git a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java
index 2d3460db33fd..706d53913bec 100644
--- a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java
+++ b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java
@@ -62,6 +62,7 @@
import org.apache.iceberg.data.Record;
import org.apache.iceberg.hadoop.HadoopCatalog;
import org.apache.iceberg.io.CloseableIterable;
+import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -502,6 +503,7 @@ public void testFirstRowIdStableAcrossManifestRewrite() throws Exception {
@Test
public void testGaReaderSeesAssignedManifests() throws Exception {
+ assumeGaRowLineageReader();
FileStoreTable table = createPaimonTable(defaultRowType(), formatVersionOptions(3), "avro");
String commitUser = UUID.randomUUID().toString();
TableWriteImpl write =
@@ -519,7 +521,7 @@ public void testGaReaderSeesAssignedManifests() throws Exception {
assertThat(icebergTable.currentSnapshot().firstRowId()).isEqualTo(0L);
for (ManifestFile manifest :
icebergTable.currentSnapshot().dataManifests(icebergTable.io())) {
- assertThat(manifest.firstRowId()).isNotNull();
+ assertThat(manifestFirstRowId(manifest)).isNotNull();
}
}
@@ -722,6 +724,7 @@ public void testCompactMetadataIfNeededMaterializesRowLineageUnderV3() throws Ex
@Test
public void testGaReaderResolvesPerFileRowLineage() throws Exception {
+ assumeGaRowLineageReader();
// standard two-commit 2+1-row setup (matches testNextRowIdAdvancesAcrossCommits):
// commit 1's file gets effective first-row-id 0, commit 2's file gets 2
FileStoreTable table = createPaimonTable(defaultRowType(), formatVersionOptions(3), "avro");
@@ -774,7 +777,7 @@ public void testGaReaderResolvesPerFileRowLineage() throws Exception {
try (ManifestReader reader =
ManifestFiles.read(manifest, icebergTable.io(), icebergTable.specs())) {
for (DataFile file : reader) {
- resolvedFirstRowIds.add(file.firstRowId());
+ resolvedFirstRowIds.add(dataFileFirstRowId(file));
}
}
}
@@ -1075,4 +1078,42 @@ private IcebergMetadata readIcebergMetadata(FileStoreTable table, long snapshotI
private String readMetadataJson(FileStoreTable table, long snapshotId) throws Exception {
return LocalFileIO.create().readFileUtf8(metadataPath(table, snapshotId));
}
+
+ /**
+ * Iceberg exposes per-manifest / per-file {@code firstRowId()} only from the GA row-lineage
+ * line (1.10+). The module compiles against 1.8.1 by default, so GA reader assertions look the
+ * method up reflectively and the tests skip when the API is absent (run with -Piceberg-ga).
+ */
+ private static final boolean GA_ROW_LINEAGE_READER = detectGaRowLineageReader();
+
+ private static boolean detectGaRowLineageReader() {
+ try {
+ ManifestFile.class.getMethod("firstRowId");
+ return true;
+ } catch (NoSuchMethodException e) {
+ return false;
+ }
+ }
+
+ private static void assumeGaRowLineageReader() {
+ Assumptions.assumeTrue(
+ GA_ROW_LINEAGE_READER,
+ "Iceberg on the test classpath predates GA row lineage; run with -Piceberg-ga");
+ }
+
+ private static Long manifestFirstRowId(ManifestFile manifest) {
+ try {
+ return (Long) ManifestFile.class.getMethod("firstRowId").invoke(manifest);
+ } catch (ReflectiveOperationException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static Long dataFileFirstRowId(DataFile file) {
+ try {
+ return (Long) DataFile.class.getMethod("firstRowId").invoke(file);
+ } catch (ReflectiveOperationException e) {
+ throw new IllegalStateException(e);
+ }
+ }
}
diff --git a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergMetadataTest.java b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergMetadataTest.java
index 563b68953f5a..79a2909df18c 100644
--- a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergMetadataTest.java
+++ b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergMetadataTest.java
@@ -29,10 +29,12 @@
import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableUtil;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.hadoop.HadoopCatalog;
@@ -385,10 +387,20 @@ void testReadMetadataWithIcebergPartitioningEvolution() throws Exception {
@Test
@DisplayName("Test FORMAT_VERSION_V3 table")
void testFormatVersionV3Table() throws Exception {
- // Create a v3 format version Iceberg table. Row lineage is always on for v3 in
- // GA Iceberg (the opt-in TableMetadata.Builder#enableRowLineage() was removed before
- // GA), so no extra commit is needed to turn it on.
+ // Create a v3 format version Iceberg table
Table icebergTable = createIcebergTableV3("v3_snapshot_table");
+ try {
+ // pre-GA Iceberg (< 1.10) required opting into v3 row lineage; the builder
+ // method was removed in GA, where row lineage is always on for v3
+ java.lang.reflect.Method enableRowLineage =
+ TableMetadata.Builder.class.getMethod("enableRowLineage");
+ TableMetadata base = ((HasTableOperations) icebergTable).operations().current();
+ TableMetadata.Builder builder = TableMetadata.buildFrom(base);
+ enableRowLineage.invoke(builder);
+ ((HasTableOperations) icebergTable).operations().commit(base, builder.build());
+ } catch (NoSuchMethodException e) {
+ // GA Iceberg: nothing to opt into
+ }
// Read metadata using Paimon's IcebergMetadata
IcebergMetadata paimonIcebergMetadata = readIcebergMetadata(icebergTable);
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 2990ab27a3c0..4d50667c3dbe 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
@@ -1379,13 +1379,16 @@ public void testRecreateWithNonZeroLineageWatermark() throws Exception {
Table reloaded = restCatalog.loadTable(TableIdentifier.of("mydb", "t"));
assertThat(reloaded.currentSnapshot().firstRowId()).isEqualTo(localSnapshot.firstRowId());
- List restDataManifestFirstRowIds = new ArrayList<>();
- for (ManifestFile manifest : reloaded.currentSnapshot().dataManifests(reloaded.io())) {
- assertThat(manifest.firstRowId()).isNotNull();
- restDataManifestFirstRowIds.add(manifest.firstRowId());
+ // manifest-level first_row_id is only exposed by the GA (1.10+) reader API
+ if (GA_ROW_LINEAGE_READER) {
+ List restDataManifestFirstRowIds = new ArrayList<>();
+ for (ManifestFile manifest : reloaded.currentSnapshot().dataManifests(reloaded.io())) {
+ assertThat(manifestFirstRowId(manifest)).isNotNull();
+ restDataManifestFirstRowIds.add(manifestFirstRowId(manifest));
+ }
+ assertThat(restDataManifestFirstRowIds)
+ .containsExactlyInAnyOrderElementsOf(localDataManifestFirstRowIds);
}
- assertThat(restDataManifestFirstRowIds)
- .containsExactlyInAnyOrderElementsOf(localDataManifestFirstRowIds);
}
private static class TestRecord {
@@ -1469,4 +1472,24 @@ private String randomFormat() {
String[] formats = new String[] {"orc", "parquet", "avro"};
return formats[i];
}
+
+ /** See IcebergRowLineageCompatibilityTest: GA reader API (1.10+) looked up reflectively. */
+ private static final boolean GA_ROW_LINEAGE_READER = detectGaRowLineageReader();
+
+ private static boolean detectGaRowLineageReader() {
+ try {
+ ManifestFile.class.getMethod("firstRowId");
+ return true;
+ } catch (NoSuchMethodException e) {
+ return false;
+ }
+ }
+
+ private static Long manifestFirstRowId(ManifestFile manifest) {
+ try {
+ return (Long) ManifestFile.class.getMethod("firstRowId").invoke(manifest);
+ } catch (ReflectiveOperationException e) {
+ throw new IllegalStateException(e);
+ }
+ }
}
From e6f5d017688cfa0079bf669a54c2ca90d31533b2 Mon Sep 17 00:00:00 2001
From: Victor Babenko <37556649+vbabenkoru@users.noreply.github.com>
Date: Mon, 17 Aug 2026 21:11:19 -0700
Subject: [PATCH 11/15] [iceberg] Exclude DELETED entries from v3 row-id
inheritance
GA ManifestReader assigns inherited first_row_id only to non-DELETED
entries. Materialization and manifest-level accounting advanced the
watermark through DELETED entries with a null field 142, shifting every
following live file relative to what readers had already inferred.
Skip DELETED entries when materializing, and reserve each unassigned
manifest's range with the spec-sanctioned added+existing upper bound
instead of reading manifest contents in the commit path: the exact
count, and the serial full-manifest scan it required, are gone;
already-materialized rows now leave spec-legal id gaps.
Regression test: a legacy manifest with a DELETED entry before a live
one keeps the live file's inherited id equal to the manifest's
first_row_id, cross-checked against the GA reader.
Reported by JingsongLi in review.
---
.../paimon/iceberg/IcebergCommitCallback.java | 63 +++-----
.../IcebergRowLineageCompatibilityTest.java | 142 ++++++++++++++++++
2 files changed, 164 insertions(+), 41 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
index 6a0054bb8bae..f8a41f56f28e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
+++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
@@ -2063,14 +2063,13 @@ private ManifestRowIdAssignment(
* newly-assigned manifest's TRUE inheriting-rows count (see {@link #trueInheritingRowsCount}),
* returned as {@link ManifestRowIdAssignment#assignedRows}.
*
- * A manifest written entirely by Layer 2 (this commit or a later one) satisfies "null-142
- * rows == ADDED rows", so {@code addedRowsCount()} is exact for it. But a manifest carried over
- * from before manifest-level assignment existed (a "Layer-1" manifest) may reach here
- * unassigned with existing/deleted entries whose per-file field 142 is also still null; for
- * those, {@code addedRowsCount()} alone would undercount the rows this assignment must cover,
- * silently shrinking the range handed out and colliding with the next commit's ids. Callers
- * MUST use {@code assignedRows} (not this commit's added-records count) to advance the
- * snapshot's added-rows / table next-row-id, precisely because of that mismatch.
+ *
A manifest written entirely under manifest-level assignment satisfies "inheriting rows ==
+ * ADDED rows", so the bound is exact for it. A manifest carried over from before assignment
+ * existed may hold EXISTING entries whose field 142 is also still null; the bound covers them
+ * without reading the manifest, at the cost of spec-legal id gaps when some of those entries
+ * were already materialized. DELETED entries never inherit ids and are excluded. Callers MUST
+ * use {@code assignedRows} (not this commit's added-records count) to advance the snapshot's
+ * added-rows / table next-row-id, precisely because of that mismatch.
*/
private ManifestRowIdAssignment assignManifestFirstRowIds(
List manifests, @Nullable Long snapshotFirstRowId) {
@@ -2083,7 +2082,12 @@ private ManifestRowIdAssignment assignManifestFirstRowIds(
if (meta.content() == IcebergManifestFileMeta.Content.DATA
&& meta.firstRowId() == null) {
result.add(meta.withFirstRowId(watermark));
- watermark += trueInheritingRowsCount(meta);
+ // spec-sanctioned upper bound: only ADDED and EXISTING rows can inherit
+ // ids from this manifest (readers never assign ids to DELETED entries).
+ // Rows whose field 142 is already materialized merely widen the reserved
+ // range, leaving legal id gaps - in exchange the commit path never has to
+ // read manifest contents.
+ watermark += meta.addedRowsCount() + meta.existingRowsCount();
} else {
result.add(meta);
}
@@ -2091,40 +2095,14 @@ private ManifestRowIdAssignment assignManifestFirstRowIds(
return new ManifestRowIdAssignment(result, watermark - snapshotFirstRowId);
}
- /**
- * The true number of rows an unassigned manifest needs from the row-id space: the sum of {@code
- * recordCount()} over entries whose per-file first_row_id (field 142) is null.
- *
- * Fast path: when the manifest has no existing/deleted entries ({@code existingFilesCount()
- * + deletedFilesCount() == 0}), every entry is ADDED and, by the Layer-2 invariant, has a null
- * field 142, so {@code addedRowsCount()} already equals this sum without having to read the
- * manifest file.
- *
- *
Otherwise (a manifest that may carry Layer-1-era existing/deleted entries whose field 142
- * was never materialized) the manifest is actually read and entries are inspected one by one,
- * since {@code addedRowsCount()} alone would not include those entries' rows.
- */
- private long trueInheritingRowsCount(IcebergManifestFileMeta meta) {
- if (meta.existingFilesCount() + meta.deletedFilesCount() == 0) {
- return meta.addedRowsCount();
- }
- long sum = 0;
- for (IcebergManifestEntry entry :
- manifestFile.read(new Path(meta.manifestPath()).getName())) {
- if (entry.file().firstRowId() == null) {
- sum += entry.file().recordCount();
- }
- }
- return sum;
- }
-
/**
* Iceberg v3 requires the inherited first_row_id to be written into file metadata when entries
* are copied into a rewritten manifest. Computes each entry's effective id in base manifest
- * order (explicit field 142, or inherited from the manifest's first_row_id) and returns entries
- * with the id materialized. No-op for delete manifests and for base manifests without an
- * assigned first_row_id (v2 metadata, or v3 metadata written before manifest-level assignment
- * existed — those stay in the spec's upgraded-table state).
+ * order (explicit field 142, or inherited from the manifest's first_row_id, skipping DELETED
+ * entries exactly like GA readers do) and returns entries with the id materialized. No-op for
+ * delete manifests and for base manifests without an assigned first_row_id (v2 metadata, or v3
+ * metadata written before manifest-level assignment existed — those stay in the spec's
+ * upgraded-table state).
*/
private static List materializeFirstRowIds(
IcebergManifestFileMeta baseMeta, List entries) {
@@ -2135,10 +2113,13 @@ private static List materializeFirstRowIds(
List result = new ArrayList<>();
long watermark = baseMeta.firstRowId();
for (IcebergManifestEntry entry : entries) {
- if (entry.file().firstRowId() == null) {
+ if (entry.status() != IcebergManifestEntry.Status.DELETED
+ && entry.file().firstRowId() == null) {
result.add(entry.withFile(entry.file().withFirstRowId(watermark)));
watermark += entry.file().recordCount();
} else {
+ // DELETED entries never inherit an id (GA readers skip them when
+ // assigning), so their field 142 stays null and the walk does not advance
result.add(entry);
}
}
diff --git a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java
index 706d53913bec..55a1eacb665d 100644
--- a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java
+++ b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java
@@ -501,6 +501,143 @@ public void testFirstRowIdStableAcrossManifestRewrite() throws Exception {
.isTrue();
}
+ @Test
+ public void testDeletedEntriesDoNotShiftInheritedIds() throws Exception {
+ // A legacy (pre-assignment) manifest holding a DELETED entry before a live one: GA
+ // readers skip DELETED entries when assigning inherited ids, so the live entry
+ // inherits the manifest's first_row_id itself, NOT shifted by the deleted rows.
+ RowType rowType = defaultRowType();
+ Map customOptions = formatVersionOptions(3);
+ customOptions.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ customOptions.put(CoreOptions.DELETION_VECTOR_BITMAP64.key(), "true");
+ FileStoreTable table = createPkPaimonTable(rowType, customOptions);
+
+ int keyBucket0 = findKeyForBucket(table, 0);
+ int keyBucket1 = findKeyForBucket(table, 1);
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl write =
+ table.newWrite(commitUser)
+ .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp"));
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(keyBucket0, 10));
+ write.write(GenericRow.of(keyBucket1, 20));
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ write.compact(BinaryRow.EMPTY_ROW, 1, true);
+ commit.commit(1, write.prepareCommit(true, 1));
+
+ // rewrite the shared manifest: bucket 0's entry becomes DELETED, bucket 1's is
+ // carried as EXISTING behind it
+ write.write(GenericRow.ofKind(RowKind.DELETE, keyBucket0, 10));
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ commit.commit(2, write.prepareCommit(true, 2));
+
+ long rewriteSnapshot = table.snapshotManager().latestSnapshotId();
+ IcebergPathFactory paths = new IcebergPathFactory(new Path(table.location(), "metadata"));
+ IcebergManifestList listReader = IcebergManifestList.create(table, paths);
+ IcebergManifestFile entryReader = IcebergManifestFile.create(table, paths);
+ String listName =
+ new Path(
+ readIcebergMetadata(table, rewriteSnapshot)
+ .currentSnapshot()
+ .manifestList())
+ .getName();
+
+ // strip lineage from the manifest with the DELETED entry and from the list,
+ // simulating metadata written before manifest-level assignment existed
+ FileStoreTable v2SchemaView =
+ table.copy(Collections.singletonMap(IcebergOptions.FORMAT_VERSION.key(), "2"));
+ IcebergManifestFile oldEntryWriter = IcebergManifestFile.create(v2SchemaView, paths);
+ String strippedPath = null;
+ for (IcebergManifestFileMeta meta : listReader.read(listName)) {
+ if (meta.content() != IcebergManifestFileMeta.Content.DATA
+ || meta.deletedFilesCount() == 0) {
+ continue;
+ }
+ strippedPath = meta.manifestPath();
+ List entries = entryReader.read(meta);
+ assertThat(entries.get(0).status()).isEqualTo(IcebergManifestEntry.Status.DELETED);
+ List rewritten =
+ oldEntryWriter.rollingWrite(
+ entries.iterator(), meta.sequenceNumber(), meta.content());
+ Path target = new Path(meta.manifestPath());
+ LocalFileIO.create().deleteQuietly(target);
+ LocalFileIO.create().rename(new Path(rewritten.get(0).manifestPath()), target);
+ }
+ assertThat(strippedPath).isNotNull();
+ final String stripped = strippedPath;
+ IcebergManifestList oldListWriter = IcebergManifestList.create(v2SchemaView, paths);
+ String rewrittenList = oldListWriter.writeWithoutRolling(listReader.read(listName));
+ LocalFileIO.create().deleteQuietly(paths.toManifestListPath(listName));
+ LocalFileIO.create()
+ .rename(
+ paths.toManifestListPath(rewrittenList),
+ paths.toManifestListPath(listName));
+
+ // next commit re-assigns the stripped manifest at list level
+ write.write(GenericRow.of(keyBucket0, 11));
+ commit.commit(3, write.prepareCommit(false, 3));
+ write.close();
+ commit.close();
+
+ long latest = table.snapshotManager().latestSnapshotId();
+ Long assignedFirst = null;
+ long upperBoundAdvance = 0;
+ String latestList =
+ new Path(readIcebergMetadata(table, latest).currentSnapshot().manifestList())
+ .getName();
+ for (IcebergManifestFileMeta meta : listReader.read(latestList)) {
+ if (meta.manifestPath().equals(strippedPath)) {
+ assignedFirst = meta.firstRowId();
+ upperBoundAdvance = meta.addedRowsCount() + meta.existingRowsCount();
+ }
+ }
+ assertThat(assignedFirst).isNotNull();
+ // the manifest reserves only its ADDED+EXISTING rows; the DELETED entry is excluded
+ assertThat(upperBoundAdvance).isEqualTo(1L);
+
+ // the live entry inherits the manifest's own first_row_id, not shifted by the
+ // 1-row DELETED entry sitting before it
+ Map effective = effectiveFileFirstRowIds(table, latest);
+ boolean sawSurvivor = false;
+ for (IcebergManifestEntry entry :
+ entryReader.read(
+ listReader.read(latestList).stream()
+ .filter(m -> m.manifestPath().equals(stripped))
+ .findFirst()
+ .get())) {
+ if (entry.isLive()) {
+ sawSurvivor = true;
+ assertThat(entry.file().firstRowId()).isNull();
+ assertThat(effective.get(entry.file().filePath())).isEqualTo(assignedFirst);
+ }
+ }
+ assertThat(sawSurvivor).isTrue();
+
+ // GA reader cross-check: Iceberg 1.11 resolves the same id for the live file
+ if (GA_ROW_LINEAGE_READER) {
+ HadoopCatalog icebergCatalog =
+ new HadoopCatalog(new Configuration(), tempDir.toString());
+ Table icebergTable = icebergCatalog.loadTable(TableIdentifier.of("mydb2.db", "t"));
+ boolean checked = false;
+ for (ManifestFile manifest :
+ icebergTable.currentSnapshot().dataManifests(icebergTable.io())) {
+ if (!manifest.path().equals(strippedPath)) {
+ continue;
+ }
+ try (ManifestReader gaReader =
+ ManifestFiles.read(manifest, icebergTable.io(), icebergTable.specs())) {
+ for (DataFile file : gaReader) {
+ checked = true;
+ assertThat(dataFileFirstRowId(file)).isEqualTo(assignedFirst);
+ }
+ }
+ }
+ assertThat(checked).isTrue();
+ }
+ }
+
@Test
public void testGaReaderSeesAssignedManifests() throws Exception {
assumeGaRowLineageReader();
@@ -1000,6 +1137,11 @@ private Map effectiveFileFirstRowIds(FileStoreTable table, long sn
}
long watermark = meta.firstRowId() == null ? -1 : meta.firstRowId();
for (IcebergManifestEntry entry : manifestFile.read(meta)) {
+ if (entry.status() == IcebergManifestEntry.Status.DELETED) {
+ // GA readers never assign ids to DELETED entries; they must not
+ // advance the inheritance walk either
+ continue;
+ }
long effective;
if (entry.file().firstRowId() != null) {
effective = entry.file().firstRowId();
From 9364112cd6808a80a76f4a554637ac654e44bbaf Mon Sep 17 00:00:00 2001
From: Victor Babenko <37556649+vbabenkoru@users.noreply.github.com>
Date: Mon, 17 Aug 2026 21:11:19 -0700
Subject: [PATCH 12/15] [iceberg] Import local metadata wholesale when
registering REST tables
Rebuilding a REST table through TableMetadata.Builder recomputes
next-row-id from each snapshot's added-rows, so an id space that does
not start at 0 (self-heal and rollback rebuilds) ended below the ids
already assigned in manifests, and a later external REST writer could
have reused them.
Recreation, initial creation of a table with history, and the
exists-but-empty recovery now write the REST-adjusted metadata to a
file and register it, importing every high-water mark verbatim. The
drop+create failure loop the empty-table recovery guarded against
cannot occur here: registration has no post-create commit step.
Registration also preserves the real partition spec when a partition
column has field id 0, where creation used to fall back to an
unpartitioned spec.
Tests assert the recreated server watermark is at least the local one
and that an external append through the Iceberg API allocates above it.
Reported by JingsongLi in review.
---
.../iceberg/IcebergRestMetadataCommitter.java | 90 +++++++++++++++----
.../IcebergRestMetadataCommitterTest.java | 46 ++++++++--
2 files changed, 112 insertions(+), 24 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 708d14b025e1..5f9fe63cf3ee 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
@@ -20,6 +20,7 @@
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.iceberg.metadata.IcebergMetadata;
import org.apache.paimon.iceberg.metadata.IcebergSchema;
@@ -79,6 +80,8 @@ public class IcebergRestMetadataCommitter implements IcebergMetadataCommitter {
private static final String REST_CATALOG_NAME = "rest-catalog";
private final RESTCatalog restCatalog;
+ private final FileIO fileIO;
+ private final Path tableLocation;
private final String icebergDatabaseName;
private final TableIdentifier icebergTableIdentifier;
private final IcebergOptions icebergOptions;
@@ -88,6 +91,8 @@ public class IcebergRestMetadataCommitter implements IcebergMetadataCommitter {
public IcebergRestMetadataCommitter(FileStoreTable table) {
Options options = new Options(table.options());
icebergOptions = new IcebergOptions(options);
+ this.fileIO = table.fileIO();
+ this.tableLocation = table.location();
Identifier identifier = Preconditions.checkNotNull(table.catalogEnvironment().identifier());
String icebergDatabase = options.get(IcebergOptions.METASTORE_DATABASE);
@@ -151,6 +156,14 @@ private void commitMetadataImpl(
try {
if (!tableExists()) {
+ if (newMetadata.currentSnapshot() != null) {
+ // metadata with history is imported wholesale: replaying it through
+ // TableMetadata.Builder would recompute next-row-id from added-rows
+ // and lose any high-water mark that does not start at 0
+ LOG.info("Table {} does not exist, register it.", icebergTableIdentifier);
+ registerAsCurrent(newIcebergMetadata, newMetadata, false);
+ return;
+ }
LOG.info("Table {} does not exist, create it.", icebergTableIdentifier);
icebergTable = createTable(newMetadata);
updateBuilder =
@@ -191,6 +204,13 @@ private void commitMetadataImpl(
LOG.info(
"Iceberg table {} exists but has no snapshots, treating as new table.",
icebergTableIdentifier);
+ if (newMetadata.currentSnapshot() != null) {
+ // same watermark argument as above; registration has no
+ // post-create commit step, so the drop+create failure loop this
+ // branch used to guard against cannot occur
+ registerAsCurrent(newIcebergMetadata, newMetadata, true);
+ return;
+ }
updateBuilder = updatesForCorrectBase(metadata, newMetadata, true);
} else {
boolean withBase = checkBase(metadata, newMetadata, baseIcebergMetadata);
@@ -204,7 +224,11 @@ private void commitMetadataImpl(
newMetadata.currentSnapshot() != null
? newMetadata.currentSnapshot().snapshotId()
: "No snapshot");
- updateBuilder = updatesForIncorrectBase(newMetadata);
+ LOG.info(
+ "the base metadata is incorrect, re-registering the iceberg"
+ + " table from local metadata.");
+ registerAsCurrent(newIcebergMetadata, newMetadata, true);
+ return;
}
}
}
@@ -284,13 +308,6 @@ private TableMetadata.Builder updatesForCorrectBase(
return updateBuilder;
}
- private TableMetadata.Builder updatesForIncorrectBase(TableMetadata newMetadata) {
- LOG.info("the base metadata is incorrect, we'll recreate the iceberg table.");
- icebergTable = recreateTable(newMetadata);
- return updatesForCorrectBase(
- ((BaseTable) icebergTable).operations().current(), newMetadata, true);
- }
-
private RESTCatalog initRestCatalog(Map restConfigs, Configuration conf) {
restConfigs.put(ICEBERG_CATALOG_TYPE, "rest");
Catalog catalog = CatalogUtil.buildIcebergCatalog(REST_CATALOG_NAME, restConfigs, conf);
@@ -321,6 +338,54 @@ void createDatabase() {
}
}
+ /**
+ * Makes the catalog's state exactly the (REST-adjusted) local metadata by registering a
+ * metadata file, instead of rebuilding the table through {@link TableMetadata.Builder}. The
+ * builder recomputes the row-id high-water mark from each snapshot's added-rows, so any id
+ * space that does not start at 0 (self-heal and rollback rebuilds) would end below the ids
+ * already assigned in manifests, and a later external writer could reuse them. Registration
+ * imports every field of the metadata verbatim, next-row-id included.
+ */
+ private void registerAsCurrent(
+ IcebergMetadata adjustedMetadata, TableMetadata newMetadata, boolean dropFirst) {
+ try {
+ Path registerPath =
+ new Path(
+ tableLocation,
+ String.format(
+ "metadata/rest-register-v%d.metadata.json",
+ adjustedMetadata.currentSnapshotId()));
+ if (!fileIO.tryToWriteAtomic(registerPath, adjustedMetadata.toJson())) {
+ fileIO.deleteQuietly(registerPath);
+ fileIO.overwriteFileUtf8(registerPath, adjustedMetadata.toJson());
+ }
+ if (dropFirst) {
+ dropTable();
+ }
+ icebergTable =
+ restCatalog.registerTable(icebergTableIdentifier, registerPath.toString());
+ long registered =
+ ((BaseTable) icebergTable)
+ .operations()
+ .current()
+ .currentSnapshot()
+ .snapshotId();
+ if (newMetadata.currentSnapshot() == null
+ || registered != newMetadata.currentSnapshot().snapshotId()) {
+ throw new IllegalStateException(
+ String.format(
+ "Registered catalog table is at snapshot %s instead of %s",
+ registered,
+ newMetadata.currentSnapshot() == null
+ ? "null"
+ : newMetadata.currentSnapshot().snapshotId()));
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Fail to register iceberg table " + icebergTableIdentifier, e);
+ }
+ }
+
private Table createTable(TableMetadata newMetadata) {
/*
Handles fieldId incompatibility between Paimon (starts at 0) and Iceberg (starts at 1).
@@ -419,15 +484,6 @@ private void dropTable() {
restCatalog.dropTable(icebergTableIdentifier, false);
}
- private Table recreateTable(TableMetadata newMetadata) {
- try {
- dropTable();
- return createTable(newMetadata);
- } catch (Exception e) {
- throw new RuntimeException("Fail to recreate iceberg table.", e);
- }
- }
-
// -------------------------------------------------------------------------------------
// metadata updates
// -------------------------------------------------------------------------------------
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 4d50667c3dbe..b104e5ca2a57 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
@@ -49,6 +49,7 @@
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableUtil;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
@@ -219,8 +220,15 @@ public void testPartitionedPrimaryKeyTable() throws Exception {
expected,
Record::toString);
- PartitionSpec expectedPartitionSpec = PartitionSpec.builderFor(new Schema()).build();
- runPartitionSpecCompatibilityTest(expectedPartitionSpec);
+ // registration imports the real partition spec even when a partition column has
+ // field id 0 (the create-path fallback used to register an unpartitioned spec)
+ PartitionSpec registeredSpec =
+ restCatalog.loadTable(TableIdentifier.of("mydb", "t")).spec();
+ assertThat(registeredSpec.fields()).hasSize(2);
+ assertThat(registeredSpec.fields().get(0).name()).isEqualTo("pt1");
+ assertThat(registeredSpec.fields().get(0).sourceId()).isEqualTo(0);
+ assertThat(registeredSpec.fields().get(1).name()).isEqualTo("pt2");
+ assertThat(registeredSpec.fields().get(1).sourceId()).isEqualTo(1);
}
@Test
@@ -411,9 +419,9 @@ public void testSchemaAndPropertiesChange() throws Exception {
Table icebergTable = restCatalog.loadTable(TableIdentifier.of("mydb", "t"));
assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(5);
- // 1 metadata for createTable + 4 history metadata
+ // the registered metadata plus later commits' history
assertThat(((BaseTable) icebergTable).operations().current().previousFiles().size())
- .isEqualTo(5);
+ .isEqualTo(4);
write.close();
commit.close();
@@ -633,11 +641,11 @@ public void testSchemaEvolutionWithInterleavedOptionAlter() throws Exception {
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).
+ // The registered table carries exactly the 3 unique field-sets: registration imports
+ // the deduplicated local schemas verbatim, with no placeholder schema.
int paimonSchemaCount = ctx.table.schemaManager().listAllIds().size();
assertThat(paimonSchemaCount).isEqualTo(4);
- assertThat(icebergTable.schemas().size()).isEqualTo(4); // 5 without dedup
+ assertThat(icebergTable.schemas().size()).isEqualTo(3); // 4 without dedup
ctx.close();
}
@@ -1388,6 +1396,19 @@ public void testRecreateWithNonZeroLineageWatermark() throws Exception {
}
assertThat(restDataManifestFirstRowIds)
.containsExactlyInAnyOrderElementsOf(localDataManifestFirstRowIds);
+
+ // the registered server table must carry the full row-id high-water mark:
+ // external REST writers allocate from it, so anything lower reuses ids
+ Long serverNextRowId =
+ tableMetadataNextRowId(((BaseTable) reloaded).operations().current());
+ assertThat(serverNextRowId).isNotNull();
+ assertThat(serverNextRowId).isGreaterThanOrEqualTo(localMetadata.nextRowId());
+
+ // one external append through the Iceberg API allocates above the watermark
+ reloaded.newAppend().commit();
+ Table afterExternal = restCatalog.loadTable(TableIdentifier.of("mydb", "t"));
+ assertThat(afterExternal.currentSnapshot().firstRowId())
+ .isGreaterThanOrEqualTo(localMetadata.nextRowId());
}
}
@@ -1492,4 +1513,15 @@ private static Long manifestFirstRowId(ManifestFile manifest) {
throw new IllegalStateException(e);
}
}
+
+ /** TableMetadata#nextRowId is a GA (1.10+) API; resolve reflectively. */
+ private static Long tableMetadataNextRowId(TableMetadata metadata) {
+ try {
+ return (Long) TableMetadata.class.getMethod("nextRowId").invoke(metadata);
+ } catch (NoSuchMethodException e) {
+ return null;
+ } catch (ReflectiveOperationException e) {
+ throw new IllegalStateException(e);
+ }
+ }
}
From 29d33c8b0a5975dc29bab9f998374873d970ad0a Mon Sep 17 00:00:00 2001
From: Victor Babenko <37556649+vbabenkoru@users.noreply.github.com>
Date: Mon, 17 Aug 2026 21:11:19 -0700
Subject: [PATCH 13/15] [iceberg] Run the GA row-lineage validation as a CI
workflow
The iceberg-ga profile was not exercised by CI, and running it as a
single 'mvn test -am' session fails: the not-yet-shaded paimon-bundle
is substituted with its unshaded constituent modules, whose direct
Avro references clash with the Avro 1.12 line Iceberg 1.11 requires
(the installed bundle relocates Avro and has no clash).
Add a JDK 17 workflow that installs the shaded bundle first and then
runs the suite, and document the two-step invocation in the profile.
Reported by JingsongLi in review.
---
.github/workflows/utitcase-iceberg-ga.yml | 66 +++++++++++++++++++++++
paimon-iceberg/pom.xml | 10 +++-
2 files changed, 75 insertions(+), 1 deletion(-)
create mode 100644 .github/workflows/utitcase-iceberg-ga.yml
diff --git a/.github/workflows/utitcase-iceberg-ga.yml b/.github/workflows/utitcase-iceberg-ga.yml
new file mode 100644
index 000000000000..deca2434fc89
--- /dev/null
+++ b/.github/workflows/utitcase-iceberg-ga.yml
@@ -0,0 +1,66 @@
+################################################################################
+# 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.
+################################################################################
+
+name: UTCase Iceberg GA row lineage on JDK 17
+
+on:
+ push:
+ paths:
+ - 'paimon-iceberg/**'
+ - 'paimon-core/**'
+ - 'paimon-common/**'
+ - 'paimon-api/**'
+ - 'paimon-format/**'
+ - 'pom.xml'
+ - '.github/workflows/utitcase-iceberg-ga.yml'
+ pull_request:
+ paths:
+ - 'paimon-iceberg/**'
+ - 'paimon-core/**'
+ - 'paimon-common/**'
+ - 'paimon-api/**'
+ - 'paimon-format/**'
+ - 'pom.xml'
+ - '.github/workflows/utitcase-iceberg-ga.yml'
+
+env:
+ JDK_VERSION: 17
+ MAVEN_OPTS: -Dmaven.wagon.httpconnectionManager.ttlSeconds=30 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.event.number || github.run_id }}
+ cancel-in-progress: true
+
+jobs:
+ build:
+ runs-on: ubuntu-latest
+
+ steps:
+ - name: Checkout code
+ uses: actions/checkout@v6
+ - name: Set up JDK ${{ env.JDK_VERSION }}
+ uses: actions/setup-java@v5
+ with:
+ java-version: ${{ env.JDK_VERSION }}
+ distribution: 'temurin'
+ - name: Build
+ run: mvn -T 1C -B -ntp clean install -DskipTests -pl paimon-iceberg -am -Ppaimon-iceberg,iceberg-ga
+ - name: Test against GA Iceberg
+ run: mvn -B -ntp test -pl paimon-iceberg -Ppaimon-iceberg,iceberg-ga
+ env:
+ MAVEN_OPTS: -Xmx4096m
diff --git a/paimon-iceberg/pom.xml b/paimon-iceberg/pom.xml
index a4e926979e49..bf02653af2d2 100644
--- a/paimon-iceberg/pom.xml
+++ b/paimon-iceberg/pom.xml
@@ -380,7 +380,15 @@ under the License.
the reference implementation of Iceberg format-version 3 row lineage. Iceberg
1.10+ ships Java-17 bytecode, so this profile requires JDK 17 and stays
opt-in: the default build keeps Iceberg 1.8.1 so the module compiles and
- tests on the JDK 11 CI, where GA-only reader assertions skip themselves. -->
+ tests on the JDK 11 CI, where GA-only reader assertions skip themselves.
+
+ Run as two invocations (as the iceberg-ga CI workflow does):
+ mvn install -DskipTests -pl paimon-iceberg -am -Ppaimon-iceberg,iceberg-ga
+ mvn test -pl paimon-iceberg -Ppaimon-iceberg,iceberg-ga
+ A single `test -am` session substitutes the not-yet-shaded paimon-bundle with
+ its unshaded constituent modules, whose direct Avro references clash with the
+ Avro 1.12 line Iceberg 1.11 requires; the installed bundle relocates Avro and
+ has no such clash. -->
iceberg-ga
1.11.0
From 1aa1241199c1a5676c23c9f9ac226aad23bb982a Mon Sep 17 00:00:00 2001
From: Victor Babenko <37556649+vbabenkoru@users.noreply.github.com>
Date: Tue, 18 Aug 2026 14:50:38 -0700
Subject: [PATCH 14/15] [iceberg] Register REST tables only when the catalog
supports it, write-once
registerTable is not part of every REST catalog: AWS Glue's Iceberg REST
endpoint implements create/load/update/delete but not register, so
routing every first publication through registration broke such
catalogs. Registration is now used only when the create/update path
would lose the row-id watermark, i.e. for format version 3 metadata
whose current snapshot has a nonzero first-row-id (rollback and
self-heal rebuilds). Format version 2 tables and zero-based v3 metadata
keep the create/update path exactly as before.
When registration is required, the committer first checks the endpoint
set the client took from the server's config response and, before
dropping an existing table, attempts the registration against it: a
server that implements the endpoint rejects it with AlreadyExists,
anything else fails the commit with the catalog table untouched.
Registered metadata files are written once under a UUID-suffixed name
and never overwritten: a catalog may keep referencing the location, and
rollback and rebuild paths reuse Paimon snapshot ids for different
timelines.
Tests simulate a catalog without the endpoint by removing it from the
client's endpoint set: zero-based v3 and v2 metadata still publish
(including drop+recreate on v2), nonzero rebuilds fail closed with the
table left in place, and repeated registration of the same snapshot id
leaves the earlier file intact.
Reported by JingsongLi in review.
---
.../iceberg/IcebergRestMetadataCommitter.java | 229 ++++++++++++----
.../IcebergRestMetadataCommitterTest.java | 245 +++++++++++++++++-
2 files changed, 414 insertions(+), 60 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 5f9fe63cf3ee..327361b9d3f8 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
@@ -46,6 +46,7 @@
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.rest.Endpoint;
import org.apache.iceberg.rest.RESTCatalog;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.types.Types.NestedField;
@@ -54,6 +55,8 @@
import javax.annotation.Nullable;
+import java.io.IOException;
+import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
@@ -61,6 +64,7 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
import java.util.stream.Collectors;
import static org.apache.iceberg.CatalogUtil.ICEBERG_CATALOG_TYPE;
@@ -81,7 +85,7 @@ public class IcebergRestMetadataCommitter implements IcebergMetadataCommitter {
private final RESTCatalog restCatalog;
private final FileIO fileIO;
- private final Path tableLocation;
+ private final Path metadataDirectory;
private final String icebergDatabaseName;
private final TableIdentifier icebergTableIdentifier;
private final IcebergOptions icebergOptions;
@@ -92,7 +96,7 @@ public IcebergRestMetadataCommitter(FileStoreTable table) {
Options options = new Options(table.options());
icebergOptions = new IcebergOptions(options);
this.fileIO = table.fileIO();
- this.tableLocation = table.location();
+ this.metadataDirectory = IcebergCommitCallback.catalogTableMetadataPath(table);
Identifier identifier = Preconditions.checkNotNull(table.catalogEnvironment().identifier());
String icebergDatabase = options.get(IcebergOptions.METASTORE_DATABASE);
@@ -156,10 +160,7 @@ private void commitMetadataImpl(
try {
if (!tableExists()) {
- if (newMetadata.currentSnapshot() != null) {
- // metadata with history is imported wholesale: replaying it through
- // TableMetadata.Builder would recompute next-row-id from added-rows
- // and lose any high-water mark that does not start at 0
+ if (requiresRegistration(newIcebergMetadata)) {
LOG.info("Table {} does not exist, register it.", icebergTableIdentifier);
registerAsCurrent(newIcebergMetadata, newMetadata, false);
return;
@@ -204,10 +205,9 @@ private void commitMetadataImpl(
LOG.info(
"Iceberg table {} exists but has no snapshots, treating as new table.",
icebergTableIdentifier);
- if (newMetadata.currentSnapshot() != null) {
- // same watermark argument as above; registration has no
- // post-create commit step, so the drop+create failure loop this
- // branch used to guard against cannot occur
+ if (requiresRegistration(newIcebergMetadata)) {
+ // registration has no post-create commit step, so the drop+create
+ // failure loop this branch guards against cannot occur
registerAsCurrent(newIcebergMetadata, newMetadata, true);
return;
}
@@ -224,11 +224,14 @@ private void commitMetadataImpl(
newMetadata.currentSnapshot() != null
? newMetadata.currentSnapshot().snapshotId()
: "No snapshot");
- LOG.info(
- "the base metadata is incorrect, re-registering the iceberg"
- + " table from local metadata.");
- registerAsCurrent(newIcebergMetadata, newMetadata, true);
- return;
+ if (requiresRegistration(newIcebergMetadata)) {
+ LOG.info(
+ "the base metadata is incorrect, re-registering the iceberg"
+ + " table from local metadata.");
+ registerAsCurrent(newIcebergMetadata, newMetadata, true);
+ return;
+ }
+ updateBuilder = updatesForIncorrectBase(newMetadata);
}
}
}
@@ -308,6 +311,13 @@ private TableMetadata.Builder updatesForCorrectBase(
return updateBuilder;
}
+ private TableMetadata.Builder updatesForIncorrectBase(TableMetadata newMetadata) {
+ LOG.info("the base metadata is incorrect, we'll recreate the iceberg table.");
+ icebergTable = recreateTable(newMetadata);
+ return updatesForCorrectBase(
+ ((BaseTable) icebergTable).operations().current(), newMetadata, true);
+ }
+
private RESTCatalog initRestCatalog(Map restConfigs, Configuration conf) {
restConfigs.put(ICEBERG_CATALOG_TYPE, "rest");
Catalog catalog = CatalogUtil.buildIcebergCatalog(REST_CATALOG_NAME, restConfigs, conf);
@@ -338,54 +348,172 @@ void createDatabase() {
}
}
+ /**
+ * Whether publishing {@code metadata} to a new or recreated catalog table must go through
+ * {@link RESTCatalog#registerTable}. The create/update path replays only the current snapshot
+ * through {@link TableMetadata.Builder}, which derives the table's next-row-id from that
+ * snapshot's added-rows alone, so the server ends at {@code added-rows} while the local
+ * watermark is {@code first-row-id + added-rows}. Any nonzero first-row-id (rollback and
+ * self-heal rebuilds) would leave the server below ids already assigned in manifests, and a
+ * later external writer could reuse them; registration imports the metadata verbatim. Format
+ * version 2 tables and zero-based v3 metadata keep the create/update path, which every REST
+ * catalog supports.
+ */
+ private static boolean requiresRegistration(IcebergMetadata metadata) {
+ IcebergSnapshot current = metadata.currentSnapshot();
+ return metadata.formatVersion() >= IcebergMetadata.FORMAT_VERSION_V3
+ && current != null
+ && current.firstRowId() != null
+ && current.firstRowId() != 0;
+ }
+
/**
* Makes the catalog's state exactly the (REST-adjusted) local metadata by registering a
- * metadata file, instead of rebuilding the table through {@link TableMetadata.Builder}. The
- * builder recomputes the row-id high-water mark from each snapshot's added-rows, so any id
- * space that does not start at 0 (self-heal and rollback rebuilds) would end below the ids
- * already assigned in manifests, and a later external writer could reuse them. Registration
- * imports every field of the metadata verbatim, next-row-id included.
+ * metadata file, instead of rebuilding the table through {@link TableMetadata.Builder}; see
+ * {@link #requiresRegistration}. Fails before touching the catalog table when the server does
+ * not support registration.
*/
private void registerAsCurrent(
IcebergMetadata adjustedMetadata, TableMetadata newMetadata, boolean dropFirst) {
+ if (!registerTableAdvertised()) {
+ throw registerTableUnsupported(
+ "The Iceberg REST catalog does not advertise registerTable",
+ adjustedMetadata,
+ null);
+ }
try {
- Path registerPath =
- new Path(
- tableLocation,
- String.format(
- "metadata/rest-register-v%d.metadata.json",
- adjustedMetadata.currentSnapshotId()));
- if (!fileIO.tryToWriteAtomic(registerPath, adjustedMetadata.toJson())) {
- fileIO.deleteQuietly(registerPath);
- fileIO.overwriteFileUtf8(registerPath, adjustedMetadata.toJson());
- }
+ Path registerPath = writeRegisterFile(adjustedMetadata);
if (dropFirst) {
+ if (probeRegisterTable(registerPath)) {
+ // the table disappeared concurrently and the probe registered it
+ verifyRegistered(newMetadata);
+ return;
+ }
dropTable();
}
icebergTable =
restCatalog.registerTable(icebergTableIdentifier, registerPath.toString());
- long registered =
- ((BaseTable) icebergTable)
- .operations()
- .current()
- .currentSnapshot()
- .snapshotId();
- if (newMetadata.currentSnapshot() == null
- || registered != newMetadata.currentSnapshot().snapshotId()) {
- throw new IllegalStateException(
- String.format(
- "Registered catalog table is at snapshot %s instead of %s",
- registered,
- newMetadata.currentSnapshot() == null
- ? "null"
- : newMetadata.currentSnapshot().snapshotId()));
- }
+ verifyRegistered(newMetadata);
+ } catch (UnsupportedOperationException e) {
+ throw e;
} catch (Exception e) {
throw new RuntimeException(
"Fail to register iceberg table " + icebergTableIdentifier, e);
}
}
+ /**
+ * Whether the server advertises the register-table endpoint. Iceberg's REST client keeps the
+ * endpoint set from the server's {@code /v1/config} response (or its backwards-compatible
+ * defaults, register-table included, when the server advertises none) in {@code
+ * RESTSessionCatalog#endpoints}, without a public accessor.
+ */
+ private boolean registerTableAdvertised() {
+ try {
+ Field sessionField = RESTCatalog.class.getDeclaredField("sessionCatalog");
+ sessionField.setAccessible(true);
+ Object sessionCatalog = sessionField.get(restCatalog);
+ Field endpointsField = null;
+ for (Class c = sessionCatalog.getClass(); c != null; c = c.getSuperclass()) {
+ try {
+ endpointsField = c.getDeclaredField("endpoints");
+ break;
+ } catch (NoSuchFieldException ignored) {
+ // keep looking in the superclass
+ }
+ }
+ if (endpointsField == null) {
+ throw new NoSuchFieldException("endpoints");
+ }
+ endpointsField.setAccessible(true);
+ Set endpoints = (Set) endpointsField.get(sessionCatalog);
+ return endpoints.contains(Endpoint.V1_REGISTER_TABLE);
+ } catch (Exception | LinkageError e) {
+ LOG.warn(
+ "Cannot read the endpoints advertised by the Iceberg REST catalog, "
+ + "assuming registerTable is unsupported.",
+ e);
+ return false;
+ }
+ }
+
+ /**
+ * A server that advertises no endpoint list gets the client's default list, register-table
+ * included, so {@link #registerTableAdvertised} alone cannot prove support. Before the existing
+ * table is dropped, the registration is attempted against it: a server that implements the
+ * endpoint rejects it with {@link AlreadyExistsException}; anything else means the endpoint is
+ * unavailable and the existing table is left untouched.
+ *
+ * @return true if the registration went through because the table no longer existed
+ */
+ private boolean probeRegisterTable(Path registerPath) {
+ try {
+ icebergTable =
+ restCatalog.registerTable(icebergTableIdentifier, registerPath.toString());
+ return true;
+ } catch (AlreadyExistsException e) {
+ return false;
+ } catch (RuntimeException e) {
+ throw registerTableUnsupported(
+ "registerTable against the existing Iceberg REST catalog table failed with "
+ + "something other than AlreadyExists, so the endpoint is assumed "
+ + "unsupported",
+ null,
+ e);
+ }
+ }
+
+ private UnsupportedOperationException registerTableUnsupported(
+ String reason, @Nullable IcebergMetadata metadata, @Nullable Throwable cause) {
+ return new UnsupportedOperationException(
+ String.format(
+ "%s; registerTable is required to publish format version 3 metadata "
+ + "whose row-id space does not start at 0%s. The catalog table "
+ + "%s was left untouched.",
+ reason,
+ metadata == null
+ ? ""
+ : String.format(
+ " (current snapshot first-row-id %s, next-row-id %s)",
+ metadata.currentSnapshot().firstRowId(),
+ metadata.nextRowId()),
+ icebergTableIdentifier),
+ cause);
+ }
+
+ /**
+ * Writes the metadata to register into a fresh, never-overwritten file. A catalog may keep
+ * referencing the registered location, and rollback and rebuild paths reuse Paimon snapshot ids
+ * for different timelines, so the name carries a UUID and an existing file is never replaced.
+ */
+ private Path writeRegisterFile(IcebergMetadata metadata) throws IOException {
+ Path registerPath =
+ new Path(
+ metadataDirectory,
+ String.format(
+ "rest-register-v%d-%s.metadata.json",
+ metadata.currentSnapshotId(), UUID.randomUUID()));
+ if (!fileIO.tryToWriteAtomic(registerPath, metadata.toJson())) {
+ throw new IOException("Metadata file to register already exists: " + registerPath);
+ }
+ return registerPath;
+ }
+
+ private void verifyRegistered(TableMetadata newMetadata) {
+ long registered =
+ ((BaseTable) icebergTable).operations().current().currentSnapshot().snapshotId();
+ if (newMetadata.currentSnapshot() == null
+ || registered != newMetadata.currentSnapshot().snapshotId()) {
+ throw new IllegalStateException(
+ String.format(
+ "Registered catalog table is at snapshot %s instead of %s",
+ registered,
+ newMetadata.currentSnapshot() == null
+ ? "null"
+ : newMetadata.currentSnapshot().snapshotId()));
+ }
+ }
+
private Table createTable(TableMetadata newMetadata) {
/*
Handles fieldId incompatibility between Paimon (starts at 0) and Iceberg (starts at 1).
@@ -475,6 +603,15 @@ static String toRestLocation(String location) {
return location;
}
+ private Table recreateTable(TableMetadata newMetadata) {
+ try {
+ dropTable();
+ return createTable(newMetadata);
+ } catch (Exception e) {
+ throw new RuntimeException("Fail to recreate iceberg table.", e);
+ }
+ }
+
private Table getTable() {
return restCatalog.loadTable(icebergTableIdentifier);
}
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 b104e5ca2a57..7fd3e5e17065 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
@@ -58,6 +58,7 @@
import org.apache.iceberg.hadoop.HadoopCatalog;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.rest.Endpoint;
import org.apache.iceberg.rest.RESTCatalog;
import org.apache.iceberg.rest.RESTCatalogServer;
import org.apache.iceberg.rest.RESTServerExtension;
@@ -72,9 +73,11 @@
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.BiFunction;
@@ -83,6 +86,7 @@
import static org.apache.paimon.iceberg.IcebergCommitCallback.catalogTableMetadataPath;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link IcebergRestMetadataCommitter}. */
public class IcebergRestMetadataCommitterTest {
@@ -220,15 +224,8 @@ public void testPartitionedPrimaryKeyTable() throws Exception {
expected,
Record::toString);
- // registration imports the real partition spec even when a partition column has
- // field id 0 (the create-path fallback used to register an unpartitioned spec)
- PartitionSpec registeredSpec =
- restCatalog.loadTable(TableIdentifier.of("mydb", "t")).spec();
- assertThat(registeredSpec.fields()).hasSize(2);
- assertThat(registeredSpec.fields().get(0).name()).isEqualTo("pt1");
- assertThat(registeredSpec.fields().get(0).sourceId()).isEqualTo(0);
- assertThat(registeredSpec.fields().get(1).name()).isEqualTo("pt2");
- assertThat(registeredSpec.fields().get(1).sourceId()).isEqualTo(1);
+ PartitionSpec expectedPartitionSpec = PartitionSpec.builderFor(new Schema()).build();
+ runPartitionSpecCompatibilityTest(expectedPartitionSpec);
}
@Test
@@ -419,9 +416,9 @@ public void testSchemaAndPropertiesChange() throws Exception {
Table icebergTable = restCatalog.loadTable(TableIdentifier.of("mydb", "t"));
assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(5);
- // the registered metadata plus later commits' history
+ // 1 metadata for createTable + 4 history metadata
assertThat(((BaseTable) icebergTable).operations().current().previousFiles().size())
- .isEqualTo(4);
+ .isEqualTo(5);
write.close();
commit.close();
@@ -641,11 +638,11 @@ public void testSchemaEvolutionWithInterleavedOptionAlter() throws Exception {
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).
- // The registered table carries exactly the 3 unique field-sets: registration imports
- // the deduplicated local schemas verbatim, with no placeholder schema.
+ // 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(3); // 4 without dedup
+ assertThat(icebergTable.schemas().size()).isEqualTo(4); // 5 without dedup
ctx.close();
}
@@ -1412,6 +1409,226 @@ public void testRecreateWithNonZeroLineageWatermark() throws Exception {
}
}
+ /**
+ * Some REST catalogs (AWS Glue's Iceberg REST endpoint, for one) implement create, load, update
+ * and delete but not registerTable. Simulated by removing the endpoint from the set the client
+ * took from the server's config response.
+ */
+ @Test
+ public void testWithoutRegisterTableEndpoint() throws Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"});
+ Map customOptions = new HashMap<>();
+ customOptions.put(IcebergOptions.FORMAT_VERSION.key(), "3");
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ -1,
+ "avro",
+ customOptions);
+ // Iceberg metadata is only produced locally; the committer under test publishes it
+ FileStoreTable localTable =
+ table.copy(
+ Collections.singletonMap(
+ IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "table-location"));
+ TableIdentifier identifier = TableIdentifier.of("mydb", "t");
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl write = localTable.newWrite(commitUser);
+ TableCommitImpl commit = localTable.newCommit(commitUser);
+ write.write(GenericRow.of(1, 10));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(1, write.prepareCommit(true, 1));
+ write.write(GenericRow.of(3, 30));
+ commit.commit(2, write.prepareCommit(true, 2));
+ IcebergMetadata v1 = localMetadata(localTable, 1);
+ IcebergMetadata v2 = localMetadata(localTable, 2);
+
+ IcebergRestMetadataCommitter committer = new IcebergRestMetadataCommitter(table);
+ removeRegisterTableEndpoint(committer);
+
+ // zero-based v3 metadata publishes through the create/update path every catalog has
+ assertThat(v1.currentSnapshot().firstRowId()).isEqualTo(0L);
+ committer.commitMetadata(v1, null);
+ committer.commitMetadata(v2, v1);
+ Table icebergTable = restCatalog.loadTable(identifier);
+ assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(2);
+ assertThat(TableUtil.formatVersion(icebergTable)).isEqualTo(3);
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder("Record(1, 10)", "Record(2, 20)", "Record(3, 30)");
+
+ // a rebuild whose row-id space does not start at 0 can only be published by
+ // registration: fail closed and leave the catalog table untouched
+ write.write(GenericRow.of(4, 40));
+ commit.commit(3, write.prepareCommit(true, 3));
+ IcebergMetadata v3 = localMetadata(localTable, 3);
+ assertThat(v3.currentSnapshot().firstRowId()).isEqualTo(3L);
+ assertThatThrownBy(() -> committer.commitMetadata(v3, null))
+ .hasStackTraceContaining("does not advertise registerTable");
+ assertThat(restCatalog.loadTable(identifier).currentSnapshot().snapshotId()).isEqualTo(2);
+ assertThat(registerFiles(localTable)).isEmpty();
+
+ // the same metadata is a plain update when the base is correct
+ committer.commitMetadata(v3, v2);
+ assertThat(restCatalog.loadTable(identifier).currentSnapshot().snapshotId()).isEqualTo(3);
+
+ // recreating a lost table needs registration as well: nothing is created
+ restCatalog.dropTable(identifier, false);
+ write.write(GenericRow.of(5, 50));
+ commit.commit(4, write.prepareCommit(true, 4));
+ IcebergMetadata v4 = localMetadata(localTable, 4);
+ assertThatThrownBy(() -> committer.commitMetadata(v4, v3))
+ .hasStackTraceContaining("does not advertise registerTable");
+ assertThat(restCatalog.tableExists(identifier)).isFalse();
+
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ public void testRecreateWithoutRegisterTableEndpointOnV2() throws Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"});
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ -1,
+ "avro",
+ Collections.emptyMap());
+ FileStoreTable localTable =
+ table.copy(
+ Collections.singletonMap(
+ IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "table-location"));
+ TableIdentifier identifier = TableIdentifier.of("mydb", "t");
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl write = localTable.newWrite(commitUser);
+ TableCommitImpl commit = localTable.newCommit(commitUser);
+ write.write(GenericRow.of(1, 10));
+ commit.commit(1, write.prepareCommit(true, 1));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(2, write.prepareCommit(true, 2));
+ write.close();
+ commit.close();
+ IcebergMetadata v1 = localMetadata(localTable, 1);
+ IcebergMetadata v2 = localMetadata(localTable, 2);
+
+ IcebergRestMetadataCommitter committer = new IcebergRestMetadataCommitter(table);
+ removeRegisterTableEndpoint(committer);
+
+ committer.commitMetadata(v1, null);
+ assertThat(restCatalog.loadTable(identifier).currentSnapshot().snapshotId()).isEqualTo(1);
+ // no base: the table is dropped and recreated, which needs no registration on v2
+ committer.commitMetadata(v2, null);
+ Table icebergTable = restCatalog.loadTable(identifier);
+ assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(2);
+ assertThat(TableUtil.formatVersion(icebergTable)).isEqualTo(2);
+ assertThat(getIcebergResult()).containsExactlyInAnyOrder("Record(1, 10)", "Record(2, 20)");
+ assertThat(registerFiles(localTable)).isEmpty();
+ }
+
+ /**
+ * A catalog may keep referencing the registered metadata location, and rollback and rebuild
+ * paths reuse Paimon snapshot ids, so every registration writes a new file and never replaces
+ * an existing one.
+ */
+ @Test
+ public void testRegisteredMetadataFilesAreWriteOnce() throws Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"});
+ Map customOptions = new HashMap<>();
+ customOptions.put(IcebergOptions.FORMAT_VERSION.key(), "3");
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ -1,
+ "avro",
+ customOptions);
+ TableIdentifier identifier = TableIdentifier.of("mydb", "t");
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl write = table.newWrite(commitUser);
+ TableCommitImpl commit = table.newCommit(commitUser);
+ write.write(GenericRow.of(1, 10));
+ commit.commit(1, write.prepareCommit(true, 1));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(2, write.prepareCommit(true, 2));
+ IcebergMetadata v2 = localMetadata(table, 2);
+ assertThat(registerFiles(table)).isEmpty();
+
+ // the lost table is recreated by registration
+ restCatalog.dropTable(identifier, false);
+ write.write(GenericRow.of(3, 30));
+ commit.commit(3, write.prepareCommit(true, 3));
+ write.close();
+ commit.close();
+ IcebergMetadata v3 = localMetadata(table, 3);
+ assertThat(restCatalog.loadTable(identifier).currentSnapshot().snapshotId()).isEqualTo(3);
+ List firstRegistration = registerFiles(table);
+ assertThat(firstRegistration).hasSize(1);
+ assertThat(firstRegistration.get(0).getName()).startsWith("rest-register-v3-");
+ String firstContent = table.fileIO().readFileUtf8(firstRegistration.get(0));
+
+ // registering the same snapshot id again writes a second file, the first is untouched
+ restCatalog.dropTable(identifier, false);
+ new IcebergRestMetadataCommitter(table).commitMetadata(v3, v2);
+ assertThat(restCatalog.loadTable(identifier).currentSnapshot().snapshotId()).isEqualTo(3);
+ List registrations = registerFiles(table);
+ assertThat(registrations).hasSize(2).contains(firstRegistration.get(0));
+ assertThat(table.fileIO().readFileUtf8(firstRegistration.get(0))).isEqualTo(firstContent);
+ for (Path path : registrations) {
+ assertThat(path.getName()).startsWith("rest-register-v3-");
+ assertThat(IcebergMetadata.fromPath(table.fileIO(), path).currentSnapshotId())
+ .isEqualTo(3);
+ }
+ }
+
+ private static IcebergMetadata localMetadata(FileStoreTable table, long snapshotId) {
+ return IcebergMetadata.fromPath(
+ table.fileIO(),
+ new Path(
+ catalogTableMetadataPath(table),
+ String.format("v%d.metadata.json", snapshotId)));
+ }
+
+ private static List registerFiles(FileStoreTable table) throws Exception {
+ List files = new ArrayList<>();
+ for (org.apache.paimon.fs.FileStatus status :
+ table.fileIO().listStatus(catalogTableMetadataPath(table))) {
+ if (status.getPath().getName().startsWith("rest-register-")) {
+ files.add(status.getPath());
+ }
+ }
+ return files;
+ }
+
+ /** Makes the committer's REST client see a server that does not advertise registerTable. */
+ private static void removeRegisterTableEndpoint(IcebergRestMetadataCommitter committer)
+ throws Exception {
+ java.lang.reflect.Field catalogField =
+ IcebergRestMetadataCommitter.class.getDeclaredField("restCatalog");
+ catalogField.setAccessible(true);
+ Object restCatalog = catalogField.get(committer);
+ java.lang.reflect.Field sessionField = RESTCatalog.class.getDeclaredField("sessionCatalog");
+ sessionField.setAccessible(true);
+ Object sessionCatalog = sessionField.get(restCatalog);
+ java.lang.reflect.Field endpointsField =
+ sessionCatalog.getClass().getDeclaredField("endpoints");
+ endpointsField.setAccessible(true);
+ Set endpoints = new HashSet<>((Set) endpointsField.get(sessionCatalog));
+ assertThat(endpoints.remove(Endpoint.V1_REGISTER_TABLE)).isTrue();
+ endpointsField.set(sessionCatalog, endpoints);
+ }
+
private static class TestRecord {
private final BinaryRow partition;
private final GenericRow record;
From c4950469d76b1eca9c0d23b5956441af4821fdab Mon Sep 17 00:00:00 2001
From: Victor Babenko <37556649+vbabenkoru@users.noreply.github.com>
Date: Sat, 15 Aug 2026 22:01:44 -0700
Subject: [PATCH 15/15] [iceberg] Allow publishing VARIANT columns with format
version 3
---
.../paimon/iceberg/IcebergCommitCallback.java | 14 +++---
.../IcebergRowLineageCompatibilityTest.java | 48 +++++++++++++++++++
2 files changed, 56 insertions(+), 6 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
index f8a41f56f28e..ea6d256a4efb 100644
--- a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
+++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
@@ -733,7 +733,7 @@ private List getPartitionFields(
return result;
}
- /** VARIANT needs Iceberg row lineage, which Paimon Iceberg compatibility cannot publish. */
+ /** VARIANT is an Iceberg format-version-3 type; reject publishing it into v2 metadata. */
static void checkVariantNotPublishable(RowType rowType) {
Collection variantFields = new LinkedHashSet<>();
for (DataField field : rowType.getFields()) {
@@ -741,9 +741,8 @@ static void checkVariantNotPublishable(RowType rowType) {
}
Preconditions.checkArgument(
variantFields.isEmpty(),
- "Columns %s use the VARIANT type, which Paimon Iceberg compatibility cannot "
- + "publish: it is an Iceberg format-version-3 type that requires row "
- + "lineage.",
+ "Columns %s use the VARIANT type, which requires Iceberg format version 3. "
+ + "Set 'metadata.iceberg.format-version' = '3' to publish this table.",
variantFields);
}
@@ -2136,8 +2135,11 @@ private IcebergSchema get(long schemaId) {
schemaId,
id -> {
TableSchema schema = schemaManager.schema(id);
- // backstop: reject variant on each schema as it is emitted
- checkVariantNotPublishable(schema.logicalRowType());
+ if (formatVersion < IcebergMetadata.FORMAT_VERSION_V3) {
+ // VARIANT is an Iceberg format-version-3 type; v2 metadata cannot
+ // represent it
+ checkVariantNotPublishable(schema.logicalRowType());
+ }
SchemaValidation.validateIcebergGeospatialTypes(
schema.logicalRowType(), table.coreOptions());
return IcebergSchema.create(schema);
diff --git a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java
index 55a1eacb665d..b10cb345d19d 100644
--- a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java
+++ b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java
@@ -23,6 +23,7 @@
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.variant.GenericVariant;
import org.apache.paimon.disk.IOManagerImpl;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.SeekableInputStream;
@@ -76,6 +77,7 @@
import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for Iceberg format-version 3 row-lineage metadata fields. */
public class IcebergRowLineageCompatibilityTest {
@@ -234,6 +236,52 @@ public void testTagPreservesNextRowId() throws Exception {
assertThat(metadata.nextRowId()).isEqualTo(3L);
}
+ @Test
+ public void testVariantPublishableWithFormatVersion3() throws Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.VARIANT()},
+ new String[] {"k", "payload"});
+ FileStoreTable table = createPaimonTable(rowType, formatVersionOptions(3), "parquet");
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl write =
+ table.newWrite(commitUser)
+ .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp"));
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(1, GenericVariant.fromJson("{\"a\": 1}")));
+ commit.commit(1, write.prepareCommit(false, 1));
+ write.close();
+ commit.close();
+
+ IcebergMetadata metadata = readIcebergMetadata(table, 1);
+ assertThat(metadata.nextRowId()).isEqualTo(1L);
+ assertThat(metadata.schemas().get(metadata.currentSchemaId()).fields().get(1).type())
+ .isEqualTo("variant");
+ }
+
+ @Test
+ public void testVariantRejectedWithFormatVersion2() throws Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.VARIANT()},
+ new String[] {"k", "payload"});
+ FileStoreTable table = createPaimonTable(rowType, formatVersionOptions(2), "parquet");
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl write =
+ table.newWrite(commitUser)
+ .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp"));
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(1, GenericVariant.fromJson("{\"a\": 1}")));
+ // hasStackTraceContaining: robust whether or not the commit path wraps the
+ // IllegalArgumentException from the guard
+ assertThatThrownBy(() -> commit.commit(1, write.prepareCommit(false, 1)))
+ .hasStackTraceContaining("VARIANT");
+ write.close();
+ commit.close();
+ }
+
@Test
public void testManifestListCarriesFirstRowIdColumn() throws Exception {
FileStoreTable table = createPaimonTable(defaultRowType(), formatVersionOptions(3), "avro");