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 d187f675d0 [iceberg] Assign manifest-level row lineage for Iceberg
format version 3 (#9245)
d187f675d0 is described below
commit d187f675d04ce5924beee04d0ba9d821dca79de1
Author: Victor Babenko <[email protected]>
AuthorDate: Wed Aug 19 19:56:46 2026 -0700
[iceberg] Assign manifest-level row lineage for Iceberg format version 3
(#9245)
---
.github/workflows/utitcase-iceberg-ga.yml | 66 ++
.../paimon/iceberg/IcebergCommitCallback.java | 181 +++-
.../iceberg/manifest/IcebergDataFileMeta.java | 44 +-
.../manifest/IcebergDataFileMetaSerializer.java | 27 +-
.../iceberg/manifest/IcebergManifestEntry.java | 13 +-
.../manifest/IcebergManifestEntrySerializer.java | 8 +-
.../iceberg/manifest/IcebergManifestFile.java | 14 +-
.../iceberg/manifest/IcebergManifestFileMeta.java | 49 +-
.../IcebergManifestFileMetaSerializer.java | 30 +-
.../iceberg/manifest/IcebergManifestList.java | 5 +-
paimon-iceberg/pom.xml | 94 ++-
.../iceberg/IcebergRestMetadataCommitter.java | 214 ++++-
.../core/IcebergRowLineageCompatibilityTest.java | 917 +++++++++++++++++++++
.../apache/paimon/iceberg/IcebergMetadataTest.java | 16 +-
.../iceberg/IcebergRestMetadataCommitterTest.java | 310 +++++++
15 files changed, 1914 insertions(+), 74 deletions(-)
diff --git a/.github/workflows/utitcase-iceberg-ga.yml
b/.github/workflows/utitcase-iceberg-ga.yml
new file mode 100644
index 0000000000..deca2434fc
--- /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-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
index 42e6ce2a8f..f8a41f56f2 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
@@ -538,6 +538,16 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
metrics.totalPositionDeletes = totalPositionDeleteRecords;
metrics.totalEqualityDeletes = 0;
+ // a rebuild replaces metadata whose ids are already out with readers:
never reuse them
+ Long snapshotFirstRowId = computeSnapshotFirstRowId(nextRowIdFloor);
+ ManifestRowIdAssignment rowIdAssignment =
+ assignManifestFirstRowIds(allManifestFileMetas,
snapshotFirstRowId);
+ allManifestFileMetas = rowIdAssignment.manifests;
+ Long addedRows = snapshotFirstRowId == null ? null :
rowIdAssignment.assignedRows;
+ Long nextRowId =
+ snapshotFirstRowId == null
+ ? null
+ : snapshotFirstRowId + rowIdAssignment.assignedRows;
String manifestListFileName =
manifestList.writeWithoutRolling(allManifestFileMetas);
// current schema follows the latest; the snapshot entry records its
own schema
@@ -551,8 +561,6 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
computeSnapshotSummary(
IcebergSnapshotSummary.APPEND.operation(),
paimonSnapshot, metrics);
- // a rebuild replaces metadata whose ids are already out with readers:
never reuse them
- RowLineage rowLineage = computeRowLineage(nextRowIdFloor,
metrics.addedRecords);
IcebergSnapshot snapshot =
new IcebergSnapshot(
snapshotId,
@@ -563,8 +571,8 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
snapshotSummary,
pathFactory.toManifestListPath(manifestListFileName).toString(),
snapshotSchemaId,
- rowLineage.firstRowId,
- rowLineage.addedRows);
+ snapshotFirstRowId,
+ addedRows);
// Tags can only be included in Iceberg if they point to an Iceberg
snapshot that
// exists. Otherwise, an Iceberg client fails to parse the metadata
and all reads fail.
@@ -607,7 +615,7 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
IcebergPartitionField.FIRST_FIELD_ID -
1),
Collections.singletonList(snapshot),
(int) snapshotId,
- rowLineage.nextRowId,
+ nextRowId,
refs);
Path metadataPath = pathFactory.toMetadataPath(snapshotId);
@@ -1086,13 +1094,6 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
// compact data manifest file if needed
newDataManifestFileMetas =
compactMetadataIfNeeded(newDataManifestFileMetas, snapshotId);
- String manifestListFileName =
- manifestList.writeWithoutRolling(
- Stream.concat(
- newDataManifestFileMetas.stream(),
- newDVManifestFileMetas.stream())
- .collect(Collectors.toList()));
-
SummaryMetrics metrics = new SummaryMetrics();
metrics.addedDataFiles = addedFiles.size();
metrics.addedRecords =
@@ -1143,6 +1144,24 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
metrics.totalPositionDeletes =
computeLiveRowCount(newDVManifestFileMetas);
metrics.totalEqualityDeletes = 0;
+ Long snapshotFirstRowId = computeSnapshotFirstRowId(rowIdFloor);
+
+ ManifestRowIdAssignment rowIdAssignment =
+ assignManifestFirstRowIds(
+ Stream.concat(
+ newDataManifestFileMetas.stream(),
+ newDVManifestFileMetas.stream())
+ .collect(Collectors.toList()),
+ snapshotFirstRowId);
+ List<IcebergManifestFileMeta> newManifestFileMetasWithRowIds =
rowIdAssignment.manifests;
+ Long addedRows = snapshotFirstRowId == null ? null :
rowIdAssignment.assignedRows;
+ Long nextRowId =
+ snapshotFirstRowId == null
+ ? null
+ : snapshotFirstRowId + rowIdAssignment.assignedRows;
+ String manifestListFileName =
+
manifestList.writeWithoutRolling(newManifestFileMetasWithRowIds);
+
IcebergSnapshotSummary snapshotSummary =
computeSnapshotSummary(operation, snapshot, metrics);
@@ -1161,8 +1180,6 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
}
// a schema-pointer rollback (validated above): only the current
pointer moves
- RowLineage rowLineage = computeRowLineage(rowIdFloor,
metrics.addedRecords);
-
List<IcebergSnapshot> snapshots = new
ArrayList<>(baseMetadata.snapshots());
snapshots.add(
new IcebergSnapshot(
@@ -1175,8 +1192,8 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
pathFactory.toManifestListPath(manifestListFileName).toString(),
// the snapshot's own schema, for time travel
snapshotSchemaId,
- rowLineage.firstRowId,
- rowLineage.addedRows));
+ snapshotFirstRowId,
+ addedRows));
// all snapshots in this list, except the last one, need to expire
List<IcebergSnapshot> toExpireExceptLast = new ArrayList<>();
@@ -1221,7 +1238,7 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
baseMetadata.lastPartitionId(),
snapshots,
(int) snapshotId,
- rowLineage.nextRowId,
+ nextRowId,
refs);
Path metadataPath = pathFactory.toMetadataPath(snapshotId);
@@ -1426,8 +1443,10 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
commitKind == Snapshot.CommitKind.COMPACT
?
IcebergSnapshotSummary.REPLACE.operation()
:
IcebergSnapshotSummary.OVERWRITE.operation();
+ List<IcebergManifestEntry> sourceEntries =
+ materializeFirstRowIds(fileMeta, entries);
List<IcebergManifestEntry> newEntries = new ArrayList<>();
- for (IcebergManifestEntry entry : entries) {
+ for (IcebergManifestEntry entry : sourceEntries) {
if (entry.isLive()) {
boolean removed =
removedFiles.containsKey(entry.file().filePath());
newEntries.add(
@@ -1489,10 +1508,13 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
Function<IcebergManifestFileMeta, List<IcebergManifestEntry>>
processor =
meta -> {
+ List<IcebergManifestEntry> sourceEntries =
+ materializeFirstRowIds(
+ meta,
+ IcebergManifestFile.create(table,
pathFactory)
+ .read(new
Path(meta.manifestPath()).getName()));
List<IcebergManifestEntry> entries = new ArrayList<>();
- for (IcebergManifestEntry entry :
- IcebergManifestFile.create(table, pathFactory)
- .read(new
Path(meta.manifestPath()).getName())) {
+ for (IcebergManifestEntry entry : sourceEntries) {
// a deletion made by this commit is recorded against
the current
// snapshot but keeps the file sequence number of the
older snapshot
// that added the file, so it has to be recognised by
snapshot id
@@ -1553,9 +1575,14 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
}
private void expireManifestList(String toExpire, String next) {
- Set<IcebergManifestFileMeta> 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<String> 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()));
@@ -2001,24 +2028,102 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
/**
* Row-lineage bookkeeping for a new snapshot, mandatory in Iceberg format
version 3: the
- * snapshot's first-row-id starts at the base metadata's next-row-id
watermark and the table's
- * next-row-id advances by the snapshot's added records. For format
version 2 all fields stay
- * null so nothing is written.
+ * snapshot's first-row-id starts at the base metadata's next-row-id
watermark. The snapshot's
+ * added-rows and the table's next-row-id are NOT derived here: they
depend on how many rows
+ * {@link #assignManifestFirstRowIds} actually assigns (which can exceed
this commit's added
+ * records when a carried-over manifest is assigned for the first time,
e.g. a Layer-1-written
+ * manifest being upgraded), so callers must recompute them from the
assignment's result. For
+ * format version 2 the field stays null so nothing is written.
+ */
+ @Nullable
+ private Long computeSnapshotFirstRowId(long baseNextRowId) {
+ return formatVersion >= IcebergMetadata.FORMAT_VERSION_V3 ?
baseNextRowId : null;
+ }
+
+ /**
+ * Result of {@link #assignManifestFirstRowIds}: the manifests with
first_row_id assigned, and
+ * the total number of rows actually consumed from the row-id space by
that assignment (which
+ * may be larger than this commit's added-records count; see the
class-level note there).
+ */
+ private static class ManifestRowIdAssignment {
+ private final List<IcebergManifestFileMeta> manifests;
+ private final long assignedRows;
+
+ private ManifestRowIdAssignment(
+ List<IcebergManifestFileMeta> 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 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}.
+ *
+ * <p>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 RowLineage computeRowLineage(long baseNextRowId, long
addedRecords) {
- RowLineage lineage = new RowLineage();
- if (formatVersion >= IcebergMetadata.FORMAT_VERSION_V3) {
- lineage.firstRowId = baseNextRowId;
- lineage.addedRows = addedRecords;
- lineage.nextRowId = baseNextRowId + addedRecords;
- }
- return lineage;
+ private ManifestRowIdAssignment assignManifestFirstRowIds(
+ List<IcebergManifestFileMeta> manifests, @Nullable Long
snapshotFirstRowId) {
+ if (snapshotFirstRowId == null) {
+ return new ManifestRowIdAssignment(manifests, 0L);
+ }
+ List<IcebergManifestFileMeta> result = new ArrayList<>();
+ long watermark = snapshotFirstRowId;
+ for (IcebergManifestFileMeta meta : manifests) {
+ if (meta.content() == IcebergManifestFileMeta.Content.DATA
+ && meta.firstRowId() == null) {
+ result.add(meta.withFirstRowId(watermark));
+ // 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);
+ }
+ }
+ return new ManifestRowIdAssignment(result, watermark -
snapshotFirstRowId);
}
- private static class RowLineage {
- @Nullable private Long firstRowId;
- @Nullable private Long addedRows;
- @Nullable private Long nextRowId;
+ /**
+ * 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, 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<IcebergManifestEntry> materializeFirstRowIds(
+ IcebergManifestFileMeta baseMeta, List<IcebergManifestEntry>
entries) {
+ if (baseMeta.content() != IcebergManifestFileMeta.Content.DATA
+ || baseMeta.firstRowId() == null) {
+ return entries;
+ }
+ List<IcebergManifestEntry> result = new ArrayList<>();
+ long watermark = baseMeta.firstRowId();
+ for (IcebergManifestEntry entry : entries) {
+ 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);
+ }
+ }
+ return result;
}
private class SchemaCache {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMeta.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMeta.java
index 950da63b9d..f9ce30bad4 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMeta.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMeta.java
@@ -87,6 +87,7 @@ public class IcebergDataFileMeta {
private final String referencedDataFile;
private final Long contentOffset;
private final Long contentSizeInBytes;
+ @Nullable private final Long firstRowId;
IcebergDataFileMeta(
Content content,
@@ -110,6 +111,7 @@ public class IcebergDataFileMeta {
upperBounds,
null,
null,
+ null,
null);
}
@@ -125,7 +127,8 @@ public class IcebergDataFileMeta {
InternalMap upperBounds,
String referencedDataFile,
Long contentOffset,
- Long contentSizeInBytes) {
+ Long contentSizeInBytes,
+ @Nullable Long firstRowId) {
this.content = content;
this.filePath = filePath;
this.fileFormat = fileFormat;
@@ -139,6 +142,7 @@ public class IcebergDataFileMeta {
this.referencedDataFile = referencedDataFile;
this.contentOffset = contentOffset;
this.contentSizeInBytes = contentSizeInBytes;
+ this.firstRowId = firstRowId;
}
public static IcebergDataFileMeta create(
@@ -245,7 +249,8 @@ public class IcebergDataFileMeta {
null,
referencedDataFile,
contentOffset,
- contentSizeInBytes);
+ contentSizeInBytes,
+ null);
}
public Content content() {
@@ -296,7 +301,33 @@ public class IcebergDataFileMeta {
return contentSizeInBytes;
}
+ @Nullable
+ public Long firstRowId() {
+ return firstRowId;
+ }
+
+ public IcebergDataFileMeta withFirstRowId(long firstRowId) {
+ return new IcebergDataFileMeta(
+ content,
+ filePath,
+ fileFormat,
+ partition,
+ recordCount,
+ fileSizeInBytes,
+ nullValueCounts,
+ lowerBounds,
+ upperBounds,
+ referencedDataFile,
+ contentOffset,
+ contentSizeInBytes,
+ firstRowId);
+ }
+
public static RowType schema(RowType partitionType) {
+ return schema(partitionType, false);
+ }
+
+ public static RowType schema(RowType partitionType, boolean
withFirstRowId) {
List<DataField> fields = new ArrayList<>();
fields.add(new DataField(134, "content", DataTypes.INT().notNull()));
fields.add(new DataField(100, "file_path",
DataTypes.STRING().notNull()));
@@ -322,6 +353,9 @@ public class IcebergDataFileMeta {
fields.add(new DataField(143, "referenced_data_file",
DataTypes.STRING()));
fields.add(new DataField(144, "content_offset", DataTypes.BIGINT()));
fields.add(new DataField(145, "content_size_in_bytes",
DataTypes.BIGINT()));
+ if (withFirstRowId) {
+ fields.add(new DataField(142, "first_row_id", DataTypes.BIGINT()));
+ }
return new RowType(false, fields);
}
@@ -345,7 +379,8 @@ public class IcebergDataFileMeta {
&& Objects.equals(upperBounds, that.upperBounds)
&& Objects.equals(referencedDataFile, that.referencedDataFile)
&& Objects.equals(contentOffset, that.contentOffset)
- && Objects.equals(contentSizeInBytes, that.contentSizeInBytes);
+ && Objects.equals(contentSizeInBytes, that.contentSizeInBytes)
+ && Objects.equals(firstRowId, that.firstRowId);
}
@Override
@@ -362,6 +397,7 @@ public class IcebergDataFileMeta {
upperBounds,
referencedDataFile,
contentOffset,
- contentSizeInBytes);
+ contentSizeInBytes,
+ firstRowId);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaSerializer.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaSerializer.java
index 020b5e1b44..5db5dea27d 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaSerializer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaSerializer.java
@@ -36,18 +36,40 @@ public class IcebergDataFileMetaSerializer extends
ObjectSerializer<IcebergDataF
private final InternalMapSerializer nullValueCountsSerializer;
private final InternalMapSerializer lowerBoundsSerializer;
private final InternalMapSerializer upperBoundsSerializer;
+ private final boolean withFirstRowId;
public IcebergDataFileMetaSerializer(RowType partitionType) {
- super(IcebergDataFileMeta.schema(partitionType));
+ this(partitionType, false);
+ }
+
+ public IcebergDataFileMetaSerializer(RowType partitionType, boolean
withFirstRowId) {
+ super(IcebergDataFileMeta.schema(partitionType, withFirstRowId));
this.partSerializer = new InternalRowSerializer(partitionType);
this.nullValueCountsSerializer =
new InternalMapSerializer(DataTypes.INT(), DataTypes.BIGINT());
this.lowerBoundsSerializer = new
InternalMapSerializer(DataTypes.INT(), DataTypes.BYTES());
this.upperBoundsSerializer = new
InternalMapSerializer(DataTypes.INT(), DataTypes.BYTES());
+ this.withFirstRowId = withFirstRowId;
}
@Override
public InternalRow toRow(IcebergDataFileMeta file) {
+ if (withFirstRowId) {
+ return GenericRow.of(
+ file.content().id(),
+ BinaryString.fromString(file.filePath()),
+ BinaryString.fromString(file.fileFormat()),
+ file.partition(),
+ file.recordCount(),
+ file.fileSizeInBytes(),
+ file.nullValueCounts(),
+ file.lowerBounds(),
+ file.upperBounds(),
+ BinaryString.fromString(file.referencedDataFile()),
+ file.contentOffset(),
+ file.contentSizeInBytes(),
+ file.firstRowId());
+ }
return GenericRow.of(
file.content().id(),
BinaryString.fromString(file.filePath()),
@@ -77,6 +99,7 @@ public class IcebergDataFileMetaSerializer extends
ObjectSerializer<IcebergDataF
upperBoundsSerializer.copy(row.getMap(8)),
row.isNullAt(9) ? null : row.getString(9).toString(),
row.isNullAt(10) ? null : row.getLong(10),
- row.isNullAt(11) ? null : row.getLong(11));
+ row.isNullAt(11) ? null : row.getLong(11),
+ withFirstRowId && !row.isNullAt(12) ? row.getLong(12) : null);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestEntry.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestEntry.java
index 40aea8c498..727c6cf570 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestEntry.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestEntry.java
@@ -111,7 +111,16 @@ public class IcebergManifestEntry {
return dataFile;
}
+ public IcebergManifestEntry withFile(IcebergDataFileMeta file) {
+ return new IcebergManifestEntry(
+ status, snapshotId, sequenceNumber, fileSequenceNumber, file);
+ }
+
public static RowType schema(RowType partitionType) {
+ return schema(partitionType, false);
+ }
+
+ public static RowType schema(RowType partitionType, boolean
withFirstRowId) {
RowType icebergPartition = icebergPartitionType(partitionType);
@@ -122,7 +131,9 @@ public class IcebergManifestEntry {
fields.add(new DataField(4, "file_sequence_number",
DataTypes.BIGINT()));
fields.add(
new DataField(
- 2, "data_file",
IcebergDataFileMeta.schema(icebergPartition).notNull()));
+ 2,
+ "data_file",
+ IcebergDataFileMeta.schema(icebergPartition,
withFirstRowId).notNull()));
return new RowType(false, fields);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestEntrySerializer.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestEntrySerializer.java
index b9d2c271b5..75172706a8 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestEntrySerializer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestEntrySerializer.java
@@ -31,8 +31,12 @@ public class IcebergManifestEntrySerializer extends
ObjectSerializer<IcebergMani
private final IcebergDataFileMetaSerializer fileSerializer;
public IcebergManifestEntrySerializer(RowType partitionType) {
- super(IcebergManifestEntry.schema(partitionType));
- this.fileSerializer = new IcebergDataFileMetaSerializer(partitionType);
+ this(partitionType, false);
+ }
+
+ public IcebergManifestEntrySerializer(RowType partitionType, boolean
withFirstRowId) {
+ super(IcebergManifestEntry.schema(partitionType, withFirstRowId));
+ this.fileSerializer = new IcebergDataFileMetaSerializer(partitionType,
withFirstRowId);
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java
index f2f11a697c..5e43a2e2fa 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java
@@ -31,6 +31,7 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.iceberg.IcebergOptions;
import org.apache.paimon.iceberg.IcebergPathFactory;
import org.apache.paimon.iceberg.manifest.IcebergManifestFileMeta.Content;
+import org.apache.paimon.iceberg.metadata.IcebergMetadata;
import org.apache.paimon.iceberg.metadata.IcebergPartitionSpec;
import org.apache.paimon.io.RollingFileWriterImpl;
import org.apache.paimon.io.SingleFileWriter;
@@ -72,6 +73,7 @@ public class IcebergManifestFile extends
ObjectsFile<IcebergManifestEntry> {
public IcebergManifestFile(
FileIO fileIO,
RowType partitionType,
+ boolean withFirstRowId,
FormatReaderFactory readerFactory,
FormatWriterFactory writerFactory,
String compression,
@@ -79,8 +81,8 @@ public class IcebergManifestFile extends
ObjectsFile<IcebergManifestEntry> {
MemorySize targetFileSize) {
super(
fileIO,
- new IcebergManifestEntrySerializer(partitionType),
- IcebergManifestEntry.schema(partitionType),
+ new IcebergManifestEntrySerializer(partitionType,
withFirstRowId),
+ IcebergManifestEntry.schema(partitionType, withFirstRowId),
readerFactory,
writerFactory,
compression,
@@ -98,8 +100,10 @@ public class IcebergManifestFile extends
ObjectsFile<IcebergManifestEntry> {
public static IcebergManifestFile create(FileStoreTable table,
IcebergPathFactory pathFactory) {
RowType partitionType = table.schema().logicalPartitionType();
- RowType entryType = IcebergManifestEntry.schema(partitionType);
Options avroOptions = Options.fromMap(table.options());
+ boolean withFirstRowId =
+ avroOptions.get(IcebergOptions.FORMAT_VERSION) >=
IcebergMetadata.FORMAT_VERSION_V3;
+ RowType entryType = IcebergManifestEntry.schema(partitionType,
withFirstRowId);
//
https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/ManifestReader.java
avroOptions.set(
"avro.row-name-mapping",
@@ -120,6 +124,7 @@ public class IcebergManifestFile extends
ObjectsFile<IcebergManifestEntry> {
return new IcebergManifestFile(
table.fileIO(),
partitionType,
+ withFirstRowId,
manifestFileAvro.createReaderFactory(entryType, entryType, new
ArrayList<>()),
manifestFileAvro.createWriterFactory(entryType),
avroOptions.get(IcebergOptions.MANIFEST_COMPRESSION),
@@ -298,7 +303,8 @@ public class IcebergManifestFile extends
ObjectsFile<IcebergManifestEntry> {
addedRowsCount,
existingRowsCount,
deletedRowsCount,
- partitionSummaries);
+ partitionSummaries,
+ null);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMeta.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMeta.java
index da3e0c2402..d5ac07c962 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMeta.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMeta.java
@@ -22,6 +22,8 @@ import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
@@ -73,6 +75,7 @@ public class IcebergManifestFileMeta {
private final long existingRowsCount;
private final long deletedRowsCount;
private final List<IcebergPartitionSummary> partitions;
+ @Nullable private final Long firstRowId;
public IcebergManifestFileMeta(
String manifestPath,
@@ -88,7 +91,8 @@ public class IcebergManifestFileMeta {
long addedRowsCount,
long existingRowsCount,
long deletedRowsCount,
- List<IcebergPartitionSummary> partitions) {
+ List<IcebergPartitionSummary> partitions,
+ @Nullable Long firstRowId) {
this.manifestPath = manifestPath;
this.manifestLength = manifestLength;
this.partitionSpecId = partitionSpecId;
@@ -103,6 +107,7 @@ public class IcebergManifestFileMeta {
this.existingRowsCount = existingRowsCount;
this.deletedRowsCount = deletedRowsCount;
this.partitions = partitions;
+ this.firstRowId = firstRowId;
}
public String manifestPath() {
@@ -165,8 +170,42 @@ public class IcebergManifestFileMeta {
return partitions;
}
+ @Nullable
+ public Long firstRowId() {
+ return firstRowId;
+ }
+
+ public IcebergManifestFileMeta withFirstRowId(long firstRowId) {
+ return new IcebergManifestFileMeta(
+ manifestPath,
+ manifestLength,
+ partitionSpecId,
+ content,
+ sequenceNumber,
+ minSequenceNumber,
+ addedSnapshotId,
+ addedFilesCount,
+ existingFilesCount,
+ deletedFilesCount,
+ addedRowsCount,
+ existingRowsCount,
+ deletedRowsCount,
+ partitions,
+ firstRowId);
+ }
+
public static RowType schema(boolean legacyVersion) {
- return legacyVersion ? schemaForIceberg1_4() : schemaForIcebergNew();
+ return schema(legacyVersion, false);
+ }
+
+ public static RowType schema(boolean legacyVersion, boolean
withFirstRowId) {
+ RowType base = legacyVersion ? schemaForIceberg1_4() :
schemaForIcebergNew();
+ if (!withFirstRowId) {
+ return base;
+ }
+ List<DataField> fields = new ArrayList<>(base.getFields());
+ fields.add(new DataField(520, "first_row_id", DataTypes.BIGINT()));
+ return new RowType(false, fields);
}
private static RowType schemaForIcebergNew() {
@@ -235,7 +274,8 @@ public class IcebergManifestFileMeta {
&& addedRowsCount == that.addedRowsCount
&& existingRowsCount == that.existingRowsCount
&& deletedRowsCount == that.deletedRowsCount
- && Objects.equals(partitions, that.partitions);
+ && Objects.equals(partitions, that.partitions)
+ && Objects.equals(firstRowId, that.firstRowId);
}
@Override
@@ -254,6 +294,7 @@ public class IcebergManifestFileMeta {
addedRowsCount,
existingRowsCount,
deletedRowsCount,
- partitions);
+ partitions,
+ firstRowId);
}
}
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 2b4c9b771c..14b6bbf91b 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
@@ -36,14 +36,41 @@ public class IcebergManifestFileMetaSerializer extends
ObjectSerializer<IcebergM
private static final long serialVersionUID = 1L;
private final IcebergPartitionSummarySerializer partitionSummarySerializer;
+ private final boolean withFirstRowId;
public IcebergManifestFileMetaSerializer(RowType schema) {
super(schema);
this.partitionSummarySerializer = new
IcebergPartitionSummarySerializer();
+ // IcebergManifestFileMeta.schema(...) always has 14 base fields (see
+ // schemaForIcebergNew/schemaForIceberg1_4) and appends exactly one
extra field,
+ // first_row_id (520), when withFirstRowId=true (see schema(boolean,
boolean)); so a
+ // field count of 15 unambiguously means the row-lineage column is
present.
+ this.withFirstRowId = schema.getFieldCount() == 15;
}
@Override
public InternalRow toRow(IcebergManifestFileMeta file) {
+ if (withFirstRowId) {
+ return GenericRow.of(
+ BinaryString.fromString(file.manifestPath()),
+ file.manifestLength(),
+ file.partitionSpecId(),
+ file.content().id(),
+ file.sequenceNumber(),
+ file.minSequenceNumber(),
+ file.addedSnapshotId(),
+ file.addedFilesCount(),
+ file.existingFilesCount(),
+ file.deletedFilesCount(),
+ file.addedRowsCount(),
+ file.existingRowsCount(),
+ file.deletedRowsCount(),
+ new GenericArray(
+ file.partitions().stream()
+ .map(partitionSummarySerializer::toRow)
+ .toArray(InternalRow[]::new)),
+ file.firstRowId());
+ }
return GenericRow.of(
BinaryString.fromString(file.manifestPath()),
file.manifestLength(),
@@ -80,7 +107,8 @@ public class IcebergManifestFileMetaSerializer extends
ObjectSerializer<IcebergM
row.getLong(10),
row.getLong(11),
row.getLong(12),
- toPartitionSummaries(row.getArray(13)));
+ toPartitionSummaries(row.getArray(13)),
+ withFirstRowId && !row.isNullAt(14) ? row.getLong(14) : null);
}
private List<IcebergPartitionSummary> toPartitionSummaries(InternalArray
array) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestList.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestList.java
index 9eb1194c6f..0439aba087 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestList.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestList.java
@@ -23,6 +23,7 @@ import org.apache.paimon.format.FileFormat;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.iceberg.IcebergOptions;
import org.apache.paimon.iceberg.IcebergPathFactory;
+import org.apache.paimon.iceberg.metadata.IcebergMetadata;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.RowType;
@@ -69,9 +70,11 @@ public class IcebergManifestList extends
ObjectsFile<IcebergManifestFileMeta> {
+ "manifest_file_partitions:r508,"
+ "array_id_r508:508");
FileFormat fileFormat = FileFormat.fromIdentifier("avro", avroOptions);
+ boolean withFirstRowId =
+ avroOptions.get(IcebergOptions.FORMAT_VERSION) >=
IcebergMetadata.FORMAT_VERSION_V3;
RowType manifestType =
IcebergManifestFileMeta.schema(
-
avroOptions.get(IcebergOptions.MANIFEST_LEGACY_VERSION));
+
avroOptions.get(IcebergOptions.MANIFEST_LEGACY_VERSION), withFirstRowId);
return new IcebergManifestList(
table.fileIO(),
fileFormat,
diff --git a/paimon-iceberg/pom.xml b/paimon-iceberg/pom.xml
index 403419d69b..bf02653af2 100644
--- a/paimon-iceberg/pom.xml
+++ b/paimon-iceberg/pom.xml
@@ -38,6 +38,7 @@ under the License.
<iceberg.flink.version>1.19</iceberg.flink.version>
<hive.version>2.3.10</hive.version>
<iceberg.flink.dropwizard.version>1.19.0</iceberg.flink.dropwizard.version>
+ <jetty.server.version>11.0.24</jetty.server.version>
</properties>
<dependencies>
@@ -192,6 +193,14 @@ under the License.
<exclusion>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.eclipse.jetty.websocket</groupId>
+ <artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>
@@ -229,6 +238,14 @@ under the License.
<exclusion>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.eclipse.jetty.websocket</groupId>
+ <artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>
@@ -275,7 +292,7 @@ under the License.
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-server</artifactId>
- <version>11.0.24</version>
+ <version>${jetty.server.version}</version>
<scope>test</scope>
</dependency>
@@ -356,4 +373,79 @@ under the License.
</dependencies>
+
+ <profiles>
+ <profile>
+ <!-- GA row-lineage validation: builds and runs this module
against Iceberg 1.11,
+ 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.
+
+ 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. -->
+ <id>iceberg-ga</id>
+ <properties>
+ <iceberg.version>1.11.0</iceberg.version>
+ <iceberg.flink.version>1.20</iceberg.flink.version>
+
<iceberg.flink.dropwizard.version>1.20.0</iceberg.flink.dropwizard.version>
+ <!-- Iceberg 1.11's REST test-fixtures embed a Jetty 12 server
built on the
+ jetty-compression module; the Jetty 11 jetty-server no
longer provides
+ the classes it needs. -->
+ <jetty.server.version>12.1.8</jetty.server.version>
+ <!-- Iceberg 1.11's iceberg-parquet pulls
parquet-avro/parquet-hadoop 1.17.1,
+ whose HadoopInputFile always calls the
FileSystem#openFile(Path) builder
+ API (Hadoop 3.3+); the reactor-wide hadoop.version
predates it, so this
+ profile pins the Hadoop 3 line Iceberg 1.11 itself builds
against. -->
+ <hadoop.version>3.4.3</hadoop.version>
+ </properties>
+ <dependencies>
+ <!-- iceberg-core 1.11 requires jackson-core/jackson-databind
2.21.x, but
+ Hadoop's older jackson-annotations otherwise wins
dependency mediation,
+ leaving a mismatched pair that fails at runtime. -->
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ <version>2.21</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.eclipse.jetty.ee10</groupId>
+ <artifactId>jetty-ee10-servlet</artifactId>
+ <version>12.1.8</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- direct declaration so the Jetty 12 security classes win
dependency
+ mediation over the Jetty 11 line pulled by the default
fixture's
+ jetty-servlet -->
+ <dependency>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-security</artifactId>
+ <version>12.1.8</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.eclipse.jetty.compression</groupId>
+ <artifactId>jetty-compression-server</artifactId>
+ <version>12.1.8</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.eclipse.jetty.compression</groupId>
+ <artifactId>jetty-compression-gzip</artifactId>
+ <version>12.1.8</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ </profile>
+ </profiles>
</project>
\ 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 bb86c53a6f..adeec647f1 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 @@ package org.apache.paimon.iceberg;
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;
@@ -45,6 +46,7 @@ import org.apache.iceberg.catalog.Catalog;
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;
@@ -53,6 +55,8 @@ import org.slf4j.LoggerFactory;
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;
@@ -60,6 +64,7 @@ import java.util.LinkedHashMap;
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;
@@ -80,6 +85,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 metadataDirectory;
private final String icebergDatabaseName;
private final TableIdentifier icebergTableIdentifier;
private final IcebergOptions icebergOptions;
@@ -89,6 +96,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.metadataDirectory =
IcebergCommitCallback.catalogTableMetadataPath(table);
Identifier identifier =
Preconditions.checkNotNull(table.catalogEnvironment().identifier());
String icebergDatabase =
options.get(IcebergOptions.METASTORE_DATABASE);
@@ -152,6 +161,11 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
try {
if (!tableExists()) {
+ if (requiresRegistration(newIcebergMetadata)) {
+ 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 =
@@ -192,6 +206,12 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
LOG.info(
"Iceberg table {} exists but has no snapshots,
treating as new table.",
icebergTableIdentifier);
+ 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;
+ }
updateBuilder = updatesForCorrectBase(metadata,
newMetadata, true);
} else {
boolean withBase = checkBase(metadata, newMetadata,
baseIcebergMetadata);
@@ -205,6 +225,13 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
newMetadata.currentSnapshot() != null
?
newMetadata.currentSnapshot().snapshotId()
: "No snapshot");
+ 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);
}
}
@@ -322,6 +349,172 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
}
}
+ /**
+ * 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}; 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 = 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());
+ 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).
@@ -411,15 +604,6 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
return location;
}
- private Table getTable() {
- return restCatalog.loadTable(icebergTableIdentifier);
- }
-
- private void dropTable() {
- // set purge to false, because we don't need to delete the data files
- restCatalog.dropTable(icebergTableIdentifier, false);
- }
-
private Table recreateTable(TableMetadata newMetadata) {
try {
dropTable();
@@ -429,6 +613,15 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
}
}
+ private Table getTable() {
+ return restCatalog.loadTable(icebergTableIdentifier);
+ }
+
+ private void dropTable() {
+ // set purge to false, because we don't need to delete the data files
+ restCatalog.dropTable(icebergTableIdentifier, false);
+ }
+
//
-------------------------------------------------------------------------------------
// metadata updates
//
-------------------------------------------------------------------------------------
@@ -718,9 +911,6 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
return String.format(
"AddSnapshot(%s)",
((MetadataUpdate.AddSnapshot)
update).snapshot().snapshotId());
- } else if (update instanceof MetadataUpdate.RemoveSnapshot) {
- return String.format(
- "RemoveSnapshot(%s)", ((MetadataUpdate.RemoveSnapshot)
update).snapshotId());
} 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 cab2fe10d4..55a1eacb66 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
@@ -21,30 +21,57 @@ package org.apache.paimon.core;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.FileSystemCatalog;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.disk.IOManagerImpl;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.iceberg.IcebergOptions;
+import org.apache.paimon.iceberg.IcebergPathFactory;
+import org.apache.paimon.iceberg.manifest.IcebergManifestEntry;
+import org.apache.paimon.iceberg.manifest.IcebergManifestFile;
+import org.apache.paimon.iceberg.manifest.IcebergManifestFileMeta;
+import org.apache.paimon.iceberg.manifest.IcebergManifestList;
import org.apache.paimon.iceberg.metadata.IcebergMetadata;
import org.apache.paimon.iceberg.metadata.IcebergSnapshot;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.FixedBucketRowKeyExtractor;
import org.apache.paimon.table.sink.TableCommitImpl;
import org.apache.paimon.table.sink.TableWriteImpl;
+import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.IOUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableMetadataParser;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.IcebergGenerics;
+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;
+import java.nio.charset.StandardCharsets;
+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;
@@ -207,6 +234,729 @@ public class IcebergRowLineageCompatibilityTest {
assertThat(metadata.nextRowId()).isEqualTo(3L);
}
+ @Test
+ public void testManifestListCarriesFirstRowIdColumn() throws Exception {
+ FileStoreTable table = createPaimonTable(defaultRowType(),
formatVersionOptions(3), "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));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(1, write.prepareCommit(false, 1));
+ write.close();
+ commit.close();
+
+ // the v3 manifest list must round-trip the first_row_id column
+ IcebergPathFactory paths = new IcebergPathFactory(new
Path(table.location(), "metadata"));
+ IcebergManifestList manifestList = IcebergManifestList.create(table,
paths);
+ List<IcebergManifestFileMeta> metas =
+ manifestList.read(
+ new Path(readIcebergMetadata(table,
1).currentSnapshot().manifestList())
+ .getName());
+ assertThat(metas).isNotEmpty();
+ // the v3 manifest list round-trips the first_row_id column; data
manifests are assigned
+ // at manifest-list write time (this table's single commit starts at
row id 0)
+ for (IcebergManifestFileMeta meta : metas) {
+ if (meta.content() == IcebergManifestFileMeta.Content.DATA) {
+ assertThat(meta.firstRowId()).isEqualTo(0L);
+ } else {
+ assertThat(meta.firstRowId()).isNull();
+ }
+ }
+ }
+
+ @Test
+ public void testManifestFirstRowIdAssignment() throws Exception {
+ FileStoreTable table = createPaimonTable(defaultRowType(),
formatVersionOptions(3), "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));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(1, write.prepareCommit(false, 1));
+
+ write.write(GenericRow.of(3, 30));
+ commit.commit(2, write.prepareCommit(false, 2));
+ write.close();
+ commit.close();
+
+ IcebergPathFactory paths = new IcebergPathFactory(new
Path(table.location(), "metadata"));
+ IcebergManifestList manifestList = IcebergManifestList.create(table,
paths);
+ List<IcebergManifestFileMeta> metas =
+ manifestList.read(
+ new Path(readIcebergMetadata(table,
2).currentSnapshot().manifestList())
+ .getName());
+
+ // every data manifest is assigned; watermark walks addedRowsCount in
list order
+ long watermark = -1;
+ long totalAssigned = 0;
+ for (IcebergManifestFileMeta meta : metas) {
+ if (meta.content() == IcebergManifestFileMeta.Content.DATA) {
+ assertThat(meta.firstRowId()).isNotNull();
+ assertThat(meta.firstRowId()).isGreaterThan(watermark);
+ watermark = meta.firstRowId();
+ totalAssigned += meta.addedRowsCount();
+ } else {
+ assertThat(meta.firstRowId()).isNull();
+ }
+ }
+ // commit1 assigned rows [0,2), commit2 assigned [2,3): manifests
carry 0 and 2
+ assertThat(
+ metas.stream()
+ .filter(m -> m.content() ==
IcebergManifestFileMeta.Content.DATA)
+ .map(IcebergManifestFileMeta::firstRowId))
+ .containsExactlyInAnyOrder(0L, 2L);
+ assertThat(totalAssigned).isEqualTo(3L);
+ }
+
+ @Test
+ public void testReadsManifestListWrittenWithoutFirstRowIdColumn() throws
Exception {
+ // A Layer-1 manifest list physically lacks column 520. The v3 reader
must resolve
+ // the missing column to null (Avro schema resolution), not fail.
+ FileStoreTable table = createPaimonTable(defaultRowType(),
formatVersionOptions(3), "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));
+ write.close();
+ commit.close();
+
+ IcebergPathFactory paths = new IcebergPathFactory(new
Path(table.location(), "metadata"));
+ String listName =
+ new Path(readIcebergMetadata(table,
1).currentSnapshot().manifestList()).getName();
+
+ // rewrite the manifest list through the OLD (14-column) serializer to
simulate Layer 1
+ FileStoreTable v2SchemaView =
+
table.copy(Collections.singletonMap(IcebergOptions.FORMAT_VERSION.key(), "2"));
+ IcebergManifestList oldWriter =
IcebergManifestList.create(v2SchemaView, paths);
+ IcebergManifestList newReader = IcebergManifestList.create(table,
paths);
+ List<IcebergManifestFileMeta> metas = newReader.read(listName);
+ String rewritten = oldWriter.writeWithoutRolling(metas);
+ LocalFileIO.create().deleteQuietly(paths.toManifestListPath(listName));
+ LocalFileIO.create()
+ .rename(paths.toManifestListPath(rewritten),
paths.toManifestListPath(listName));
+
+ // the v3 reader resolves the absent column to null for every meta
+ for (IcebergManifestFileMeta meta : newReader.read(listName)) {
+ assertThat(meta.firstRowId()).isNull();
+ }
+ }
+
+ @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<String, String> 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<String> 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");
+ 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));
+ write.close();
+ commit.close();
+
+ IcebergPathFactory paths = new IcebergPathFactory(new
Path(table.location(), "metadata"));
+ IcebergManifestList manifestList = IcebergManifestList.create(table,
paths);
+ IcebergManifestFile manifestFile = IcebergManifestFile.create(table,
paths);
+ List<IcebergManifestFileMeta> metas =
+ manifestList.read(
+ new Path(readIcebergMetadata(table,
1).currentSnapshot().manifestList())
+ .getName());
+ for (IcebergManifestFileMeta meta : metas) {
+ for (IcebergManifestEntry entry : manifestFile.read(meta)) {
+ // ADDED entries are unassigned by definition; the column must
round-trip as null
+ assertThat(entry.file().firstRowId()).isNull();
+ }
+ }
+ }
+
+ @Test
+ public void testFirstRowIdStableAcrossManifestRewrite() throws Exception {
+ // A single-bucket table always rewrites its one-and-only file whole,
so no entry ever
+ // survives a rewrite unchanged (the same-path invariant this test
targets never fires).
+ // Use two buckets instead, and only touch one of them: the other
bucket's file must
+ // then be carried, byte-for-byte unchanged, as an EXISTING entry into
the manifest that
+ // gets rewritten because its sibling entry was removed.
+ RowType rowType = defaultRowType();
+ Map<String, String> 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));
+
+ // an append bundled with a full compaction is committed as two
separate physical
+ // snapshots (append, then compact), so the Iceberg metadata id to
inspect is whatever
+ // snapshot id is actually latest now, not the external commit
identifier used above
+ Map<String, Long> idsBefore =
+ effectiveFileFirstRowIds(table,
table.snapshotManager().latestSnapshotId());
+ assertThat(idsBefore).isNotEmpty();
+
+ // delete the bucket-0 key and compact only bucket 0: bucket 1's file
is left entirely
+ // untouched, so the shared manifest is rewritten (bucket 0's entry
removed) while
+ // bucket 1's entry must survive, under the same file path, as a
materialized EXISTING
+ // entry
+ write.write(GenericRow.ofKind(RowKind.DELETE, keyBucket0, 10));
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ commit.commit(2, write.prepareCommit(true, 2));
+ write.close();
+ commit.close();
+
+ Map<String, Long> idsAfter =
+ effectiveFileFirstRowIds(table,
table.snapshotManager().latestSnapshotId());
+ assertThat(idsAfter).isNotEmpty();
+ boolean checkedAtLeastOneSurvivor = false;
+ for (Map.Entry<String, Long> e : idsAfter.entrySet()) {
+ Long before = idsBefore.get(e.getKey());
+ if (before != null) {
+ checkedAtLeastOneSurvivor = true;
+ // a file carried across the rewrite keeps its effective first
row id
+ assertThat(e.getValue()).as("file %s",
e.getKey()).isEqualTo(before);
+ }
+ }
+ assertThat(checkedAtLeastOneSurvivor)
+ .as("expected bucket 1's file to survive the manifest rewrite")
+ .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<String, String> 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<IcebergManifestEntry> entries = entryReader.read(meta);
+
assertThat(entries.get(0).status()).isEqualTo(IcebergManifestEntry.Status.DELETED);
+ List<IcebergManifestFileMeta> 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<String, Long> 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<DataFile> 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();
+ FileStoreTable table = createPaimonTable(defaultRowType(),
formatVersionOptions(3), "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));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(1, write.prepareCommit(false, 1));
+ write.close();
+ commit.close();
+
+ HadoopCatalog icebergCatalog = new HadoopCatalog(new Configuration(),
tempDir.toString());
+ Table icebergTable =
icebergCatalog.loadTable(TableIdentifier.of("mydb.db", "t"));
+ assertThat(icebergTable.currentSnapshot().firstRowId()).isEqualTo(0L);
+ for (ManifestFile manifest :
+
icebergTable.currentSnapshot().dataManifests(icebergTable.io())) {
+ assertThat(manifestFirstRowId(manifest)).isNotNull();
+ }
+ }
+
+ @Test
+ public void testLayer1TableUpgradesOnNextCommit() throws Exception {
+ FileStoreTable table = createPaimonTable(defaultRowType(),
formatVersionOptions(3), "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));
+
+ // simulate a Layer-1-written manifest list: strip the assigned
first_row_id
+ IcebergPathFactory paths = new IcebergPathFactory(new
Path(table.location(), "metadata"));
+ IcebergManifestList manifestList = IcebergManifestList.create(table,
paths);
+ String listName =
+ new Path(readIcebergMetadata(table,
1).currentSnapshot().manifestList()).getName();
+ List<IcebergManifestFileMeta> stripped = new ArrayList<>();
+ for (IcebergManifestFileMeta meta : manifestList.read(listName)) {
+ stripped.add(
+ new IcebergManifestFileMeta(
+ meta.manifestPath(),
+ meta.manifestLength(),
+ meta.partitionSpecId(),
+ meta.content(),
+ meta.sequenceNumber(),
+ meta.minSequenceNumber(),
+ meta.addedSnapshotId(),
+ meta.addedFilesCount(),
+ meta.existingFilesCount(),
+ meta.deletedFilesCount(),
+ meta.addedRowsCount(),
+ meta.existingRowsCount(),
+ meta.deletedRowsCount(),
+ meta.partitions(),
+ null));
+ }
+ // overwrite the manifest list in place with unassigned metas
+ LocalFileIO.create().deleteQuietly(paths.toManifestListPath(listName));
+ // writeWithoutRolling creates a new file; rename it over the original
+ String rewritten = manifestList.writeWithoutRolling(stripped);
+ 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));
+
+ IcebergMetadata metadataAfterCommit2 = readIcebergMetadata(table, 2);
+ IcebergSnapshot snapshotAfterCommit2 =
metadataAfterCommit2.currentSnapshot();
+ List<IcebergManifestFileMeta> dataManifestsAfterCommit2 = new
ArrayList<>();
+ for (IcebergManifestFileMeta meta :
+ 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<String, Long> 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
+ public void testRowTrackingTableUsesSyntheticIds() throws Exception {
+ Map<String, String> options = formatVersionOptions(3);
+ options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ 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));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(1, write.prepareCommit(false, 1));
+ write.close();
+ commit.close();
+
+ // documented independence: assignment behaves exactly as for any
other table
+ IcebergMetadata metadata = readIcebergMetadata(table, 1);
+ assertThat(metadata.nextRowId()).isEqualTo(2L);
+ assertThat(effectiveFileFirstRowIds(table,
1).values()).containsExactly(0L);
+ }
+
+ @Test
+ public void testCompactMetadataIfNeededMaterializesRowLineageUnderV3()
throws Exception {
+ // exercises the `compactMetadataIfNeeded` manifest-metadata-merge
call site under v3,
+ // which no existing test hits (all format-version-3 tests leave
COMPACT_MIN_FILE_NUM
+ // at its default of 10, and all tests that force compaction stay on
format version 2).
+ RowType rowType = defaultRowType();
+ Map<String, String> customOptions = formatVersionOptions(3);
+ customOptions.put(IcebergOptions.COMPACT_MIN_FILE_NUM.key(), "2");
+ customOptions.put(IcebergOptions.COMPACT_MAX_FILE_NUM.key(), "2");
+ // large enough that manifests are never excluded as "already big
enough", so the
+ // min/max file-count thresholds above are what actually triggers the
merge
+ customOptions.put(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), "64
mb");
+ FileStoreTable table = createPaimonTable(rowType, customOptions,
"avro");
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl<?> write =
+ table.newWrite(commitUser)
+ .withIOManager(new IOManagerImpl(tempDir.toString() +
"/tmp"));
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ // commit 1: single manifest M1 (candidates=1 < COMPACT_MIN_FILE_NUM,
no compaction)
+ write.write(GenericRow.of(1, 10));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(1, write.prepareCommit(false, 1));
+ Map<String, Long> idsAfterCommit1 =
+ effectiveFileFirstRowIds(table,
table.snapshotManager().latestSnapshotId());
+ assertThat(idsAfterCommit1).hasSize(1);
+
+ // commit 2: M1 (already assigned) + fresh M2 => 2 candidates, meets
both thresholds,
+ // manifest metadata compaction merges them into a single manifest
this same commit
+ write.write(GenericRow.of(3, 30));
+ commit.commit(2, write.prepareCommit(false, 2));
+ assertThat(dataManifestCount(table,
table.snapshotManager().latestSnapshotId()))
+ .as("commit 2 should have merged M1+M2 into a single data
manifest")
+ .isEqualTo(1);
+ Map<String, Long> idsAfterCommit2 =
+ effectiveFileFirstRowIds(table,
table.snapshotManager().latestSnapshotId());
+ assertThat(idsAfterCommit2).hasSize(2);
+ for (Map.Entry<String, Long> e : idsAfterCommit1.entrySet()) {
+ // every file's effective first-row-id survives the
metadata-compaction commit
+ // unchanged, whether it was inherited or already explicit before
the merge
+ assertThat(idsAfterCommit2)
+ .as("file %s", e.getKey())
+ .containsEntry(e.getKey(), e.getValue());
+ }
+ assertMergedManifestExistingEntriesHaveExplicitFirstRowId(table);
+
+ // commit 3: merges again, this time the base manifest already
contains an
+ // EXISTING entry with an explicit field 142 from commit 2's merge
(file 1/2) sitting
+ // alongside an ADDED entry with inherited-only field 142 (file from
commit 2) -- this
+ // is the "explicit-142 passthrough" branch of materializeFirstRowIds
that no other
+ // test reaches
+ write.write(GenericRow.of(4, 40));
+ commit.commit(3, write.prepareCommit(false, 3));
+ write.close();
+ commit.close();
+ assertThat(dataManifestCount(table,
table.snapshotManager().latestSnapshotId()))
+ .as("commit 3 should merge again into a single data manifest")
+ .isEqualTo(1);
+ Map<String, Long> idsAfterCommit3 =
+ effectiveFileFirstRowIds(table,
table.snapshotManager().latestSnapshotId());
+ assertThat(idsAfterCommit3).hasSize(3);
+ for (Map.Entry<String, Long> e : idsAfterCommit2.entrySet()) {
+ assertThat(idsAfterCommit3)
+ .as("file %s", e.getKey())
+ .containsEntry(e.getKey(), e.getValue());
+ }
+ assertMergedManifestExistingEntriesHaveExplicitFirstRowId(table);
+ }
+
+ @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");
+ 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));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(1, write.prepareCommit(false, 1));
+ write.write(GenericRow.of(3, 30));
+ commit.commit(2, write.prepareCommit(false, 2));
+ write.close();
+ commit.close();
+
+ HadoopCatalog icebergCatalog = new HadoopCatalog(new Configuration(),
tempDir.toString());
+ Table icebergTable =
icebergCatalog.loadTable(TableIdentifier.of("mydb.db", "t"));
+
+ // Attempt 1: resolve a per-row `_row_id` value through iceberg-data's
GA generics
+ // reader. `select(...)` does not throw, but the requested metadata
column is silently
+ // dropped from the projected schema:
GenericReader/InternalRecordWrapper in
+ // iceberg-data 1.11.0 have no wiring for MetadataColumns.ROW_ID
(unlike _pos/_file/
+ // _deleted/_spec_id, which the same reader stack does resolve), so
every record's
+ // "_row_id" field comes back null instead of the assigned/inherited
value. This is
+ // verified here rather than assumed: if a future Iceberg release adds
real support,
+ // this loop will start observing non-null values and the assertion
below must change.
+ boolean anyRowIdResolved = false;
+ try (CloseableIterable<Record> records =
+ IcebergGenerics.read(icebergTable).select("k", "v",
"_row_id").build()) {
+ for (Record record : records) {
+ if (record.getField("_row_id") != null) {
+ anyRowIdResolved = true;
+ }
+ }
+ }
+ assertThat(anyRowIdResolved)
+ .as(
+ "iceberg-data 1.11.0 GA generics do not resolve the
_row_id metadata "
+ + "column; if this starts failing, generics
gained support and "
+ + "the fallback below can be
simplified/removed")
+ .isFalse();
+
+ // Fallback: GA's ManifestFiles/DataFile APIs DO resolve the per-file
first_row_id
+ // (including inheritance from the manifest-level value), so assert
actual values,
+ // not just non-nullity.
+ List<Long> resolvedFirstRowIds = new ArrayList<>();
+ for (ManifestFile manifest :
+
icebergTable.currentSnapshot().dataManifests(icebergTable.io())) {
+ try (ManifestReader<DataFile> reader =
+ ManifestFiles.read(manifest, icebergTable.io(),
icebergTable.specs())) {
+ for (DataFile file : reader) {
+ resolvedFirstRowIds.add(dataFileFirstRowId(file));
+ }
+ }
+ }
+ assertThat(resolvedFirstRowIds).containsExactlyInAnyOrder(0L, 2L);
+ }
+
+ @Test
+ public void testV2ManifestListSchemaHasNoFirstRowIdColumn() throws
Exception {
+ // pins the v2 manifest-list shape: 14 columns, no first_row_id
anywhere, so a future
+ // change to the v3 schema construction cannot silently leak into v2's
byte-identical
+ // output
+
assertThat(IcebergManifestFileMeta.schema(false).getFieldCount()).isEqualTo(14);
+
assertThat(IcebergManifestFileMeta.schema(false).getFields().stream().map(DataField::name))
+ .doesNotContain("first_row_id");
+
assertThat(IcebergManifestFileMeta.schema(true).getFieldCount()).isEqualTo(14);
+
assertThat(IcebergManifestFileMeta.schema(true).getFields().stream().map(DataField::name))
+ .doesNotContain("first_row_id");
+
+ FileStoreTable table = createPaimonTable(defaultRowType(),
formatVersionOptions(2), "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));
+ write.close();
+ commit.close();
+
+ // the written manifest-list file's raw bytes must not contain
"first_row_id" anywhere;
+ // Avro embeds the writer schema as JSON in the file header, so a
plain byte-scan of the
+ // whole file is a valid (and stronger-than-parsed) check of the
physical shape
+ Path manifestListPath =
+ new Path(readIcebergMetadata(table,
1).currentSnapshot().manifestList());
+ byte[] bytes;
+ try (SeekableInputStream in =
table.fileIO().newInputStream(manifestListPath)) {
+ bytes = IOUtils.readFully(in, false);
+ }
+ String content = new String(bytes, StandardCharsets.ISO_8859_1);
+ assertThat(content).doesNotContain("first_row_id");
+ }
+
// ------------------------------------------------------------------------
// helpers
// ------------------------------------------------------------------------
@@ -330,6 +1080,135 @@ public class IcebergRowLineageCompatibilityTest {
}
}
+ private FileStoreTable createPkPaimonTable(RowType rowType, Map<String,
String> customOptions)
+ throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path path = new Path(tempDir.toString());
+ Options options = new Options(customOptions);
+ // two fixed buckets so a manifest rewrite can leave one bucket's file
untouched
+ // (see testFirstRowIdStableAcrossManifestRewrite)
+ options.set(CoreOptions.BUCKET, 2);
+ options.set(
+ IcebergOptions.METADATA_ICEBERG_STORAGE,
IcebergOptions.StorageType.TABLE_LOCATION);
+ options.set(CoreOptions.FILE_FORMAT, "avro");
+ Schema schema =
+ new Schema(
+ rowType.getFields(),
+ Collections.<String>emptyList(),
+ Collections.singletonList("k"),
+ options.toMap(),
+ "");
+ try (FileSystemCatalog paimonCatalog = new FileSystemCatalog(fileIO,
path)) {
+ paimonCatalog.createDatabase("mydb2", false);
+ Identifier id = Identifier.create("mydb2", "t");
+ paimonCatalog.createTable(id, schema, false);
+ return (FileStoreTable) paimonCatalog.getTable(id);
+ }
+ }
+
+ /** Finds the smallest positive key whose fixed-bucket hash lands in
{@code targetBucket}. */
+ private int findKeyForBucket(FileStoreTable table, int targetBucket) {
+ FixedBucketRowKeyExtractor extractor = new
FixedBucketRowKeyExtractor(table.schema());
+ for (int k = 1; k < 10_000; k++) {
+ extractor.setRecord(GenericRow.of(k, 0));
+ if (extractor.bucket() == targetBucket) {
+ return k;
+ }
+ }
+ throw new IllegalStateException("No key found for bucket " +
targetBucket);
+ }
+
+ /** Effective per-file first row id: explicit field 142, or inherited per
the spec rules. */
+ private Map<String, Long> effectiveFileFirstRowIds(FileStoreTable table,
long snapshotId)
+ throws Exception {
+ IcebergPathFactory paths = new IcebergPathFactory(new
Path(table.location(), "metadata"));
+ IcebergManifestList manifestList = IcebergManifestList.create(table,
paths);
+ IcebergManifestFile manifestFile = IcebergManifestFile.create(table,
paths);
+ Map<String, Long> result = new HashMap<>();
+ for (IcebergManifestFileMeta meta :
+ manifestList.read(
+ new Path(
+ readIcebergMetadata(table, snapshotId)
+ .currentSnapshot()
+ .manifestList())
+ .getName())) {
+ if (meta.content() != IcebergManifestFileMeta.Content.DATA) {
+ continue;
+ }
+ 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();
+ } else {
+ effective = watermark;
+ watermark += entry.file().recordCount();
+ }
+ if (entry.isLive()) {
+ result.put(entry.file().filePath(), effective);
+ }
+ }
+ }
+ return result;
+ }
+
+ /** Number of DATA-content manifests referenced by the given snapshot's
manifest list. */
+ private int dataManifestCount(FileStoreTable table, long snapshotId)
throws Exception {
+ return dataManifestMetas(table, snapshotId).size();
+ }
+
+ private List<IcebergManifestFileMeta> dataManifestMetas(FileStoreTable
table, long snapshotId)
+ throws Exception {
+ IcebergPathFactory paths = new IcebergPathFactory(new
Path(table.location(), "metadata"));
+ IcebergManifestList manifestList = IcebergManifestList.create(table,
paths);
+ List<IcebergManifestFileMeta> result = new ArrayList<>();
+ for (IcebergManifestFileMeta meta :
+ manifestList.read(
+ new Path(
+ readIcebergMetadata(table, snapshotId)
+ .currentSnapshot()
+ .manifestList())
+ .getName())) {
+ if (meta.content() == IcebergManifestFileMeta.Content.DATA) {
+ result.add(meta);
+ }
+ }
+ return result;
+ }
+
+ /**
+ * After a manifest-metadata-compaction merge, every EXISTING entry in the
merged manifest(s)
+ * for the table's current snapshot must carry an explicit (non-null)
field 142: EXISTING
+ * entries are, by definition, carried-over/rewritten entries, so their
first_row_id must have
+ * been materialized by {@code materializeFirstRowIds} rather than left to
be inherited from the
+ * (now-merged-away) original manifest.
+ */
+ private void
assertMergedManifestExistingEntriesHaveExplicitFirstRowId(FileStoreTable table)
+ throws Exception {
+ IcebergPathFactory paths = new IcebergPathFactory(new
Path(table.location(), "metadata"));
+ IcebergManifestFile manifestFile = IcebergManifestFile.create(table,
paths);
+ long snapshotId = table.snapshotManager().latestSnapshotId();
+ boolean checkedAtLeastOneExistingEntry = false;
+ for (IcebergManifestFileMeta meta : dataManifestMetas(table,
snapshotId)) {
+ for (IcebergManifestEntry entry : manifestFile.read(meta)) {
+ if (entry.status() == IcebergManifestEntry.Status.EXISTING) {
+ checkedAtLeastOneExistingEntry = true;
+ assertThat(entry.file().firstRowId())
+ .as("EXISTING entry for file %s",
entry.file().filePath())
+ .isNotNull();
+ }
+ }
+ }
+ assertThat(checkedAtLeastOneExistingEntry)
+ .as("expected at least one materialized EXISTING entry after
the merge")
+ .isTrue();
+ }
+
private Path metadataPath(FileStoreTable table, long snapshotId) {
return new Path(table.location(),
String.format("metadata/v%d.metadata.json", snapshotId));
}
@@ -341,4 +1220,42 @@ public class IcebergRowLineageCompatibilityTest {
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 a6205dc107..79a2909df1 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
@@ -389,10 +389,18 @@ class IcebergMetadataTest {
void testFormatVersionV3Table() throws Exception {
// Create a v3 format version Iceberg table
Table icebergTable = createIcebergTableV3("v3_snapshot_table");
- TableMetadata base = ((HasTableOperations)
icebergTable).operations().current();
- ((HasTableOperations) icebergTable)
- .operations()
- .commit(base,
TableMetadata.buildFrom(base).enableRowLineage().build());
+ 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 15782acdc5..b20aa47aa8 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
@@ -28,6 +28,8 @@ import org.apache.paimon.data.GenericRow;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.iceberg.manifest.IcebergManifestFileMeta;
+import org.apache.paimon.iceberg.manifest.IcebergManifestList;
import org.apache.paimon.iceberg.metadata.IcebergMetadata;
import org.apache.paimon.iceberg.metadata.IcebergSnapshot;
import org.apache.paimon.options.MemorySize;
@@ -43,9 +45,11 @@ import org.apache.paimon.types.RowType;
import org.apache.iceberg.BaseTable;
import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.ManifestFile;
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;
@@ -54,6 +58,7 @@ import org.apache.iceberg.data.Record;
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;
@@ -68,9 +73,11 @@ import java.util.ArrayList;
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;
@@ -79,6 +86,7 @@ import java.util.stream.Collectors;
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 {
@@ -1482,6 +1490,277 @@ public class IcebergRestMetadataCommitterTest {
// Known Layer 1 limitation (documented in the spec): the server-side
next-row-id
// watermark restarts on recreation and stays behind the local
metadata; commits
// must keep succeeding regardless (validation is first-row-id >=
next-row-id).
+
+ // reader-visible lineage from the REST catalog matches the file-based
mirror:
+ // snapshot first-row-id and manifest assignments come from local
metadata, never
+ // from the server's table-level watermark. Compare by value, not just
non-nullity,
+ // against the locally-written IcebergMetadata + manifest list under
the paimon
+ // table's own metadata dir (catalogTableMetadataPath), which is the
source of truth
+ // the REST-registered table's metadata-location actually points at.
+ long latestSnapshotId = table.snapshotManager().latestSnapshotId();
+ IcebergMetadata localMetadata =
+ IcebergMetadata.fromPath(
+ table.fileIO(),
+ new Path(
+ catalogTableMetadataPath(table),
+ String.format("v%d.metadata.json",
latestSnapshotId)));
+ IcebergSnapshot localSnapshot = localMetadata.currentSnapshot();
+ assertThat(localSnapshot.firstRowId()).isNotNull();
+
+ IcebergPathFactory pathFactory = new
IcebergPathFactory(catalogTableMetadataPath(table));
+ IcebergManifestList localManifestList =
IcebergManifestList.create(table, pathFactory);
+ List<Long> localDataManifestFirstRowIds =
+ localManifestList.read(new
Path(localSnapshot.manifestList()).getName()).stream()
+ .filter(m -> m.content() ==
IcebergManifestFileMeta.Content.DATA)
+ .map(IcebergManifestFileMeta::firstRowId)
+ .collect(Collectors.toList());
+
assertThat(localDataManifestFirstRowIds).isNotEmpty().doesNotContainNull();
+
+ Table reloaded = restCatalog.loadTable(TableIdentifier.of("mydb",
"t"));
+
assertThat(reloaded.currentSnapshot().firstRowId()).isEqualTo(localSnapshot.firstRowId());
+ // manifest-level first_row_id is only exposed by the GA (1.10+)
reader API
+ if (GA_ROW_LINEAGE_READER) {
+ List<Long> restDataManifestFirstRowIds = new ArrayList<>();
+ for (ManifestFile manifest :
reloaded.currentSnapshot().dataManifests(reloaded.io())) {
+ assertThat(manifestFirstRowId(manifest)).isNotNull();
+ restDataManifestFirstRowIds.add(manifestFirstRowId(manifest));
+ }
+ 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());
+ }
+ }
+
+ /**
+ * 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<String, String> 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<String, String> 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<Path> 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<Path> 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<Path> registerFiles(FileStoreTable table) throws
Exception {
+ List<Path> 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<Endpoint> endpoints = new HashSet<>((Set<Endpoint>)
endpointsField.get(sessionCatalog));
+ assertThat(endpoints.remove(Endpoint.V1_REGISTER_TABLE)).isTrue();
+ endpointsField.set(sessionCatalog, endpoints);
}
private static class TestRecord {
@@ -1565,4 +1844,35 @@ public class IcebergRestMetadataCommitterTest {
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);
+ }
+ }
+
+ /** 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);
+ }
+ }
}
