Re: [PR] fix(occ): manifest list rebuild [iceberg-go]

["laskoviymishka (via GitHub)" <gi...@apache.org>]

laskoviymishka commented on code in PR #982:
URL: https://github.com/apache/iceberg-go/pull/982#discussion_r3200076769


##########
table/snapshot_producers.go:
##########
@@ -903,9 +943,103 @@ func (sp *snapshotProducer) commit(ctx context.Context) (_ []Update, _ []Require
 		})
 	}
 
+	// Build the manifest-list rebuild closure. It is called by doCommit
+	// on each OCC retry to regenerate the manifest list so it correctly
+	// inherits all data files committed by concurrent writers since the
+	// original snapshot was built.
+	formatVersion := sp.txn.meta.formatVersion
+	snapshotID := sp.snapshotID
+	commitUUID := sp.commitUuid
+	capturedSnapshot := snapshot // copy the value so the closure is self-contained
+	processManifestsFn := func(m []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) {
+		return sp.processManifests(m)
+	}
+
+	rebuildFn := func(_ context.Context, freshParent *Snapshot, fio iceio.WriteFileIO, attempt int) (_ *Snapshot, retErr error) {
+		// Load inherited manifests from the fresh parent.
+		var inherited []iceberg.ManifestFile
+		if freshParent != nil {
+			inherited, retErr = freshParent.Manifests(fio)
+			if retErr != nil {
+				return nil, fmt.Errorf("rebuild manifest list: load parent manifests: %w", retErr)
+			}
+		}
+
+		// Combine own manifests with inherited ones, applying any
+		// producer-specific processing (no-op for fast/merge-append).
+		combined, procErr := processManifestsFn(slices.Concat(ownManifests, inherited))
+		if procErr != nil {
+			return nil, fmt.Errorf("rebuild manifest list: process manifests: %w", procErr)
+		}
+
+		// Derive the sequence number. When there is a fresh parent, use its
+		// sequence number + 1 so the rebuilt snapshot is strictly greater than
+		// any committed peer. When there is no fresh parent (first snapshot in
+		// the table or unknown parent), preserve the original sequence number
+		// from the initial build.
+		var newSeq int64
+		if freshParent != nil && formatVersion >= 2 {
+			newSeq = freshParent.SequenceNumber + 1
+		} else {
+			newSeq = capturedSnapshot.SequenceNumber
+		}
+
+		// Write the rebuilt manifest list to a path unique to this retry
+		// attempt. Each retry uses a different attempt counter in the filename
+		// (snap-{id}-{attempt}-{uuid}.avro) so that S3 conditional-write
+		// semantics (if-none-match) do not reject the overwrite. Orphaned files
+		// from superseded retry attempts are removed by doCommit after the
+		// commit succeeds.
+		fname := newManifestListFileName(snapshotID, attempt, commitUUID)
+		manifestListPath := locProvider.NewMetadataLocation(fname)
+
+		out, createErr := fio.Create(manifestListPath)
+		if createErr != nil {
+			return nil, fmt.Errorf("rebuild manifest list: create file: %w", createErr)
+		}
+		defer internal.CheckedClose(out, &retErr)
+
+		var parentID *int64
+		if freshParent != nil {
+			id := freshParent.SnapshotID
+			parentID = &id
+		}
+
+		firstRowID := int64(0)

Review Comment:
   [Blocking] V3 row-lineage stale on retry. The rebuild closure hard-codes `firstRowID := int64(0)` while `capturedSnapshot.FirstRowID` keeps the value `sp.txn.meta.NextRowID()` produced at attempt 0. On a v3 retry the manifest-list container metadata declares `first-row-id=0` while the snapshot record advertises something else, and `validateAndUpdateRowLineage` will compare a stale `*FirstRowID` against the catalog's advanced `nextRowID`. Suggest threading `freshMeta.NextRowID()` into the closure, reassigning `rebuilt.FirstRowID`, and recomputing `addedRows` from `writer.NextRowID()` after `AddManifests` — surgical change inside this closure, no producer-architecture impact.



##########
table/snapshot_producers.go:
##########
@@ -903,9 +943,103 @@ func (sp *snapshotProducer) commit(ctx context.Context) (_ []Update, _ []Require
 		})
 	}
 
+	// Build the manifest-list rebuild closure. It is called by doCommit
+	// on each OCC retry to regenerate the manifest list so it correctly
+	// inherits all data files committed by concurrent writers since the
+	// original snapshot was built.
+	formatVersion := sp.txn.meta.formatVersion
+	snapshotID := sp.snapshotID
+	commitUUID := sp.commitUuid
+	capturedSnapshot := snapshot // copy the value so the closure is self-contained
+	processManifestsFn := func(m []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) {
+		return sp.processManifests(m)
+	}
+
+	rebuildFn := func(_ context.Context, freshParent *Snapshot, fio iceio.WriteFileIO, attempt int) (_ *Snapshot, retErr error) {
+		// Load inherited manifests from the fresh parent.
+		var inherited []iceberg.ManifestFile
+		if freshParent != nil {
+			inherited, retErr = freshParent.Manifests(fio)
+			if retErr != nil {
+				return nil, fmt.Errorf("rebuild manifest list: load parent manifests: %w", retErr)
+			}
+		}
+
+		// Combine own manifests with inherited ones, applying any
+		// producer-specific processing (no-op for fast/merge-append).
+		combined, procErr := processManifestsFn(slices.Concat(ownManifests, inherited))
+		if procErr != nil {
+			return nil, fmt.Errorf("rebuild manifest list: process manifests: %w", procErr)
+		}
+
+		// Derive the sequence number. When there is a fresh parent, use its
+		// sequence number + 1 so the rebuilt snapshot is strictly greater than
+		// any committed peer. When there is no fresh parent (first snapshot in
+		// the table or unknown parent), preserve the original sequence number
+		// from the initial build.
+		var newSeq int64
+		if freshParent != nil && formatVersion >= 2 {
+			newSeq = freshParent.SequenceNumber + 1
+		} else {
+			newSeq = capturedSnapshot.SequenceNumber
+		}
+
+		// Write the rebuilt manifest list to a path unique to this retry
+		// attempt. Each retry uses a different attempt counter in the filename
+		// (snap-{id}-{attempt}-{uuid}.avro) so that S3 conditional-write
+		// semantics (if-none-match) do not reject the overwrite. Orphaned files
+		// from superseded retry attempts are removed by doCommit after the
+		// commit succeeds.
+		fname := newManifestListFileName(snapshotID, attempt, commitUUID)
+		manifestListPath := locProvider.NewMetadataLocation(fname)
+
+		out, createErr := fio.Create(manifestListPath)
+		if createErr != nil {
+			return nil, fmt.Errorf("rebuild manifest list: create file: %w", createErr)
+		}
+		defer internal.CheckedClose(out, &retErr)
+
+		var parentID *int64
+		if freshParent != nil {
+			id := freshParent.SnapshotID
+			parentID = &id
+		}
+
+		firstRowID := int64(0)
+		if formatVersion == 3 {
+			writer, wrErr := iceberg.NewManifestListWriterV3(out, snapshotID, newSeq, firstRowID, parentID)
+			if wrErr != nil {
+				return nil, fmt.Errorf("rebuild manifest list: create v3 writer: %w", wrErr)
+			}
+			defer internal.CheckedClose(writer, &retErr)
+			if addErr := writer.AddManifests(combined); addErr != nil {
+				return nil, fmt.Errorf("rebuild manifest list: add manifests: %w", addErr)
+			}
+		} else {
+			if wErr := iceberg.WriteManifestList(formatVersion, out, snapshotID, parentID, &newSeq, firstRowID, combined); wErr != nil {
+				return nil, fmt.Errorf("rebuild manifest list: write: %w", wErr)
+			}
+		}
+
+		rebuilt := capturedSnapshot

Review Comment:
   [Blocking] Snapshot summary captured at attempt 0 is reused on retry. `updateSnapshotSummaries` rebased `total-records` / `total-data-files` / `total-files-size` against the *stale* parent's summary at build time; once a peer commits, the rebuilt snapshot publishes regressed totals to every consumer that reads the summary (Iceberg metrics, BI tools, expire-snapshots accounting). Fix scopes naturally inside this closure: keep the per-attempt added/removed counts (already in `summaryProps`/`sp.snapshotProps`) and recompute via `updateSnapshotSummaries(..., freshParent.Summary)` before assigning to `rebuilt.Summary`. Doesn't require touching the producer architecture.



##########
table/table.go:
##########
@@ -483,6 +490,19 @@ func (t Table) doCommit(ctx context.Context, updates []Update, reqs []Requiremen
 		return nil, err
 	}
 
+	// Delete manifest-list files that were written during failed retry
+	// attempts and have now been superseded by the committed rebuild.
+	// These are orphaned objects that will never be referenced again.
+	if len(orphanedManifests) > 0 {

Review Comment:
   [Important] Orphan cleanup runs only on success. Non-`ErrCommitFailed` errors (`ErrCommitDiverged` from validators, 5xx unknown-state) early-return at lines 485 / 490 before reaching this cleanup. Files written by previous rebuild attempts then leak permanently — they are never referenced by any committed snapshot. A `defer` running this loop with a "this exit is cleanable" flag would handle most cases without much surgery. The only path where the orphan list should *not* be removed is the unknown-state 5xx, where one of the orphans may actually be the snapshot the catalog accepted but didn't acknowledge — worth distinguishing explicitly.



##########
table/table.go:
##########
@@ -428,6 +421,20 @@ func (t Table) doCommit(ctx context.Context, updates []Update, reqs []Requiremen
 			}
 			current = fresh.metadata
 			reqs = rewriteRefSnapshotRequirements(reqs, co.branch, current)
+
+			// Rebuild snapshot manifest lists to inherit all files committed
+			// by concurrent writers since the snapshot was originally built.
+			// Without this, the new snapshot's manifest list would only
+			// contain its own files and callers scanning the current snapshot
+			// would miss every concurrent writer's data.
+			if wfs, ok := fs.(icebergio.WriteFileIO); ok {

Review Comment:
   [Important] Silent `WriteFileIO` fallback reintroduces the original bug. When `fs` doesn't implement `icebergio.WriteFileIO` the rebuild silently no-ops and the original stale-parent manifest list is sent to the catalog — exactly the failure mode this PR is fixing. The same cast is repeated for orphan cleanup at line 497, with the same silent-skip behavior. Suggest hoisting one cast to the top of `doCommit` and either erroring out or logging at error level when it fails. Every real commit-path FS implements `WriteFileIO`; a silent fallback here is the worst possible failure mode.



##########
table/snapshot_producers.go:
##########
@@ -815,12 +815,52 @@ func (sp *snapshotProducer) summary(props iceberg.Properties) (Summary, error) {
 	}, previousSummary)
 }
 
+// computeOwnManifests returns the subset of allManifests that were written
+// by this producer (i.e. not inherited from the parent snapshot). These are
+// preserved across OCC retry attempts when the manifest list is rebuilt
+// against a fresh parent.
+func (sp *snapshotProducer) computeOwnManifests(allManifests []iceberg.ManifestFile) []iceberg.ManifestFile {
+	if sp.parentSnapshotID <= 0 {
+		// No parent means all manifests are new — nothing to exclude.
+		return allManifests
+	}
+
+	parent, err := sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
+	if err != nil || parent == nil {

Review Comment:
   [Blocking] All three error returns (`SnapshotByID` lookup at line 829, `parent.Manifests` read at 833, missing parent for `parentSnapshotID > 0`) collapse to `return allManifests` — "treat every manifest as own". On a transient parent-manifest-list read failure the rebuild then concatenates `ownManifests = newManifests` with `freshParent.Manifests(fio)`, so every inherited manifest appears twice and downstream readers count rows twice for those entries. Propagate the error instead: change the signature to `(ownManifests []iceberg.ManifestFile, err error)` and bubble up through `commit`. The `parentSnapshotID <= 0` early return (line 823) is fine to keep as the all-new fallback.



##########
table/snapshot_producers.go:
##########
@@ -903,9 +943,103 @@ func (sp *snapshotProducer) commit(ctx context.Context) (_ []Update, _ []Require
 		})
 	}
 
+	// Build the manifest-list rebuild closure. It is called by doCommit
+	// on each OCC retry to regenerate the manifest list so it correctly
+	// inherits all data files committed by concurrent writers since the
+	// original snapshot was built.
+	formatVersion := sp.txn.meta.formatVersion
+	snapshotID := sp.snapshotID
+	commitUUID := sp.commitUuid
+	capturedSnapshot := snapshot // copy the value so the closure is self-contained
+	processManifestsFn := func(m []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) {
+		return sp.processManifests(m)
+	}
+
+	rebuildFn := func(_ context.Context, freshParent *Snapshot, fio iceio.WriteFileIO, attempt int) (_ *Snapshot, retErr error) {
+		// Load inherited manifests from the fresh parent.
+		var inherited []iceberg.ManifestFile
+		if freshParent != nil {
+			inherited, retErr = freshParent.Manifests(fio)
+			if retErr != nil {
+				return nil, fmt.Errorf("rebuild manifest list: load parent manifests: %w", retErr)
+			}
+		}
+
+		// Combine own manifests with inherited ones, applying any
+		// producer-specific processing (no-op for fast/merge-append).
+		combined, procErr := processManifestsFn(slices.Concat(ownManifests, inherited))
+		if procErr != nil {
+			return nil, fmt.Errorf("rebuild manifest list: process manifests: %w", procErr)
+		}
+
+		// Derive the sequence number. When there is a fresh parent, use its
+		// sequence number + 1 so the rebuilt snapshot is strictly greater than
+		// any committed peer. When there is no fresh parent (first snapshot in
+		// the table or unknown parent), preserve the original sequence number
+		// from the initial build.
+		var newSeq int64
+		if freshParent != nil && formatVersion >= 2 {
+			newSeq = freshParent.SequenceNumber + 1

Review Comment:
   [Blocking] `freshParent.SequenceNumber + 1` violates the spec invariant when a peer writes to a *different* branch and bumps the table-wide `last-sequence-number` without advancing this branch's parent. `MetadataBuilder.AddSnapshot` rejects `SequenceNumber <= lastSequenceNumber`, and other clients reject the metadata on read. The `else` branch (`freshParent == nil`) reuses `capturedSnapshot.SequenceNumber` and has the same problem against a freshly loaded metadata. Fix: derive from the fresh metadata — plumb `freshMeta` (or precompute `freshMeta.LastSequenceNumber()`) into the closure and use `freshMeta.LastSequenceNumber() + 1` for both branches when `formatVersion >= 2`.



-- 
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: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org