vbabenkoru commented on code in PR #9245:
URL: https://github.com/apache/paimon/pull/9245#discussion_r3801611969
##########
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:
Adopted your suggestion: each unassigned manifest now reserves
`added_rows_count + existing_rows_count` from the watermark (DELETED rows
excluded by construction), so the commit path never reads manifest contents
anymore. Rows that were already materialized leave id gaps, which the spec
allows.
--
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]
