JingsongLi commented on code in PR #9245:
URL: https://github.com/apache/paimon/pull/9245#discussion_r3800806418
##########
paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java:
##########
@@ -2001,24 +2028,121 @@ private boolean isSameFormatVersion(int
baseFormatVersion) {
/**
* 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 by Layer 2 (this commit or a later one)
satisfies "null-142
+ * rows == ADDED rows", so {@code addedRowsCount()} is exact for it. But a
manifest carried over
+ * from before manifest-level assignment existed (a "Layer-1" manifest)
may reach here
+ * unassigned with existing/deleted entries whose per-file field 142 is
also still null; for
+ * those, {@code addedRowsCount()} alone would undercount the rows this
assignment must cover,
+ * silently shrinking the range handed out and colliding with the next
commit's ids. Callers
+ * MUST use {@code assignedRows} (not this commit's added-records count)
to advance the
+ * snapshot's added-rows / table next-row-id, precisely because of that
mismatch.
+ */
+ private 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));
Review Comment:
[P1] Preserve lineage when compaction replaces data files
This assigns a new range to every unassigned manifest. Replacement files
produced by Paimon data compaction reach this point as ADDED entries with field
142 set to null, even when they contain unchanged rows. A pure REPLACE/COMPACT
commit therefore gives those rows new `_row_id` values and a new inherited
`_last_updated_sequence_number`. Iceberg v3 requires existing rows moved for
any reason to copy both lineage values. Please carry row-level lineage into
replacement data files (or keep v3 publication disabled until that is
supported) and add a GA test comparing both metadata columns per logical row
before and after data compaction; the current rewrite test only checks an
untouched file path.
##########
paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java:
##########
@@ -2001,24 +2028,121 @@ private boolean isSameFormatVersion(int
baseFormatVersion) {
/**
* 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 by Layer 2 (this commit or a later one)
satisfies "null-142
+ * rows == ADDED rows", so {@code addedRowsCount()} is exact for it. But a
manifest carried over
+ * from before manifest-level assignment existed (a "Layer-1" manifest)
may reach here
+ * unassigned with existing/deleted entries whose per-file field 142 is
also still null; for
+ * those, {@code addedRowsCount()} alone would undercount the rows this
assignment must cover,
+ * silently shrinking the range handed out and colliding with the next
commit's ids. Callers
+ * MUST use {@code assignedRows} (not this commit's added-records count)
to advance the
+ * snapshot's added-rows / table next-row-id, precisely because of that
mismatch.
+ */
+ private 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));
+ watermark += trueInheritingRowsCount(meta);
+ } else {
+ result.add(meta);
+ }
+ }
+ return new ManifestRowIdAssignment(result, watermark -
snapshotFirstRowId);
+ }
+
+ /**
+ * The true number of rows an unassigned manifest needs from the row-id
space: the sum of {@code
+ * recordCount()} over entries whose per-file first_row_id (field 142) is
null.
+ *
+ * <p>Fast path: when the manifest has no existing/deleted entries ({@code
existingFilesCount()
+ * + deletedFilesCount() == 0}), every entry is ADDED and, by the Layer-2
invariant, has a null
+ * field 142, so {@code addedRowsCount()} already equals this sum without
having to read the
+ * manifest file.
+ *
+ * <p>Otherwise (a manifest that may carry Layer-1-era existing/deleted
entries whose field 142
+ * was never materialized) the manifest is actually read and entries are
inspected one by one,
+ * since {@code addedRowsCount()} alone would not include those entries'
rows.
*/
- private 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 long trueInheritingRowsCount(IcebergManifestFileMeta meta) {
+ if (meta.existingFilesCount() + meta.deletedFilesCount() == 0) {
+ return meta.addedRowsCount();
+ }
+ long sum = 0;
+ for (IcebergManifestEntry entry :
+ manifestFile.read(new Path(meta.manifestPath()).getName())) {
+ if (entry.file().firstRowId() == null) {
+ sum += entry.file().recordCount();
+ }
+ }
+ return sum;
}
- 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) 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.file().firstRowId() == null) {
Review Comment:
[P1] Do not materialize or advance through DELETED entries
Iceberg's GA `ManifestReader` assigns inherited `first_row_id` only when
`status != DELETED`. This loop instead assigns an ID to a legacy DELETED entry
with null field 142 and advances the watermark by its record count, shifting
every following live file. For example, with manifest first ID 100, a 5-row
DELETED entry before an EXISTING entry is initially read as EXISTING=100 by
Iceberg, but this rewrite persists EXISTING=105. Please skip DELETED entries in
both `materializeFirstRowIds` and `trueInheritingRowsCount`, fix the test
oracle that currently advances through deletes as well, and add a GA regression
with a deleted entry before a live entry.
##########
paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java:
##########
@@ -2001,24 +2028,121 @@ private boolean isSameFormatVersion(int
baseFormatVersion) {
/**
* 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 by Layer 2 (this commit or a later one)
satisfies "null-142
+ * rows == ADDED rows", so {@code addedRowsCount()} is exact for it. But a
manifest carried over
+ * from before manifest-level assignment existed (a "Layer-1" manifest)
may reach here
+ * unassigned with existing/deleted entries whose per-file field 142 is
also still null; for
+ * those, {@code addedRowsCount()} alone would undercount the rows this
assignment must cover,
+ * silently shrinking the range handed out and colliding with the next
commit's ids. Callers
+ * MUST use {@code assignedRows} (not this commit's added-records count)
to advance the
+ * snapshot's added-rows / table next-row-id, precisely because of that
mismatch.
+ */
+ private 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));
+ watermark += trueInheritingRowsCount(meta);
+ } else {
+ result.add(meta);
+ }
+ }
+ return new ManifestRowIdAssignment(result, watermark -
snapshotFirstRowId);
+ }
+
+ /**
+ * The true number of rows an unassigned manifest needs from the row-id
space: the sum of {@code
+ * recordCount()} over entries whose per-file first_row_id (field 142) is
null.
+ *
+ * <p>Fast path: when the manifest has no existing/deleted entries ({@code
existingFilesCount()
+ * + deletedFilesCount() == 0}), every entry is ADDED and, by the Layer-2
invariant, has a null
+ * field 142, so {@code addedRowsCount()} already equals this sum without
having to read the
+ * manifest file.
+ *
+ * <p>Otherwise (a manifest that may carry Layer-1-era existing/deleted
entries whose field 142
+ * was never materialized) the manifest is actually read and entries are
inspected one by one,
+ * since {@code addedRowsCount()} alone would not include those entries'
rows.
*/
- private 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 long trueInheritingRowsCount(IcebergManifestFileMeta meta) {
+ if (meta.existingFilesCount() + meta.deletedFilesCount() == 0) {
+ return meta.addedRowsCount();
+ }
+ long sum = 0;
+ for (IcebergManifestEntry entry :
Review Comment:
[P2] Avoid a serial full-manifest scan in the commit callback
The first Layer-2 commit for a large Layer-1 table reads every entry of
every unassigned manifest containing EXISTING or DELETED files, serially in the
commit path; retries repeat the same work. This can turn a small commit into an
O(total historical manifest entries) metadata operation. The Iceberg spec
permits `added_rows_count + existing_rows_count` as a safe upper bound,
accepting ID gaps while avoiding these scans and naturally excluding deleted
rows. If exact accounting is retained, please bound/parallelize it and add
latency/entry-count observability.
##########
paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestList.java:
##########
@@ -69,9 +70,11 @@ public static IcebergManifestList create(FileStoreTable
table, IcebergPathFactor
+ "manifest_file_partitions:r508,"
+ "array_id_r508:508");
FileFormat fileFormat = FileFormat.fromIdentifier("avro", avroOptions);
+ boolean withFirstRowId =
Review Comment:
[P1] Fence older writers before enabling this schema
A #9244 binary always writes the 14-column manifest-list schema. It can
project a 15-column list written here, but its next commit reconstructs the
list with the old schema and silently drops field 520; a manifest rewrite can
similarly drop field 142. That makes current `_row_id` values null, and a later
new writer can reassign the same physical files from a different range. Please
define an enforceable writer-capability barrier (or require and document a
quiesced atomic rollout with rollback prohibited) and add a new-writer ->
old-writer -> new-writer compatibility test using a GA reader.
##########
paimon-iceberg/pom.xml:
##########
@@ -356,4 +373,71 @@ 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. -->
+ <id>iceberg-ga</id>
+ <properties>
+ <iceberg.version>1.11.0</iceberg.version>
Review Comment:
[P2] Make the GA validation profile executable in CI
On this head, the six selected row-lineage tests under JDK 17 and
`-Piceberg-ga` all fail before their assertions: Iceberg 1.11 resolves Avro
1.12.1, while the reactor's Paimon format classes call
`DataFileStream.DataBlock#getNumEntries()` from the 1.11.4 ABI, producing
`NoSuchMethodError`. This profile is also not referenced by a GitHub workflow,
so the advertised GA validation is not a merge gate. Please align the Avro ABI,
make the profile pass from a clean reactor build, and wire it into CI.
##########
paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java:
##########
@@ -1348,6 +1351,44 @@ public void testRecreateWithNonZeroLineageWatermark()
throws Exception {
// 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());
Review Comment:
[P1] Recover the server watermark instead of validating only reader-visible
fields
This recreation starts the REST table at `next-row-id=0` while the local
snapshot has `first-row-id=3` and `added-rows=1`. Iceberg's
`TableMetadata.Builder` increments the server watermark by `addedRows`, so it
ends at 1 rather than 4, even though manifest IDs 0 through 3 are already
assigned. A subsequent external REST writer can therefore reuse IDs. Recreate
must import/replay the local high-water mark or fail safely. Please assert
`server.nextRowId() >= local.nextRowId()` here and exercise one external
append; if the REST table is intended to be read-only, that restriction needs
server-side enforcement.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
