vbabenkoru commented on code in PR #9245:
URL: https://github.com/apache/paimon/pull/9245#discussion_r3801604954
##########
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:
Materialization now skips DELETED entries exactly like the GA reader does
(their field 142 stays null and they don't advance the walk), and the test
oracle got the same fix. Regression test added: a legacy manifest with a 1-row
DELETED entry before a live entry keeps the live file at the manifest's own
`first_row_id`, cross-checked against the Iceberg 1.11 reader.
--
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]
