[PR] Fix/occ manifest list rebuild [iceberg-go]

["mzzz-zzm (via GitHub)" <gi...@apache.org>]

mzzz-zzm opened a new pull request, #982:
URL: https://github.com/apache/iceberg-go/pull/982

   fix(occ): rebuild manifest list on OCC retry to inherit concurrent writes
   
   Fixes #976
   ## Problem
   
   When `doCommit` retries after an OCC conflict, it re-uses the stale snapshot that was built before the first attempt. That snapshot's manifest list was written against the original parent and does not include any data files committed by concurrent writers in the meantime.
   
   On catalogs that perform server-side snapshot validation (e.g. AWS S3 Tables), this causes **silent data loss**: the retried commit succeeds (HTTP 200) but the stale manifest list effectively drops the concurrent writer's files from the table history.
   
   ## Fix
   
   Each `snapshotProducer` now records which manifests it wrote itself (`ownManifests` — those not inherited from the original parent). A `rebuildManifestList` closure is attached to the `addSnapshotUpdate` and called by `doCommit` on every retry. The closure:
   
   1. Loads the fresh branch head's manifest list (concurrent manifests).
   2. Concatenates `ownManifests` with the fresh inherited manifests.
   3. Writes a new manifest list file with a retry-attempt suffix so each attempt gets a unique path.
   4. Adjusts `sequence-number` relative to the fresh parent.
   
   After a successful commit, `doCommit` deletes the manifest list files produced by superseded retry attempts (orphan cleanup).
   
   ## Files changed
   
   - `table/snapshot_producers.go`: `computeOwnManifests`, `rebuildFn` closure
   - `table/updates.go`: `ownManifests` / `rebuildManifestList` fields on `addSnapshotUpdate`, propagated through `Apply`
   - `table/table.go`: `rebuildSnapshotUpdates` helper, orphan cleanup after commit
   - `table/rebuild_manifest_test.go`: unit tests for `rebuildSnapshotUpdates`


-- 
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

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_r3202392983


##########
table/rebuild_manifest_test.go:
##########
@@ -0,0 +1,476 @@
+// 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.
+
+package table
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	iceio "github.com/apache/iceberg-go/io"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// newMetadataWithLastSeqNum builds a v2 Metadata whose last-sequence-number
+// equals lastSeqNum by grafting a synthetic snapshot at that sequence number.
+// Used by tests that need to control freshMeta.LastSequenceNumber().
+func newMetadataWithLastSeqNum(t *testing.T, lastSeqNum int64) Metadata {
+	t.Helper()
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/seqtest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	builder, err := MetadataBuilderFromBase(meta, "")
+	require.NoError(t, err)
+	snap := Snapshot{
+		SnapshotID:     1000,
+		SequenceNumber: lastSeqNum,
+		TimestampMs:    meta.LastUpdatedMillis() + 1,
+		Summary:        &Summary{Operation: OpAppend},
+	}
+	require.NoError(t, builder.AddSnapshot(&snap))
+	require.NoError(t, builder.SetSnapshotRef(MainBranch, 1000, BranchRef))
+	out, err := builder.Build()
+	require.NoError(t, err)
+
+	return out
+}
+
+// newV3MetadataWithNextRowID builds a v3 Metadata whose NextRowID() returns
+// nextRowID by adding a synthetic snapshot that consumes that many rows.
+// Used by tests that need to control freshMeta.NextRowID() for v3 row lineage.
+func newV3MetadataWithNextRowID(t *testing.T, nextRowID int64) Metadata {
+	t.Helper()
+	spec := iceberg.NewPartitionSpec()
+	txn, _ := createTestTransactionWithMemIO(t, spec)
+	txn.meta.formatVersion = 3
+
+	firstRowID := int64(0)
+	addedRows := nextRowID
+	snap := Snapshot{
+		SnapshotID:     1000,
+		SequenceNumber: 1,
+		TimestampMs:    txn.meta.base.LastUpdatedMillis() + 1,
+		Summary:        &Summary{Operation: OpAppend},
+		FirstRowID:     &firstRowID,
+		AddedRows:      &addedRows,
+	}
+	require.NoError(t, txn.meta.AddSnapshot(&snap))
+	require.NoError(t, txn.meta.SetSnapshotRef(MainBranch, 1000, BranchRef))
+
+	meta, err := txn.meta.Build()
+	require.NoError(t, err)
+	require.Equal(t, nextRowID, meta.NextRowID())
+
+	return meta
+}
+
+// ptr returns a pointer to v, used in test helper expressions.
+// NOTE: ptr is also declared in pos_delete_partitioned_fanout_writer_test.go;
+// both files share the same package so only one declaration is needed.
+// This comment documents the shared usage — do not re-add a declaration here.
+
+// rebuildUpdate constructs an addSnapshotUpdate whose rebuildManifestList
+// closure simply records the freshParent it received and returns a new
+// snapshot whose ManifestList is the given newManifestList value.
+func rebuildUpdate(snap *Snapshot, newManifestList string, gotParent **Snapshot) *addSnapshotUpdate {
+	return &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, freshParent *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			*gotParent = freshParent
+			rebuilt := *snap
+			rebuilt.ManifestList = newManifestList
+
+			return &rebuilt, nil
+		},
+	}
+}
+
+// TestRebuildSnapshotUpdates_CallsClosureWithFreshParent verifies that
+// rebuildSnapshotUpdates invokes the rebuild closure and passes the fresh
+// branch head as freshParent.
+func TestRebuildSnapshotUpdates_CallsClosureWithFreshParent(t *testing.T) {
+	const oldManifest = "s3://bucket/old-manifest-list.avro"
+	const newManifest = "s3://bucket/new-manifest-list.avro"
+
+	// Build a fresh metadata that has a different snapshot than the one
+	// embedded in the update, so the parent-hasn't-changed guard is
+	// bypassed and the closure must be called.
+	freshHead := int64(42)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	parentID := int64(7) // original snap's parent — does NOT match freshHead (42)
+	snap := &Snapshot{
+		SnapshotID:       99,
+		ParentSnapshotID: &parentID,
+		ManifestList:     oldManifest,
+		Summary:          &Summary{Operation: OpAppend},
+	}
+
+	var receivedParent *Snapshot
+	upd := rebuildUpdate(snap, newManifest, &receivedParent)
+
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{upd},
+		freshMeta,
+		MainBranch,
+		iceio.LocalFS{},
+		1,
+	)
+	require.NoError(t, err)
+	require.Len(t, rebuilt, 1)
+	require.Len(t, orphaned, 1, "old manifest list must be recorded as orphaned")
+
+	// The closure should have been called with the branch's current head.
+	require.NotNil(t, receivedParent)
+	assert.Equal(t, freshHead, receivedParent.SnapshotID)
+
+	// The rebuilt update must carry the new manifest list.
+	addUpd, ok := rebuilt[0].(*addSnapshotUpdate)
+	require.True(t, ok)
+	assert.Equal(t, newManifest, addUpd.Snapshot.ManifestList)
+
+	// The superseded manifest list becomes an orphan.
+	assert.Equal(t, oldManifest, orphaned[0])
+}
+
+// TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged verifies that
+// rebuildSnapshotUpdates skips the rebuild when the update's snapshot
+// already has the fresh branch head as its parent (no-op retry).
+func TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged(t *testing.T) {
+	const manifest = "s3://bucket/manifest-list.avro"
+
+	freshHead := int64(42)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	// Parent already equals the fresh head — rebuild must be skipped.
+	snap := &Snapshot{
+		SnapshotID:       99,
+		ParentSnapshotID: &freshHead, // same as fresh head
+		ManifestList:     manifest,
+		Summary:          &Summary{Operation: OpAppend},
+	}
+
+	called := false
+	upd := &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, _ *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			called = true
+
+			return snap, nil
+		},
+	}
+
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{upd},
+		freshMeta,
+		MainBranch,
+		iceio.LocalFS{},
+		1,
+	)
+	require.NoError(t, err)
+	assert.False(t, called, "rebuild closure must not be called when parent is already up-to-date")
+	assert.Empty(t, orphaned, "no orphans when rebuild is skipped")
+	assert.Same(t, upd, rebuilt[0].(*addSnapshotUpdate), "original update must pass through unchanged")
+}
+
+// TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates verifies that
+// updates without a rebuildManifestList closure are returned unmodified.
+func TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates(t *testing.T) {
+	plainUpd := NewAddSnapshotUpdate(&Snapshot{
+		SnapshotID:   1,
+		ManifestList: "s3://bucket/no-rebuild.avro",
+		Summary:      &Summary{Operation: OpAppend},
+	})
+
+	// freshMeta may be nil — the plain update must not be touched.
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{plainUpd},
+		nil,
+		MainBranch,
+		iceio.LocalFS{},
+		0,
+	)
+	require.NoError(t, err)
+	assert.Empty(t, orphaned)
+	assert.Same(t, plainUpd, rebuilt[0].(*addSnapshotUpdate))
+}
+
+// TestRebuildSnapshotUpdates_PropagatesClosureError verifies that an error
+// returned by the rebuild closure surfaces as the function's return error.
+func TestRebuildSnapshotUpdates_PropagatesClosureError(t *testing.T) {
+	freshHead := int64(5)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	parent := int64(1)
+	snap := &Snapshot{
+		SnapshotID:       10,
+		ParentSnapshotID: &parent,
+		ManifestList:     "s3://bucket/old.avro",
+		Summary:          &Summary{Operation: OpAppend},
+	}
+	wantErr := errors.New("simulated S3 write failure")
+	upd := &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, _ *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			return nil, wantErr
+		},
+	}
+
+	_, _, err := rebuildSnapshotUpdates(t.Context(), []Update{upd}, freshMeta, MainBranch, iceio.LocalFS{}, 1)
+	assert.ErrorIs(t, err, wantErr)
+}
+
+// ---------------------------------------------------------------------------
+// Fix 3 — newSeq derived from freshMeta.LastSequenceNumber()
+// ---------------------------------------------------------------------------
+
+// TestRebuildFn_SeqNumDerivedFromFreshMeta verifies that the rebuilt snapshot's
+// SequenceNumber equals freshMeta.LastSequenceNumber()+1, NOT
+// freshParent.SequenceNumber+1. A concurrent writer on a different branch can
+// advance the table-wide last-sequence-number without advancing this branch's
+// parent, so using freshParent.SequenceNumber+1 would violate the spec invariant.
+func TestRebuildFn_SeqNumDerivedFromFreshMeta(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	txn, wfs := createTestTransactionWithMemIO(t, spec)
+
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, _, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	require.NotNil(t, addSnap.rebuildManifestList, "rebuildManifestList closure must be set")
+
+	// Simulate a concurrent writer on another branch that bumped the global
+	// last-sequence-number to 99 without advancing this branch's parent.
+	// old code: newSeq = capturedSnapshot.SequenceNumber (stale, ≤ 99 — spec violation)
+	// new code: newSeq = freshMeta.LastSequenceNumber() + 1 = 100
+	freshMeta := newMetadataWithLastSeqNum(t, 99)
+	require.Equal(t, int64(99), freshMeta.LastSequenceNumber())
+
+	rebuilt, err := addSnap.rebuildManifestList(context.Background(), freshMeta, nil, wfs, 1)
+	require.NoError(t, err)
+	require.Equal(t, int64(100), rebuilt.SequenceNumber,
+		"SequenceNumber must be freshMeta.LastSequenceNumber()+1; using freshParent.SequenceNumber+1 violates the spec when another branch advanced the global counter")
+}
+
+// TestRebuildFn_SeqNumV1TableIsZero verifies that v1 tables keep SequenceNumber == 0
+// regardless of what freshMeta reports (v1 does not use sequence numbers).
+func TestRebuildFn_SeqNumV1TableIsZero(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	txn, wfs := createTestTransactionWithMemIO(t, spec)
+	txn.meta.formatVersion = 1 // override to v1
+
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, _, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	require.NotNil(t, addSnap.rebuildManifestList)
+
+	freshMeta := newMetadataWithLastSeqNum(t, 99)
+	rebuilt, err := addSnap.rebuildManifestList(context.Background(), freshMeta, nil, wfs, 1)
+	require.NoError(t, err)
+	require.Equal(t, int64(0), rebuilt.SequenceNumber, "v1 tables must always have SequenceNumber == 0")
+}
+
+// ---------------------------------------------------------------------------
+// Fix 1 — firstRowID derived from freshMeta.NextRowID()
+// ---------------------------------------------------------------------------
+
+// TestRebuildFn_V3FirstRowIDDerivedFromFreshMeta verifies that on a v3 table
+// the rebuilt snapshot's FirstRowID equals freshMeta.NextRowID() rather than
+// the hardcoded 0. If two writers race and the peer commits first, the
+// catalog's nextRowID has already advanced; using 0 would produce a
+// first-row-id that disagrees with the catalog's view and fails row-lineage
+// validation.
+func TestRebuildFn_V3FirstRowIDDerivedFromFreshMeta(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	txn, wfs := createTestTransactionWithMemIO(t, spec)
+	txn.meta.formatVersion = 3
+
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, _, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	require.NotNil(t, addSnap.rebuildManifestList)
+
+	// Simulate a concurrent writer that added 50 rows, advancing nextRowID to 50.
+	// old code: firstRowID = 0 (hardcoded)
+	// new code: firstRowID = freshMeta.NextRowID() = 50
+	freshMeta := newV3MetadataWithNextRowID(t, 50)
+	require.Equal(t, int64(50), freshMeta.NextRowID(), "freshMeta.NextRowID() must be 50")
+
+	rebuilt, err := addSnap.rebuildManifestList(context.Background(), freshMeta, nil, wfs, 1)
+	require.NoError(t, err)
+	require.NotNil(t, rebuilt.FirstRowID, "v3 rebuilt snapshot must have FirstRowID")
+	require.Equal(t, int64(50), *rebuilt.FirstRowID,
+		"FirstRowID must equal freshMeta.NextRowID(); hardcoded 0 would conflict with catalog's advanced nextRowID")
+
+	// AddedRows must be recomputed from writer.NextRowID() - firstRowID for
+	// this attempt (NOT carried over from capturedSnapshot at attempt 0).
+	// The single test data file has record count 1, so the producer's own
+	// contribution this attempt is exactly 1 row.
+	require.NotNil(t, rebuilt.AddedRows, "v3 rebuilt snapshot must have AddedRows")
+	require.Equal(t, int64(1), *rebuilt.AddedRows,
+		"AddedRows must equal writer.NextRowID() - firstRowID (1 row from this producer's data file)")

Review Comment:
   [Important] AddedRows assertion is regression-insensitive. The producer's data file has record-count 1, so `capturedSnapshot.AddedRows == 1` from attempt 0 — and `*rebuilt.AddedRows == 1` is the same value. If the rebuild closure were reverted to propagate `capturedSnapshot.AddedRows` (the pre-fix path) instead of recomputing from `writer.NextRowID() - firstRowID`, this assertion still passes. The fixture symmetry shadows the very regression the comment names. To pin the recompute, set up a state where the captured value and the recomputed value disagree — e.g. multi-row data file with `freshMeta.NextRowID() = 50` and capture a producer whose attempt-0 AddedRows was different, or use two data files (captured AddedRows=2) against a freshMeta that yields a different count. Same fixture issue applies to `TestRebuildFn_V3FirstRowIDZeroWhenNilNextRowID`.



##########
table/commit_retry_test.go:
##########
@@ -266,3 +307,244 @@ func TestReadRetryConfig_ClampsNegativeProperties(t *testing.T) {
 	assert.Equal(t, uint(CommitMaxRetryWaitMsDefault), cfg.maxWaitMs)
 	assert.Equal(t, uint(CommitTotalRetryTimeoutMsDefault), cfg.totalTimeoutMs)
 }
+
+// ---------------------------------------------------------------------------
+// Fix 5 — mandatory WriteFileIO check at top of doCommit
+// ---------------------------------------------------------------------------
+
+// TestDoCommit_NonWriteFileIOReturnsError verifies that doCommit fails
+// immediately when the table's file system does not implement WriteFileIO.
+// A silent skip would reuse the stale manifest list — exactly the bug
+// this PR was designed to fix.
+func TestDoCommit_NonWriteFileIOReturnsError(t *testing.T) {
+	cat := &flakyCatalog{}
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/rotest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	cat.metadata = meta
+
+	tbl := New(
+		Identifier{"db", "ro-test"},
+		meta, "file:///tmp/rotest/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return readOnlyIO{}, nil },
+		cat,
+	)
+
+	_, doErr := tbl.doCommit(t.Context(), nil, nil)
+	require.Error(t, doErr, "doCommit must fail when FS does not implement WriteFileIO")
+	assert.Contains(t, doErr.Error(), "WriteFileIO",
+		"error message must mention WriteFileIO so callers understand the requirement")
+	assert.Equal(t, int32(0), cat.attempts.Load(),
+		"CommitTable must not be called when FS check fails")
+}
+
+// ---------------------------------------------------------------------------
+// Fix 6 — orphan cleanup via defer
+// ---------------------------------------------------------------------------
+
+// newOCCTable creates a table that uses the given wfs for its FS and the given
+// catalog for commits. meta should include retry-config properties so that
+// doCommit's retry loop allows at least one retry.
+func newOCCTable(t *testing.T, meta Metadata, wfs iceio.WriteFileIO, cat CatalogIO) *Table {
+	t.Helper()
+
+	return New(
+		Identifier{"db", "occ-cleanup-test"},
+		meta,
+		"mem://default/table-location/metadata/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return wfs, nil },
+		cat,
+	)
+}
+
+// newMemIOWithRetryMeta creates a test memIO and a matching table Metadata that
+// includes retry-config properties, so doCommit's retry loop allows retries.
+// The location matches createTestTransactionWithMemIO so they share the same
+// memIO for writing manifest files.
+func newMemIOWithRetryMeta(t *testing.T, spec iceberg.PartitionSpec) (*memIO, Metadata) {
+	t.Helper()
+	wfs := newMemIO(1<<20, nil)
+	schema := simpleSchema()
+	meta, err := NewMetadata(schema, &spec, UnsortedSortOrder, "mem://default/table-location",
+		iceberg.Properties{
+			CommitNumRetriesKey:          "3",
+			CommitMinRetryWaitMsKey:      "1",
+			CommitMaxRetryWaitMsKey:      "2",
+			CommitTotalRetryTimeoutMsKey: "60000",
+		})
+	require.NoError(t, err, "new metadata")
+
+	return wfs, meta
+}
+
+// TestDoCommit_OrphanCleanedOnSuccess verifies that manifest-list files
+// orphaned by OCC retries are removed after a successful commit. These files
+// are written during rebuild and must not leak on the happy path.
+func TestDoCommit_OrphanCleanedOnSuccess(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	// Build a transaction from the retry-enabled meta and commit via the producer.
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	// Pre-populate the original manifest list path in the IO so that the
+	// defer's wfs.Remove call has a concrete entry to delete.
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	// Catalog: fail once with ErrCommitFailed (triggers rebuild that orphans
+	// originalManifestList), then succeed.
+	cat := &sequentialCatalog{
+		metadata: meta,
+		errs:     []error{ErrCommitFailed},
+	}
+	tbl = newOCCTable(t, meta, wfs, cat)
+
+	_, err = tbl.doCommit(t.Context(), updates, reqs, withCommitBranch(MainBranch))
+	require.NoError(t, err, "doCommit must succeed on the second attempt")
+
+	_, stillExists := wfs.files[originalManifestList]
+	assert.False(t, stillExists,
+		"orphaned manifest list must be removed after successful commit")
+}
+
+// TestDoCommit_OrphanNotCleanedOnUnknownError verifies that manifest-list
+// files are NOT removed when CommitTable returns an unknown non-ErrCommitFailed
+// error (5xx / gateway timeout). In that case the catalog may have silently
+// accepted the commit, meaning one of the "orphaned" files is actually the
+// live snapshot. Deleting it would permanently corrupt the table.
+func TestDoCommit_OrphanNotCleanedOnUnknownError(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	unknown5xxErr := errors.New("simulated 5xx: internal server error")
+	// Catalog: fail once (ErrCommitFailed → rebuild → orphan created),
+	// then return a 5xx (non-ErrCommitFailed → cleanupOrphans=false).
+	cat := &sequentialCatalog{
+		metadata: meta,
+		errs:     []error{ErrCommitFailed, unknown5xxErr},
+	}
+	tbl = newOCCTable(t, meta, wfs, cat)
+
+	_, err = tbl.doCommit(t.Context(), updates, reqs, withCommitBranch(MainBranch))
+	require.ErrorIs(t, err, unknown5xxErr, "5xx error must propagate")
+
+	_, stillExists := wfs.files[originalManifestList]
+	assert.True(t, stillExists,
+		"orphaned manifest list must NOT be removed when commit outcome is unknown (5xx)")
+}
+
+// TestDoCommit_OrphanCleanedOnCommitDiverged verifies that manifest-list files
+// orphaned by rebuild attempts are removed when ErrCommitDiverged is returned
+// by a conflict validator. Diverged commits are terminal (no retry), and since
+// neither of the orphaned files was ever accepted by the catalog, they are safe
+// to delete. The defer cleanup runs with cleanupOrphans=true on this path.
+func TestDoCommit_OrphanCleanedOnCommitDiverged(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	// freshMeta has a snapshot on MainBranch so validators run on the retry attempt.
+	freshID := int64(42)
+	freshMeta := newConflictTestMetadataWithProps(t, &freshID, iceberg.Properties{
+		CommitNumRetriesKey:          "3",
+		CommitMinRetryWaitMsKey:      "1",
+		CommitMaxRetryWaitMsKey:      "2",
+		CommitTotalRetryTimeoutMsKey: "60000",
+	})
+
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	// Catalog: fail once (ErrCommitFailed → rebuild → orphan created).
+	// LoadTable returns freshMeta (has branch snapshot → validators run on retry).
+	// Validator returns ErrCommitDiverged immediately.
+	cat := &sequentialCatalog{
+		metadata: meta,
+		loadMeta: freshMeta,
+		errs:     []error{ErrCommitFailed},
+	}
+	tbl = newOCCTable(t, meta, wfs, cat)
+
+	divergedValidator := func(*conflictContext) error { return ErrCommitDiverged }
+
+	_, err = tbl.doCommit(t.Context(), updates, reqs,
+		withCommitBranch(MainBranch),
+		withCommitValidators(divergedValidator),
+	)
+	require.ErrorIs(t, err, ErrCommitDiverged)
+
+	// The defer must have fired with cleanupOrphans=true and removed the orphan.
+	_, stillExists := wfs.files[originalManifestList]
+	assert.False(t, stillExists,
+		"orphaned manifest list must be removed even on ErrCommitDiverged: "+
+			"the file was never accepted by the catalog so it is safe to delete")
+}
+
+// TestDoCommit_OrphanCleanedOnRetriesExhausted verifies that when every retry
+// attempt fails with ErrCommitFailed and the loop exits with the budget
+// exhausted, the defer still fires with cleanupOrphans=true. None of the
+// orphaned manifest-list files were ever accepted by the catalog, so they are
+// safe to delete on this terminal exit.
+func TestDoCommit_OrphanCleanedOnRetriesExhausted(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	// numRetries=3 → 4 attempts; every attempt fails with ErrCommitFailed.
+	cat := &sequentialCatalog{
+		metadata: meta,
+		errs:     []error{ErrCommitFailed, ErrCommitFailed, ErrCommitFailed, ErrCommitFailed},

Review Comment:
   [Non-blocking] None of the `TestDoCommit_Orphan*` tests progress `freshMeta` between retries — `sequentialCatalog.metadata` is the same object on every `LoadTable` call, so on each retry the rebuild closure sees a `freshHead` that equals the original parent (or is nil for an empty table). The skip-if-parent-unchanged predicate could short-circuit, and even when it doesn't, the closure runs in degenerate mode where every per-retry recompute (newSeq, firstRowID, summary) reproduces the captured-attempt-0 values. So this test exercises the orphan-cleanup state machine, but not the actual *retry* semantics the fixes target. Worth one end-to-end test that drives ≥2 ErrCommitFailed retries with a `freshMeta` that progresses between attempts (the existing `headTrackingCatalog` in `commit_refresh_replay_test.go` is most of the way there). That single test would close more coverage than the four orphan-cleanup variants combined — every fix in this PR is exercised at once: orphan list
  grows by N, each rebuild sees a different `LastSequenceNumber`/`NextRowID`, summary chains on top of an evolving parent.



##########
table/rebuild_manifest_test.go:
##########
@@ -0,0 +1,476 @@
+// 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.
+
+package table
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	iceio "github.com/apache/iceberg-go/io"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// newMetadataWithLastSeqNum builds a v2 Metadata whose last-sequence-number
+// equals lastSeqNum by grafting a synthetic snapshot at that sequence number.
+// Used by tests that need to control freshMeta.LastSequenceNumber().
+func newMetadataWithLastSeqNum(t *testing.T, lastSeqNum int64) Metadata {
+	t.Helper()
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/seqtest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	builder, err := MetadataBuilderFromBase(meta, "")
+	require.NoError(t, err)
+	snap := Snapshot{
+		SnapshotID:     1000,
+		SequenceNumber: lastSeqNum,
+		TimestampMs:    meta.LastUpdatedMillis() + 1,
+		Summary:        &Summary{Operation: OpAppend},
+	}
+	require.NoError(t, builder.AddSnapshot(&snap))
+	require.NoError(t, builder.SetSnapshotRef(MainBranch, 1000, BranchRef))
+	out, err := builder.Build()
+	require.NoError(t, err)
+
+	return out
+}
+
+// newV3MetadataWithNextRowID builds a v3 Metadata whose NextRowID() returns
+// nextRowID by adding a synthetic snapshot that consumes that many rows.
+// Used by tests that need to control freshMeta.NextRowID() for v3 row lineage.
+func newV3MetadataWithNextRowID(t *testing.T, nextRowID int64) Metadata {
+	t.Helper()
+	spec := iceberg.NewPartitionSpec()
+	txn, _ := createTestTransactionWithMemIO(t, spec)
+	txn.meta.formatVersion = 3
+
+	firstRowID := int64(0)
+	addedRows := nextRowID
+	snap := Snapshot{
+		SnapshotID:     1000,
+		SequenceNumber: 1,
+		TimestampMs:    txn.meta.base.LastUpdatedMillis() + 1,
+		Summary:        &Summary{Operation: OpAppend},
+		FirstRowID:     &firstRowID,
+		AddedRows:      &addedRows,
+	}
+	require.NoError(t, txn.meta.AddSnapshot(&snap))
+	require.NoError(t, txn.meta.SetSnapshotRef(MainBranch, 1000, BranchRef))
+
+	meta, err := txn.meta.Build()
+	require.NoError(t, err)
+	require.Equal(t, nextRowID, meta.NextRowID())
+
+	return meta
+}
+
+// ptr returns a pointer to v, used in test helper expressions.
+// NOTE: ptr is also declared in pos_delete_partitioned_fanout_writer_test.go;
+// both files share the same package so only one declaration is needed.
+// This comment documents the shared usage — do not re-add a declaration here.
+
+// rebuildUpdate constructs an addSnapshotUpdate whose rebuildManifestList
+// closure simply records the freshParent it received and returns a new
+// snapshot whose ManifestList is the given newManifestList value.
+func rebuildUpdate(snap *Snapshot, newManifestList string, gotParent **Snapshot) *addSnapshotUpdate {
+	return &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, freshParent *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			*gotParent = freshParent
+			rebuilt := *snap
+			rebuilt.ManifestList = newManifestList
+
+			return &rebuilt, nil
+		},
+	}
+}
+
+// TestRebuildSnapshotUpdates_CallsClosureWithFreshParent verifies that
+// rebuildSnapshotUpdates invokes the rebuild closure and passes the fresh
+// branch head as freshParent.
+func TestRebuildSnapshotUpdates_CallsClosureWithFreshParent(t *testing.T) {
+	const oldManifest = "s3://bucket/old-manifest-list.avro"
+	const newManifest = "s3://bucket/new-manifest-list.avro"
+
+	// Build a fresh metadata that has a different snapshot than the one
+	// embedded in the update, so the parent-hasn't-changed guard is
+	// bypassed and the closure must be called.
+	freshHead := int64(42)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	parentID := int64(7) // original snap's parent — does NOT match freshHead (42)
+	snap := &Snapshot{
+		SnapshotID:       99,
+		ParentSnapshotID: &parentID,
+		ManifestList:     oldManifest,
+		Summary:          &Summary{Operation: OpAppend},
+	}
+
+	var receivedParent *Snapshot
+	upd := rebuildUpdate(snap, newManifest, &receivedParent)
+
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{upd},
+		freshMeta,
+		MainBranch,
+		iceio.LocalFS{},
+		1,
+	)
+	require.NoError(t, err)
+	require.Len(t, rebuilt, 1)
+	require.Len(t, orphaned, 1, "old manifest list must be recorded as orphaned")
+
+	// The closure should have been called with the branch's current head.
+	require.NotNil(t, receivedParent)
+	assert.Equal(t, freshHead, receivedParent.SnapshotID)
+
+	// The rebuilt update must carry the new manifest list.
+	addUpd, ok := rebuilt[0].(*addSnapshotUpdate)
+	require.True(t, ok)
+	assert.Equal(t, newManifest, addUpd.Snapshot.ManifestList)
+
+	// The superseded manifest list becomes an orphan.
+	assert.Equal(t, oldManifest, orphaned[0])
+}
+
+// TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged verifies that
+// rebuildSnapshotUpdates skips the rebuild when the update's snapshot
+// already has the fresh branch head as its parent (no-op retry).
+func TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged(t *testing.T) {
+	const manifest = "s3://bucket/manifest-list.avro"
+
+	freshHead := int64(42)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	// Parent already equals the fresh head — rebuild must be skipped.
+	snap := &Snapshot{
+		SnapshotID:       99,
+		ParentSnapshotID: &freshHead, // same as fresh head
+		ManifestList:     manifest,
+		Summary:          &Summary{Operation: OpAppend},
+	}
+
+	called := false
+	upd := &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, _ *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			called = true
+
+			return snap, nil
+		},
+	}
+
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{upd},
+		freshMeta,
+		MainBranch,
+		iceio.LocalFS{},
+		1,
+	)
+	require.NoError(t, err)
+	assert.False(t, called, "rebuild closure must not be called when parent is already up-to-date")
+	assert.Empty(t, orphaned, "no orphans when rebuild is skipped")
+	assert.Same(t, upd, rebuilt[0].(*addSnapshotUpdate), "original update must pass through unchanged")
+}
+
+// TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates verifies that
+// updates without a rebuildManifestList closure are returned unmodified.
+func TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates(t *testing.T) {
+	plainUpd := NewAddSnapshotUpdate(&Snapshot{
+		SnapshotID:   1,
+		ManifestList: "s3://bucket/no-rebuild.avro",
+		Summary:      &Summary{Operation: OpAppend},
+	})
+
+	// freshMeta may be nil — the plain update must not be touched.
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{plainUpd},
+		nil,
+		MainBranch,
+		iceio.LocalFS{},
+		0,
+	)
+	require.NoError(t, err)
+	assert.Empty(t, orphaned)
+	assert.Same(t, plainUpd, rebuilt[0].(*addSnapshotUpdate))
+}
+
+// TestRebuildSnapshotUpdates_PropagatesClosureError verifies that an error
+// returned by the rebuild closure surfaces as the function's return error.
+func TestRebuildSnapshotUpdates_PropagatesClosureError(t *testing.T) {
+	freshHead := int64(5)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	parent := int64(1)
+	snap := &Snapshot{
+		SnapshotID:       10,
+		ParentSnapshotID: &parent,
+		ManifestList:     "s3://bucket/old.avro",
+		Summary:          &Summary{Operation: OpAppend},
+	}
+	wantErr := errors.New("simulated S3 write failure")
+	upd := &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, _ *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			return nil, wantErr
+		},
+	}
+
+	_, _, err := rebuildSnapshotUpdates(t.Context(), []Update{upd}, freshMeta, MainBranch, iceio.LocalFS{}, 1)
+	assert.ErrorIs(t, err, wantErr)
+}
+
+// ---------------------------------------------------------------------------
+// Fix 3 — newSeq derived from freshMeta.LastSequenceNumber()
+// ---------------------------------------------------------------------------
+
+// TestRebuildFn_SeqNumDerivedFromFreshMeta verifies that the rebuilt snapshot's
+// SequenceNumber equals freshMeta.LastSequenceNumber()+1, NOT
+// freshParent.SequenceNumber+1. A concurrent writer on a different branch can
+// advance the table-wide last-sequence-number without advancing this branch's
+// parent, so using freshParent.SequenceNumber+1 would violate the spec invariant.
+func TestRebuildFn_SeqNumDerivedFromFreshMeta(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	txn, wfs := createTestTransactionWithMemIO(t, spec)
+
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, _, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	require.NotNil(t, addSnap.rebuildManifestList, "rebuildManifestList closure must be set")
+
+	// Simulate a concurrent writer on another branch that bumped the global
+	// last-sequence-number to 99 without advancing this branch's parent.
+	// old code: newSeq = capturedSnapshot.SequenceNumber (stale, ≤ 99 — spec violation)
+	// new code: newSeq = freshMeta.LastSequenceNumber() + 1 = 100
+	freshMeta := newMetadataWithLastSeqNum(t, 99)
+	require.Equal(t, int64(99), freshMeta.LastSequenceNumber())
+
+	rebuilt, err := addSnap.rebuildManifestList(context.Background(), freshMeta, nil, wfs, 1)

Review Comment:
   [Important] The headline scenario the fix targets — peer-branch advancement of `last-sequence-number` while this branch's parent stays put — is not actually exercised. The test passes `freshParent=nil`, so the buggy `freshParent != nil` branch (`newSeq = freshParent.SequenceNumber + 1`) never runs. An intermediate regression that reverts only the non-nil branch back to `freshParent.SequenceNumber + 1` would not be caught here, and `TestManifestListInheritedAfterConflict` doesn't catch it either because in that fixture both arithmetic paths happen to yield the same value. Strengthening: graft a peer-branch snapshot onto `freshMeta` with a higher `SequenceNumber` while keeping main's `freshParent.SequenceNumber` lower, then call the closure with both `freshMeta` and that lower `freshParent`, asserting `rebuilt.SequenceNumber == freshMeta.LastSequenceNumber()+1` — not `freshParent.SequenceNumber+1`. Alternatively, add `require.Equal(t, freshMeta.LastSequenceNumber()+1, rebuilt.
 SequenceNumber)` to the existing `TestRebuildFn_SummaryRebasedAgainstFreshParent` (which already passes both `freshParent` and `freshMeta`) — one-line strengthening that closes the gap.



##########
table/commit_retry_test.go:
##########
@@ -266,3 +307,244 @@ func TestReadRetryConfig_ClampsNegativeProperties(t *testing.T) {
 	assert.Equal(t, uint(CommitMaxRetryWaitMsDefault), cfg.maxWaitMs)
 	assert.Equal(t, uint(CommitTotalRetryTimeoutMsDefault), cfg.totalTimeoutMs)
 }
+
+// ---------------------------------------------------------------------------
+// Fix 5 — mandatory WriteFileIO check at top of doCommit
+// ---------------------------------------------------------------------------
+
+// TestDoCommit_NonWriteFileIOReturnsError verifies that doCommit fails
+// immediately when the table's file system does not implement WriteFileIO.
+// A silent skip would reuse the stale manifest list — exactly the bug
+// this PR was designed to fix.
+func TestDoCommit_NonWriteFileIOReturnsError(t *testing.T) {
+	cat := &flakyCatalog{}
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/rotest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	cat.metadata = meta
+
+	tbl := New(
+		Identifier{"db", "ro-test"},
+		meta, "file:///tmp/rotest/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return readOnlyIO{}, nil },
+		cat,
+	)
+
+	_, doErr := tbl.doCommit(t.Context(), nil, nil)
+	require.Error(t, doErr, "doCommit must fail when FS does not implement WriteFileIO")
+	assert.Contains(t, doErr.Error(), "WriteFileIO",

Review Comment:
   [Nit] `assert.Contains(t, doErr.Error(), "WriteFileIO")` is a substring match on the rendered error message — any rewording (e.g. "table requires writable IO") silently breaks the test. Define a sentinel in production (`var ErrWriteIORequired = errors.New(...)`) wrapped via `%w` and assert with `require.ErrorIs(t, doErr, ErrWriteIORequired)`. Same pattern would help `TestComputeOwnManifests_SnapshotByIDError` and `_ParentManifestsIOError`, which currently use `require.Error` and would accept any error including a programming bug recovered into one — anchor on `errors.Is(err, fs.ErrNotExist)` (or whatever sentinel the production code wraps) so the test pins the meaning, not just the presence of an error.



-- 
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

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

["mzzz-zzm (via GitHub)" <gi...@apache.org>]

mzzz-zzm commented on PR #982:
URL: https://github.com/apache/iceberg-go/pull/982#issuecomment-4399201682

   Refactored in commit `a6f6a0a` (Path A — clean fix, as you preferred).
   
   `MetadataBuilder` now exposes a public `AddSnapshotUpdate(*addSnapshotUpdate)` entry point that stores the update **as-is**, preserving both `ownManifests` and `rebuildManifestList`. `Apply` is now a one-liner that delegates to it:
   
   ```go
   // table/updates.go
   func (u *addSnapshotUpdate) Apply(builder *MetadataBuilder) error {
       return builder.AddSnapshotUpdate(u)
   }
   ```
   
   The `builder.updates[n-1]` reach-back is gone. Identity-pinning tests in `table/add_snapshot_update_test.go` now use `assert.Same` so any future change that constructs a fresh `*addSnapshotUpdate` inside `AddSnapshotUpdate` (the original "silent fix-disable" failure mode) fails loudly:
   
   - `TestAddSnapshotUpdate_PreservesIdentity` — the stored update pointer must equal the input pointer.
   - `TestAddSnapshotUpdate_PreservesRebuildClosure` — `ownManifests` and `rebuildManifestList` survive `Apply` end-to-end.
   - `TestAddSnapshotUpdate_ApplyRoutesThroughAddSnapshotUpdate` — `Apply` does not bypass the new method.
   - `TestAddSnapshotUpdate_NilNoOp` — guard for the historical no-snapshot case.
   
   Verified the new tests fail on the previous back-door implementation. Blast radius was small enough to make the refactor cleanly in this PR rather than punt to a follow-up.


-- 
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

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_r3202509210


##########
table/rebuild_manifest_test.go:
##########
@@ -0,0 +1,476 @@
+// 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.
+
+package table
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	iceio "github.com/apache/iceberg-go/io"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// newMetadataWithLastSeqNum builds a v2 Metadata whose last-sequence-number
+// equals lastSeqNum by grafting a synthetic snapshot at that sequence number.
+// Used by tests that need to control freshMeta.LastSequenceNumber().
+func newMetadataWithLastSeqNum(t *testing.T, lastSeqNum int64) Metadata {
+	t.Helper()
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/seqtest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	builder, err := MetadataBuilderFromBase(meta, "")
+	require.NoError(t, err)
+	snap := Snapshot{
+		SnapshotID:     1000,
+		SequenceNumber: lastSeqNum,
+		TimestampMs:    meta.LastUpdatedMillis() + 1,
+		Summary:        &Summary{Operation: OpAppend},
+	}
+	require.NoError(t, builder.AddSnapshot(&snap))
+	require.NoError(t, builder.SetSnapshotRef(MainBranch, 1000, BranchRef))
+	out, err := builder.Build()
+	require.NoError(t, err)
+
+	return out
+}
+
+// newV3MetadataWithNextRowID builds a v3 Metadata whose NextRowID() returns
+// nextRowID by adding a synthetic snapshot that consumes that many rows.
+// Used by tests that need to control freshMeta.NextRowID() for v3 row lineage.
+func newV3MetadataWithNextRowID(t *testing.T, nextRowID int64) Metadata {
+	t.Helper()
+	spec := iceberg.NewPartitionSpec()
+	txn, _ := createTestTransactionWithMemIO(t, spec)
+	txn.meta.formatVersion = 3
+
+	firstRowID := int64(0)
+	addedRows := nextRowID
+	snap := Snapshot{
+		SnapshotID:     1000,
+		SequenceNumber: 1,
+		TimestampMs:    txn.meta.base.LastUpdatedMillis() + 1,
+		Summary:        &Summary{Operation: OpAppend},
+		FirstRowID:     &firstRowID,
+		AddedRows:      &addedRows,
+	}
+	require.NoError(t, txn.meta.AddSnapshot(&snap))
+	require.NoError(t, txn.meta.SetSnapshotRef(MainBranch, 1000, BranchRef))
+
+	meta, err := txn.meta.Build()
+	require.NoError(t, err)
+	require.Equal(t, nextRowID, meta.NextRowID())
+
+	return meta
+}
+
+// ptr returns a pointer to v, used in test helper expressions.
+// NOTE: ptr is also declared in pos_delete_partitioned_fanout_writer_test.go;
+// both files share the same package so only one declaration is needed.
+// This comment documents the shared usage — do not re-add a declaration here.
+
+// rebuildUpdate constructs an addSnapshotUpdate whose rebuildManifestList
+// closure simply records the freshParent it received and returns a new
+// snapshot whose ManifestList is the given newManifestList value.
+func rebuildUpdate(snap *Snapshot, newManifestList string, gotParent **Snapshot) *addSnapshotUpdate {
+	return &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, freshParent *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			*gotParent = freshParent
+			rebuilt := *snap
+			rebuilt.ManifestList = newManifestList
+
+			return &rebuilt, nil
+		},
+	}
+}
+
+// TestRebuildSnapshotUpdates_CallsClosureWithFreshParent verifies that
+// rebuildSnapshotUpdates invokes the rebuild closure and passes the fresh
+// branch head as freshParent.
+func TestRebuildSnapshotUpdates_CallsClosureWithFreshParent(t *testing.T) {
+	const oldManifest = "s3://bucket/old-manifest-list.avro"
+	const newManifest = "s3://bucket/new-manifest-list.avro"
+
+	// Build a fresh metadata that has a different snapshot than the one
+	// embedded in the update, so the parent-hasn't-changed guard is
+	// bypassed and the closure must be called.
+	freshHead := int64(42)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	parentID := int64(7) // original snap's parent — does NOT match freshHead (42)
+	snap := &Snapshot{
+		SnapshotID:       99,
+		ParentSnapshotID: &parentID,
+		ManifestList:     oldManifest,
+		Summary:          &Summary{Operation: OpAppend},
+	}
+
+	var receivedParent *Snapshot
+	upd := rebuildUpdate(snap, newManifest, &receivedParent)
+
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{upd},
+		freshMeta,
+		MainBranch,
+		iceio.LocalFS{},
+		1,
+	)
+	require.NoError(t, err)
+	require.Len(t, rebuilt, 1)
+	require.Len(t, orphaned, 1, "old manifest list must be recorded as orphaned")
+
+	// The closure should have been called with the branch's current head.
+	require.NotNil(t, receivedParent)
+	assert.Equal(t, freshHead, receivedParent.SnapshotID)
+
+	// The rebuilt update must carry the new manifest list.
+	addUpd, ok := rebuilt[0].(*addSnapshotUpdate)
+	require.True(t, ok)
+	assert.Equal(t, newManifest, addUpd.Snapshot.ManifestList)
+
+	// The superseded manifest list becomes an orphan.
+	assert.Equal(t, oldManifest, orphaned[0])
+}
+
+// TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged verifies that
+// rebuildSnapshotUpdates skips the rebuild when the update's snapshot
+// already has the fresh branch head as its parent (no-op retry).
+func TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged(t *testing.T) {
+	const manifest = "s3://bucket/manifest-list.avro"
+
+	freshHead := int64(42)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	// Parent already equals the fresh head — rebuild must be skipped.
+	snap := &Snapshot{
+		SnapshotID:       99,
+		ParentSnapshotID: &freshHead, // same as fresh head
+		ManifestList:     manifest,
+		Summary:          &Summary{Operation: OpAppend},
+	}
+
+	called := false
+	upd := &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, _ *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			called = true
+
+			return snap, nil
+		},
+	}
+
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{upd},
+		freshMeta,
+		MainBranch,
+		iceio.LocalFS{},
+		1,
+	)
+	require.NoError(t, err)
+	assert.False(t, called, "rebuild closure must not be called when parent is already up-to-date")
+	assert.Empty(t, orphaned, "no orphans when rebuild is skipped")
+	assert.Same(t, upd, rebuilt[0].(*addSnapshotUpdate), "original update must pass through unchanged")
+}
+
+// TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates verifies that
+// updates without a rebuildManifestList closure are returned unmodified.
+func TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates(t *testing.T) {
+	plainUpd := NewAddSnapshotUpdate(&Snapshot{
+		SnapshotID:   1,
+		ManifestList: "s3://bucket/no-rebuild.avro",
+		Summary:      &Summary{Operation: OpAppend},
+	})
+
+	// freshMeta may be nil — the plain update must not be touched.
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{plainUpd},
+		nil,
+		MainBranch,
+		iceio.LocalFS{},
+		0,
+	)
+	require.NoError(t, err)
+	assert.Empty(t, orphaned)
+	assert.Same(t, plainUpd, rebuilt[0].(*addSnapshotUpdate))
+}
+
+// TestRebuildSnapshotUpdates_PropagatesClosureError verifies that an error
+// returned by the rebuild closure surfaces as the function's return error.
+func TestRebuildSnapshotUpdates_PropagatesClosureError(t *testing.T) {
+	freshHead := int64(5)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	parent := int64(1)
+	snap := &Snapshot{
+		SnapshotID:       10,
+		ParentSnapshotID: &parent,
+		ManifestList:     "s3://bucket/old.avro",
+		Summary:          &Summary{Operation: OpAppend},
+	}
+	wantErr := errors.New("simulated S3 write failure")
+	upd := &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, _ *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			return nil, wantErr
+		},
+	}
+
+	_, _, err := rebuildSnapshotUpdates(t.Context(), []Update{upd}, freshMeta, MainBranch, iceio.LocalFS{}, 1)
+	assert.ErrorIs(t, err, wantErr)
+}
+
+// ---------------------------------------------------------------------------
+// Fix 3 — newSeq derived from freshMeta.LastSequenceNumber()
+// ---------------------------------------------------------------------------
+
+// TestRebuildFn_SeqNumDerivedFromFreshMeta verifies that the rebuilt snapshot's
+// SequenceNumber equals freshMeta.LastSequenceNumber()+1, NOT
+// freshParent.SequenceNumber+1. A concurrent writer on a different branch can
+// advance the table-wide last-sequence-number without advancing this branch's
+// parent, so using freshParent.SequenceNumber+1 would violate the spec invariant.
+func TestRebuildFn_SeqNumDerivedFromFreshMeta(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	txn, wfs := createTestTransactionWithMemIO(t, spec)
+
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, _, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	require.NotNil(t, addSnap.rebuildManifestList, "rebuildManifestList closure must be set")
+
+	// Simulate a concurrent writer on another branch that bumped the global
+	// last-sequence-number to 99 without advancing this branch's parent.
+	// old code: newSeq = capturedSnapshot.SequenceNumber (stale, ≤ 99 — spec violation)
+	// new code: newSeq = freshMeta.LastSequenceNumber() + 1 = 100
+	freshMeta := newMetadataWithLastSeqNum(t, 99)
+	require.Equal(t, int64(99), freshMeta.LastSequenceNumber())
+
+	rebuilt, err := addSnap.rebuildManifestList(context.Background(), freshMeta, nil, wfs, 1)
+	require.NoError(t, err)
+	require.Equal(t, int64(100), rebuilt.SequenceNumber,
+		"SequenceNumber must be freshMeta.LastSequenceNumber()+1; using freshParent.SequenceNumber+1 violates the spec when another branch advanced the global counter")
+}
+
+// TestRebuildFn_SeqNumV1TableIsZero verifies that v1 tables keep SequenceNumber == 0
+// regardless of what freshMeta reports (v1 does not use sequence numbers).
+func TestRebuildFn_SeqNumV1TableIsZero(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	txn, wfs := createTestTransactionWithMemIO(t, spec)
+	txn.meta.formatVersion = 1 // override to v1
+
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, _, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	require.NotNil(t, addSnap.rebuildManifestList)
+
+	freshMeta := newMetadataWithLastSeqNum(t, 99)
+	rebuilt, err := addSnap.rebuildManifestList(context.Background(), freshMeta, nil, wfs, 1)
+	require.NoError(t, err)
+	require.Equal(t, int64(0), rebuilt.SequenceNumber, "v1 tables must always have SequenceNumber == 0")
+}
+
+// ---------------------------------------------------------------------------
+// Fix 1 — firstRowID derived from freshMeta.NextRowID()
+// ---------------------------------------------------------------------------
+
+// TestRebuildFn_V3FirstRowIDDerivedFromFreshMeta verifies that on a v3 table
+// the rebuilt snapshot's FirstRowID equals freshMeta.NextRowID() rather than
+// the hardcoded 0. If two writers race and the peer commits first, the
+// catalog's nextRowID has already advanced; using 0 would produce a
+// first-row-id that disagrees with the catalog's view and fails row-lineage
+// validation.
+func TestRebuildFn_V3FirstRowIDDerivedFromFreshMeta(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	txn, wfs := createTestTransactionWithMemIO(t, spec)
+	txn.meta.formatVersion = 3
+
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, _, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	require.NotNil(t, addSnap.rebuildManifestList)
+
+	// Simulate a concurrent writer that added 50 rows, advancing nextRowID to 50.
+	// old code: firstRowID = 0 (hardcoded)
+	// new code: firstRowID = freshMeta.NextRowID() = 50
+	freshMeta := newV3MetadataWithNextRowID(t, 50)
+	require.Equal(t, int64(50), freshMeta.NextRowID(), "freshMeta.NextRowID() must be 50")
+
+	rebuilt, err := addSnap.rebuildManifestList(context.Background(), freshMeta, nil, wfs, 1)
+	require.NoError(t, err)
+	require.NotNil(t, rebuilt.FirstRowID, "v3 rebuilt snapshot must have FirstRowID")
+	require.Equal(t, int64(50), *rebuilt.FirstRowID,
+		"FirstRowID must equal freshMeta.NextRowID(); hardcoded 0 would conflict with catalog's advanced nextRowID")
+
+	// AddedRows must be recomputed from writer.NextRowID() - firstRowID for
+	// this attempt (NOT carried over from capturedSnapshot at attempt 0).
+	// The single test data file has record count 1, so the producer's own
+	// contribution this attempt is exactly 1 row.
+	require.NotNil(t, rebuilt.AddedRows, "v3 rebuilt snapshot must have AddedRows")
+	require.Equal(t, int64(1), *rebuilt.AddedRows,

Review Comment:
   The producer's single data file has record-count 1, so `capturedSnapshot.AddedRows == 1` and `*rebuilt.AddedRows == 1` line up by accident. A revert of the recompute path (carrying the captured AddedRows over instead of deriving from `writer.NextRowID() - firstRowID`) still passes this assertion — same shape in `TestRebuildFn_V3FirstRowIDZeroWhenNilNextRowID`.
   
   I'd reshape the fixture so captured ≠ recomputed. Seeding `capturedSnapshot.AddedRows` to something distinct (e.g. 99) before the rebuild, or using a producer whose data files sum to a different count, makes the recompute the only path that passes.



##########
table/rebuild_manifest_test.go:
##########
@@ -0,0 +1,476 @@
+// 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.
+
+package table
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	iceio "github.com/apache/iceberg-go/io"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// newMetadataWithLastSeqNum builds a v2 Metadata whose last-sequence-number
+// equals lastSeqNum by grafting a synthetic snapshot at that sequence number.
+// Used by tests that need to control freshMeta.LastSequenceNumber().
+func newMetadataWithLastSeqNum(t *testing.T, lastSeqNum int64) Metadata {
+	t.Helper()
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/seqtest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	builder, err := MetadataBuilderFromBase(meta, "")
+	require.NoError(t, err)
+	snap := Snapshot{
+		SnapshotID:     1000,
+		SequenceNumber: lastSeqNum,
+		TimestampMs:    meta.LastUpdatedMillis() + 1,
+		Summary:        &Summary{Operation: OpAppend},
+	}
+	require.NoError(t, builder.AddSnapshot(&snap))
+	require.NoError(t, builder.SetSnapshotRef(MainBranch, 1000, BranchRef))
+	out, err := builder.Build()
+	require.NoError(t, err)
+
+	return out
+}
+
+// newV3MetadataWithNextRowID builds a v3 Metadata whose NextRowID() returns
+// nextRowID by adding a synthetic snapshot that consumes that many rows.
+// Used by tests that need to control freshMeta.NextRowID() for v3 row lineage.
+func newV3MetadataWithNextRowID(t *testing.T, nextRowID int64) Metadata {
+	t.Helper()
+	spec := iceberg.NewPartitionSpec()
+	txn, _ := createTestTransactionWithMemIO(t, spec)
+	txn.meta.formatVersion = 3
+
+	firstRowID := int64(0)
+	addedRows := nextRowID
+	snap := Snapshot{
+		SnapshotID:     1000,
+		SequenceNumber: 1,
+		TimestampMs:    txn.meta.base.LastUpdatedMillis() + 1,
+		Summary:        &Summary{Operation: OpAppend},
+		FirstRowID:     &firstRowID,
+		AddedRows:      &addedRows,
+	}
+	require.NoError(t, txn.meta.AddSnapshot(&snap))
+	require.NoError(t, txn.meta.SetSnapshotRef(MainBranch, 1000, BranchRef))
+
+	meta, err := txn.meta.Build()
+	require.NoError(t, err)
+	require.Equal(t, nextRowID, meta.NextRowID())
+
+	return meta
+}
+
+// ptr returns a pointer to v, used in test helper expressions.
+// NOTE: ptr is also declared in pos_delete_partitioned_fanout_writer_test.go;
+// both files share the same package so only one declaration is needed.
+// This comment documents the shared usage — do not re-add a declaration here.
+
+// rebuildUpdate constructs an addSnapshotUpdate whose rebuildManifestList
+// closure simply records the freshParent it received and returns a new
+// snapshot whose ManifestList is the given newManifestList value.
+func rebuildUpdate(snap *Snapshot, newManifestList string, gotParent **Snapshot) *addSnapshotUpdate {
+	return &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, freshParent *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			*gotParent = freshParent
+			rebuilt := *snap
+			rebuilt.ManifestList = newManifestList
+
+			return &rebuilt, nil
+		},
+	}
+}
+
+// TestRebuildSnapshotUpdates_CallsClosureWithFreshParent verifies that
+// rebuildSnapshotUpdates invokes the rebuild closure and passes the fresh
+// branch head as freshParent.
+func TestRebuildSnapshotUpdates_CallsClosureWithFreshParent(t *testing.T) {
+	const oldManifest = "s3://bucket/old-manifest-list.avro"
+	const newManifest = "s3://bucket/new-manifest-list.avro"
+
+	// Build a fresh metadata that has a different snapshot than the one
+	// embedded in the update, so the parent-hasn't-changed guard is
+	// bypassed and the closure must be called.
+	freshHead := int64(42)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	parentID := int64(7) // original snap's parent — does NOT match freshHead (42)
+	snap := &Snapshot{
+		SnapshotID:       99,
+		ParentSnapshotID: &parentID,
+		ManifestList:     oldManifest,
+		Summary:          &Summary{Operation: OpAppend},
+	}
+
+	var receivedParent *Snapshot
+	upd := rebuildUpdate(snap, newManifest, &receivedParent)
+
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{upd},
+		freshMeta,
+		MainBranch,
+		iceio.LocalFS{},
+		1,
+	)
+	require.NoError(t, err)
+	require.Len(t, rebuilt, 1)
+	require.Len(t, orphaned, 1, "old manifest list must be recorded as orphaned")
+
+	// The closure should have been called with the branch's current head.
+	require.NotNil(t, receivedParent)
+	assert.Equal(t, freshHead, receivedParent.SnapshotID)
+
+	// The rebuilt update must carry the new manifest list.
+	addUpd, ok := rebuilt[0].(*addSnapshotUpdate)
+	require.True(t, ok)
+	assert.Equal(t, newManifest, addUpd.Snapshot.ManifestList)
+
+	// The superseded manifest list becomes an orphan.
+	assert.Equal(t, oldManifest, orphaned[0])
+}
+
+// TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged verifies that
+// rebuildSnapshotUpdates skips the rebuild when the update's snapshot
+// already has the fresh branch head as its parent (no-op retry).
+func TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged(t *testing.T) {
+	const manifest = "s3://bucket/manifest-list.avro"
+
+	freshHead := int64(42)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	// Parent already equals the fresh head — rebuild must be skipped.
+	snap := &Snapshot{
+		SnapshotID:       99,
+		ParentSnapshotID: &freshHead, // same as fresh head
+		ManifestList:     manifest,
+		Summary:          &Summary{Operation: OpAppend},
+	}
+
+	called := false
+	upd := &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, _ *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			called = true
+
+			return snap, nil
+		},
+	}
+
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{upd},
+		freshMeta,
+		MainBranch,
+		iceio.LocalFS{},
+		1,
+	)
+	require.NoError(t, err)
+	assert.False(t, called, "rebuild closure must not be called when parent is already up-to-date")
+	assert.Empty(t, orphaned, "no orphans when rebuild is skipped")
+	assert.Same(t, upd, rebuilt[0].(*addSnapshotUpdate), "original update must pass through unchanged")
+}
+
+// TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates verifies that
+// updates without a rebuildManifestList closure are returned unmodified.
+func TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates(t *testing.T) {
+	plainUpd := NewAddSnapshotUpdate(&Snapshot{
+		SnapshotID:   1,
+		ManifestList: "s3://bucket/no-rebuild.avro",
+		Summary:      &Summary{Operation: OpAppend},
+	})
+
+	// freshMeta may be nil — the plain update must not be touched.
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{plainUpd},
+		nil,
+		MainBranch,
+		iceio.LocalFS{},
+		0,
+	)
+	require.NoError(t, err)
+	assert.Empty(t, orphaned)
+	assert.Same(t, plainUpd, rebuilt[0].(*addSnapshotUpdate))
+}
+
+// TestRebuildSnapshotUpdates_PropagatesClosureError verifies that an error
+// returned by the rebuild closure surfaces as the function's return error.
+func TestRebuildSnapshotUpdates_PropagatesClosureError(t *testing.T) {
+	freshHead := int64(5)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	parent := int64(1)
+	snap := &Snapshot{
+		SnapshotID:       10,
+		ParentSnapshotID: &parent,
+		ManifestList:     "s3://bucket/old.avro",
+		Summary:          &Summary{Operation: OpAppend},
+	}
+	wantErr := errors.New("simulated S3 write failure")
+	upd := &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, _ *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			return nil, wantErr
+		},
+	}
+
+	_, _, err := rebuildSnapshotUpdates(t.Context(), []Update{upd}, freshMeta, MainBranch, iceio.LocalFS{}, 1)
+	assert.ErrorIs(t, err, wantErr)
+}
+
+// ---------------------------------------------------------------------------
+// Fix 3 — newSeq derived from freshMeta.LastSequenceNumber()
+// ---------------------------------------------------------------------------
+
+// TestRebuildFn_SeqNumDerivedFromFreshMeta verifies that the rebuilt snapshot's
+// SequenceNumber equals freshMeta.LastSequenceNumber()+1, NOT
+// freshParent.SequenceNumber+1. A concurrent writer on a different branch can
+// advance the table-wide last-sequence-number without advancing this branch's
+// parent, so using freshParent.SequenceNumber+1 would violate the spec invariant.
+func TestRebuildFn_SeqNumDerivedFromFreshMeta(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	txn, wfs := createTestTransactionWithMemIO(t, spec)
+
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, _, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	require.NotNil(t, addSnap.rebuildManifestList, "rebuildManifestList closure must be set")
+
+	// Simulate a concurrent writer on another branch that bumped the global
+	// last-sequence-number to 99 without advancing this branch's parent.
+	// old code: newSeq = capturedSnapshot.SequenceNumber (stale, ≤ 99 — spec violation)
+	// new code: newSeq = freshMeta.LastSequenceNumber() + 1 = 100
+	freshMeta := newMetadataWithLastSeqNum(t, 99)
+	require.Equal(t, int64(99), freshMeta.LastSequenceNumber())
+
+	rebuilt, err := addSnap.rebuildManifestList(context.Background(), freshMeta, nil, wfs, 1)

Review Comment:
   `freshParent=nil` here, so the closure takes the nil branch and the actual fix site — `freshParent != nil` switching from `freshParent.SequenceNumber+1` to `freshMeta.LastSequenceNumber()+1` — has zero coverage. The peer-branch-advances-the-table-wide-seqnum scenario from the commit message never gets exercised.
   
   `TestRebuildFn_SummaryRebasedAgainstFreshParent` already wires both a non-nil freshParent and a freshMeta. Adding `require.Equal(t, freshMeta.LastSequenceNumber()+1, rebuilt.SequenceNumber)` there pins the freshParent branch without a new test.



##########
table/commit_retry_test.go:
##########
@@ -266,3 +307,244 @@ func TestReadRetryConfig_ClampsNegativeProperties(t *testing.T) {
 	assert.Equal(t, uint(CommitMaxRetryWaitMsDefault), cfg.maxWaitMs)
 	assert.Equal(t, uint(CommitTotalRetryTimeoutMsDefault), cfg.totalTimeoutMs)
 }
+
+// ---------------------------------------------------------------------------
+// Fix 5 — mandatory WriteFileIO check at top of doCommit
+// ---------------------------------------------------------------------------
+
+// TestDoCommit_NonWriteFileIOReturnsError verifies that doCommit fails
+// immediately when the table's file system does not implement WriteFileIO.
+// A silent skip would reuse the stale manifest list — exactly the bug
+// this PR was designed to fix.
+func TestDoCommit_NonWriteFileIOReturnsError(t *testing.T) {
+	cat := &flakyCatalog{}
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/rotest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	cat.metadata = meta
+
+	tbl := New(
+		Identifier{"db", "ro-test"},
+		meta, "file:///tmp/rotest/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return readOnlyIO{}, nil },
+		cat,
+	)
+
+	_, doErr := tbl.doCommit(t.Context(), nil, nil)
+	require.Error(t, doErr, "doCommit must fail when FS does not implement WriteFileIO")
+	assert.Contains(t, doErr.Error(), "WriteFileIO",
+		"error message must mention WriteFileIO so callers understand the requirement")
+	assert.Equal(t, int32(0), cat.attempts.Load(),
+		"CommitTable must not be called when FS check fails")
+}
+
+// ---------------------------------------------------------------------------
+// Fix 6 — orphan cleanup via defer
+// ---------------------------------------------------------------------------
+
+// newOCCTable creates a table that uses the given wfs for its FS and the given
+// catalog for commits. meta should include retry-config properties so that
+// doCommit's retry loop allows at least one retry.
+func newOCCTable(t *testing.T, meta Metadata, wfs iceio.WriteFileIO, cat CatalogIO) *Table {
+	t.Helper()
+
+	return New(
+		Identifier{"db", "occ-cleanup-test"},
+		meta,
+		"mem://default/table-location/metadata/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return wfs, nil },
+		cat,
+	)
+}
+
+// newMemIOWithRetryMeta creates a test memIO and a matching table Metadata that
+// includes retry-config properties, so doCommit's retry loop allows retries.
+// The location matches createTestTransactionWithMemIO so they share the same
+// memIO for writing manifest files.
+func newMemIOWithRetryMeta(t *testing.T, spec iceberg.PartitionSpec) (*memIO, Metadata) {
+	t.Helper()
+	wfs := newMemIO(1<<20, nil)
+	schema := simpleSchema()
+	meta, err := NewMetadata(schema, &spec, UnsortedSortOrder, "mem://default/table-location",
+		iceberg.Properties{
+			CommitNumRetriesKey:          "3",
+			CommitMinRetryWaitMsKey:      "1",
+			CommitMaxRetryWaitMsKey:      "2",
+			CommitTotalRetryTimeoutMsKey: "60000",
+		})
+	require.NoError(t, err, "new metadata")
+
+	return wfs, meta
+}
+
+// TestDoCommit_OrphanCleanedOnSuccess verifies that manifest-list files
+// orphaned by OCC retries are removed after a successful commit. These files
+// are written during rebuild and must not leak on the happy path.
+func TestDoCommit_OrphanCleanedOnSuccess(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	// Build a transaction from the retry-enabled meta and commit via the producer.
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	// Pre-populate the original manifest list path in the IO so that the
+	// defer's wfs.Remove call has a concrete entry to delete.
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	// Catalog: fail once with ErrCommitFailed (triggers rebuild that orphans
+	// originalManifestList), then succeed.
+	cat := &sequentialCatalog{
+		metadata: meta,
+		errs:     []error{ErrCommitFailed},
+	}
+	tbl = newOCCTable(t, meta, wfs, cat)
+
+	_, err = tbl.doCommit(t.Context(), updates, reqs, withCommitBranch(MainBranch))
+	require.NoError(t, err, "doCommit must succeed on the second attempt")
+
+	_, stillExists := wfs.files[originalManifestList]
+	assert.False(t, stillExists,
+		"orphaned manifest list must be removed after successful commit")
+}
+
+// TestDoCommit_OrphanNotCleanedOnUnknownError verifies that manifest-list
+// files are NOT removed when CommitTable returns an unknown non-ErrCommitFailed
+// error (5xx / gateway timeout). In that case the catalog may have silently
+// accepted the commit, meaning one of the "orphaned" files is actually the
+// live snapshot. Deleting it would permanently corrupt the table.
+func TestDoCommit_OrphanNotCleanedOnUnknownError(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	unknown5xxErr := errors.New("simulated 5xx: internal server error")
+	// Catalog: fail once (ErrCommitFailed → rebuild → orphan created),
+	// then return a 5xx (non-ErrCommitFailed → cleanupOrphans=false).
+	cat := &sequentialCatalog{
+		metadata: meta,
+		errs:     []error{ErrCommitFailed, unknown5xxErr},
+	}
+	tbl = newOCCTable(t, meta, wfs, cat)
+
+	_, err = tbl.doCommit(t.Context(), updates, reqs, withCommitBranch(MainBranch))
+	require.ErrorIs(t, err, unknown5xxErr, "5xx error must propagate")
+
+	_, stillExists := wfs.files[originalManifestList]
+	assert.True(t, stillExists,
+		"orphaned manifest list must NOT be removed when commit outcome is unknown (5xx)")
+}
+
+// TestDoCommit_OrphanCleanedOnCommitDiverged verifies that manifest-list files
+// orphaned by rebuild attempts are removed when ErrCommitDiverged is returned
+// by a conflict validator. Diverged commits are terminal (no retry), and since
+// neither of the orphaned files was ever accepted by the catalog, they are safe
+// to delete. The defer cleanup runs with cleanupOrphans=true on this path.
+func TestDoCommit_OrphanCleanedOnCommitDiverged(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	// freshMeta has a snapshot on MainBranch so validators run on the retry attempt.
+	freshID := int64(42)
+	freshMeta := newConflictTestMetadataWithProps(t, &freshID, iceberg.Properties{
+		CommitNumRetriesKey:          "3",
+		CommitMinRetryWaitMsKey:      "1",
+		CommitMaxRetryWaitMsKey:      "2",
+		CommitTotalRetryTimeoutMsKey: "60000",
+	})
+
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	// Catalog: fail once (ErrCommitFailed → rebuild → orphan created).
+	// LoadTable returns freshMeta (has branch snapshot → validators run on retry).
+	// Validator returns ErrCommitDiverged immediately.
+	cat := &sequentialCatalog{
+		metadata: meta,
+		loadMeta: freshMeta,
+		errs:     []error{ErrCommitFailed},
+	}
+	tbl = newOCCTable(t, meta, wfs, cat)
+
+	divergedValidator := func(*conflictContext) error { return ErrCommitDiverged }
+
+	_, err = tbl.doCommit(t.Context(), updates, reqs,
+		withCommitBranch(MainBranch),
+		withCommitValidators(divergedValidator),
+	)
+	require.ErrorIs(t, err, ErrCommitDiverged)
+
+	// The defer must have fired with cleanupOrphans=true and removed the orphan.
+	_, stillExists := wfs.files[originalManifestList]
+	assert.False(t, stillExists,
+		"orphaned manifest list must be removed even on ErrCommitDiverged: "+
+			"the file was never accepted by the catalog so it is safe to delete")
+}
+
+// TestDoCommit_OrphanCleanedOnRetriesExhausted verifies that when every retry
+// attempt fails with ErrCommitFailed and the loop exits with the budget
+// exhausted, the defer still fires with cleanupOrphans=true. None of the
+// orphaned manifest-list files were ever accepted by the catalog, so they are
+// safe to delete on this terminal exit.
+func TestDoCommit_OrphanCleanedOnRetriesExhausted(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	// numRetries=3 → 4 attempts; every attempt fails with ErrCommitFailed.
+	cat := &sequentialCatalog{
+		metadata: meta,
+		errs:     []error{ErrCommitFailed, ErrCommitFailed, ErrCommitFailed, ErrCommitFailed},

Review Comment:
   `sequentialCatalog.metadata` is the same object across all four LoadTable calls, so on each retry the closure sees a freshHead identical to base. None of the per-retry recompute (newSeq from `freshMeta.LastSequenceNumber()`, firstRowID from `freshMeta.NextRowID()`, summary rebased against freshParent) is actually exercised — this test pins terminal cleanup, not retry behavior.
   
   `headTrackingCatalog` in `commit_refresh_replay_test.go` already advances per attempt. One ≥2-retry test using it would close more coverage than these four orphan variants combined — worth a follow-up?



##########
table/commit_retry_test.go:
##########
@@ -266,3 +307,244 @@ func TestReadRetryConfig_ClampsNegativeProperties(t *testing.T) {
 	assert.Equal(t, uint(CommitMaxRetryWaitMsDefault), cfg.maxWaitMs)
 	assert.Equal(t, uint(CommitTotalRetryTimeoutMsDefault), cfg.totalTimeoutMs)
 }
+
+// ---------------------------------------------------------------------------
+// Fix 5 — mandatory WriteFileIO check at top of doCommit
+// ---------------------------------------------------------------------------
+
+// TestDoCommit_NonWriteFileIOReturnsError verifies that doCommit fails
+// immediately when the table's file system does not implement WriteFileIO.
+// A silent skip would reuse the stale manifest list — exactly the bug
+// this PR was designed to fix.
+func TestDoCommit_NonWriteFileIOReturnsError(t *testing.T) {
+	cat := &flakyCatalog{}
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/rotest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	cat.metadata = meta
+
+	tbl := New(
+		Identifier{"db", "ro-test"},
+		meta, "file:///tmp/rotest/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return readOnlyIO{}, nil },
+		cat,
+	)
+
+	_, doErr := tbl.doCommit(t.Context(), nil, nil)
+	require.Error(t, doErr, "doCommit must fail when FS does not implement WriteFileIO")
+	assert.Contains(t, doErr.Error(), "WriteFileIO",
+		"error message must mention WriteFileIO so callers understand the requirement")
+	assert.Equal(t, int32(0), cat.attempts.Load(),
+		"CommitTable must not be called when FS check fails")
+}
+
+// ---------------------------------------------------------------------------
+// Fix 6 — orphan cleanup via defer
+// ---------------------------------------------------------------------------
+
+// newOCCTable creates a table that uses the given wfs for its FS and the given
+// catalog for commits. meta should include retry-config properties so that
+// doCommit's retry loop allows at least one retry.
+func newOCCTable(t *testing.T, meta Metadata, wfs iceio.WriteFileIO, cat CatalogIO) *Table {
+	t.Helper()
+
+	return New(
+		Identifier{"db", "occ-cleanup-test"},
+		meta,
+		"mem://default/table-location/metadata/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return wfs, nil },
+		cat,
+	)
+}
+
+// newMemIOWithRetryMeta creates a test memIO and a matching table Metadata that
+// includes retry-config properties, so doCommit's retry loop allows retries.
+// The location matches createTestTransactionWithMemIO so they share the same
+// memIO for writing manifest files.
+func newMemIOWithRetryMeta(t *testing.T, spec iceberg.PartitionSpec) (*memIO, Metadata) {
+	t.Helper()
+	wfs := newMemIO(1<<20, nil)
+	schema := simpleSchema()
+	meta, err := NewMetadata(schema, &spec, UnsortedSortOrder, "mem://default/table-location",
+		iceberg.Properties{
+			CommitNumRetriesKey:          "3",
+			CommitMinRetryWaitMsKey:      "1",
+			CommitMaxRetryWaitMsKey:      "2",
+			CommitTotalRetryTimeoutMsKey: "60000",
+		})
+	require.NoError(t, err, "new metadata")
+
+	return wfs, meta
+}
+
+// TestDoCommit_OrphanCleanedOnSuccess verifies that manifest-list files
+// orphaned by OCC retries are removed after a successful commit. These files
+// are written during rebuild and must not leak on the happy path.
+func TestDoCommit_OrphanCleanedOnSuccess(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	// Build a transaction from the retry-enabled meta and commit via the producer.
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	// Pre-populate the original manifest list path in the IO so that the
+	// defer's wfs.Remove call has a concrete entry to delete.
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	// Catalog: fail once with ErrCommitFailed (triggers rebuild that orphans
+	// originalManifestList), then succeed.
+	cat := &sequentialCatalog{
+		metadata: meta,
+		errs:     []error{ErrCommitFailed},
+	}
+	tbl = newOCCTable(t, meta, wfs, cat)
+
+	_, err = tbl.doCommit(t.Context(), updates, reqs, withCommitBranch(MainBranch))
+	require.NoError(t, err, "doCommit must succeed on the second attempt")
+
+	_, stillExists := wfs.files[originalManifestList]

Review Comment:
   `memIO.Create` returns a `limitedWriteCloser` that doesn't write back into `wfs.files`, so neither the original nor the rebuilt manifest list is actually present in the FS. The test only sees `originalManifestList` because of the manual placeholder write on line 403 — meaning a regression that flipped the cleanup to delete the *committed* path instead of the orphan would slip through, since the committed path also isn't there.
   
   Either persist `Create` writes into `wfs.files`, or add an assertion that `cat.metadata.CurrentSnapshot().ManifestList` differs from `originalManifestList` and was not removed. Same shape applies to the other three orphan-cleanup tests in this file.



##########
table/commit_retry_test.go:
##########
@@ -266,3 +307,244 @@ func TestReadRetryConfig_ClampsNegativeProperties(t *testing.T) {
 	assert.Equal(t, uint(CommitMaxRetryWaitMsDefault), cfg.maxWaitMs)
 	assert.Equal(t, uint(CommitTotalRetryTimeoutMsDefault), cfg.totalTimeoutMs)
 }
+
+// ---------------------------------------------------------------------------
+// Fix 5 — mandatory WriteFileIO check at top of doCommit
+// ---------------------------------------------------------------------------
+
+// TestDoCommit_NonWriteFileIOReturnsError verifies that doCommit fails
+// immediately when the table's file system does not implement WriteFileIO.
+// A silent skip would reuse the stale manifest list — exactly the bug
+// this PR was designed to fix.
+func TestDoCommit_NonWriteFileIOReturnsError(t *testing.T) {
+	cat := &flakyCatalog{}
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/rotest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	cat.metadata = meta
+
+	tbl := New(
+		Identifier{"db", "ro-test"},
+		meta, "file:///tmp/rotest/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return readOnlyIO{}, nil },
+		cat,
+	)
+
+	_, doErr := tbl.doCommit(t.Context(), nil, nil)
+	require.Error(t, doErr, "doCommit must fail when FS does not implement WriteFileIO")
+	assert.Contains(t, doErr.Error(), "WriteFileIO",

Review Comment:
   While here — substring match on `"WriteFileIO"` is brittle to any reword of the error. `TestComputeOwnManifests_SnapshotByIDError` and `_ParentManifestsIOError` use bare `require.Error` for the same shape. A sentinel + `errors.Is` would be the more durable form.



-- 
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

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

["mzzz-zzm (via GitHub)" <gi...@apache.org>]

mzzz-zzm commented on code in PR #982:
URL: https://github.com/apache/iceberg-go/pull/982#discussion_r3202226525


##########
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:
   The explicit post-loop cleanup block is replaced with a `defer` guarded by a `cleanupOrphans` flag (default `true`):
   
   ```go
   cleanupOrphans := true
   defer func() {
       if !cleanupOrphans || len(orphanedManifests) == 0 {
           return
       }
       for _, path := range orphanedManifests {
           if removeErr := wfs.Remove(path); removeErr != nil {
               log.Printf("Warning: failed to delete orphaned manifest list %s: %v", path, removeErr)
           }
       }
   }()
   ```
   
   The defer fires on every exit path. `cleanupOrphans = false` is set only for the non-`ErrCommitFailed` `CommitTable` error (5xx / unknown-state), where one of the orphaned files may actually be the snapshot the catalog silently accepted:
   
   ```go
   if !errors.Is(err, ErrCommitFailed) {
       cleanupOrphans = false  // outcome unknown; catalog may have accepted an orphan
       return nil, err
   }
   ```
   
   All other exits — success, `ErrCommitDiverged`, validator errors, context cancellation — leave `cleanupOrphans = true` and the defer removes the orphaned files.
   
   Tests added: `TestDoCommit_OrphanCleanedOnSuccess`, `TestDoCommit_OrphanNotCleanedOnUnknownError`, `TestDoCommit_OrphanCleanedOnCommitDiverged`, `TestDoCommit_OrphanCleanedOnRetriesExhausted` (every attempt fails with `ErrCommitFailed` until budget is exhausted — defer fires with `cleanupOrphans=true`; orphans removed).



-- 
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

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_r3202392992


##########
table/commit_retry_test.go:
##########
@@ -266,3 +307,244 @@ func TestReadRetryConfig_ClampsNegativeProperties(t *testing.T) {
 	assert.Equal(t, uint(CommitMaxRetryWaitMsDefault), cfg.maxWaitMs)
 	assert.Equal(t, uint(CommitTotalRetryTimeoutMsDefault), cfg.totalTimeoutMs)
 }
+
+// ---------------------------------------------------------------------------
+// Fix 5 — mandatory WriteFileIO check at top of doCommit
+// ---------------------------------------------------------------------------
+
+// TestDoCommit_NonWriteFileIOReturnsError verifies that doCommit fails
+// immediately when the table's file system does not implement WriteFileIO.
+// A silent skip would reuse the stale manifest list — exactly the bug
+// this PR was designed to fix.
+func TestDoCommit_NonWriteFileIOReturnsError(t *testing.T) {
+	cat := &flakyCatalog{}
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/rotest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	cat.metadata = meta
+
+	tbl := New(
+		Identifier{"db", "ro-test"},
+		meta, "file:///tmp/rotest/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return readOnlyIO{}, nil },
+		cat,
+	)
+
+	_, doErr := tbl.doCommit(t.Context(), nil, nil)
+	require.Error(t, doErr, "doCommit must fail when FS does not implement WriteFileIO")
+	assert.Contains(t, doErr.Error(), "WriteFileIO",
+		"error message must mention WriteFileIO so callers understand the requirement")
+	assert.Equal(t, int32(0), cat.attempts.Load(),
+		"CommitTable must not be called when FS check fails")
+}
+
+// ---------------------------------------------------------------------------
+// Fix 6 — orphan cleanup via defer
+// ---------------------------------------------------------------------------
+
+// newOCCTable creates a table that uses the given wfs for its FS and the given
+// catalog for commits. meta should include retry-config properties so that
+// doCommit's retry loop allows at least one retry.
+func newOCCTable(t *testing.T, meta Metadata, wfs iceio.WriteFileIO, cat CatalogIO) *Table {
+	t.Helper()
+
+	return New(
+		Identifier{"db", "occ-cleanup-test"},
+		meta,
+		"mem://default/table-location/metadata/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return wfs, nil },
+		cat,
+	)
+}
+
+// newMemIOWithRetryMeta creates a test memIO and a matching table Metadata that
+// includes retry-config properties, so doCommit's retry loop allows retries.
+// The location matches createTestTransactionWithMemIO so they share the same
+// memIO for writing manifest files.
+func newMemIOWithRetryMeta(t *testing.T, spec iceberg.PartitionSpec) (*memIO, Metadata) {
+	t.Helper()
+	wfs := newMemIO(1<<20, nil)
+	schema := simpleSchema()
+	meta, err := NewMetadata(schema, &spec, UnsortedSortOrder, "mem://default/table-location",
+		iceberg.Properties{
+			CommitNumRetriesKey:          "3",
+			CommitMinRetryWaitMsKey:      "1",
+			CommitMaxRetryWaitMsKey:      "2",
+			CommitTotalRetryTimeoutMsKey: "60000",
+		})
+	require.NoError(t, err, "new metadata")
+
+	return wfs, meta
+}
+
+// TestDoCommit_OrphanCleanedOnSuccess verifies that manifest-list files
+// orphaned by OCC retries are removed after a successful commit. These files
+// are written during rebuild and must not leak on the happy path.
+func TestDoCommit_OrphanCleanedOnSuccess(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	// Build a transaction from the retry-enabled meta and commit via the producer.
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	// Pre-populate the original manifest list path in the IO so that the
+	// defer's wfs.Remove call has a concrete entry to delete.
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	// Catalog: fail once with ErrCommitFailed (triggers rebuild that orphans
+	// originalManifestList), then succeed.
+	cat := &sequentialCatalog{
+		metadata: meta,
+		errs:     []error{ErrCommitFailed},
+	}
+	tbl = newOCCTable(t, meta, wfs, cat)
+
+	_, err = tbl.doCommit(t.Context(), updates, reqs, withCommitBranch(MainBranch))
+	require.NoError(t, err, "doCommit must succeed on the second attempt")
+
+	_, stillExists := wfs.files[originalManifestList]

Review Comment:
   [Important] The orphan tests don't actually exercise a real write→delete cycle. `memIO.Create` returns a `limitedWriteCloser` whose contents are not persisted into `wfs.files`, so the rebuilt manifest list path is never present in `wfs.files` to begin with. The test pre-populates the *original* path with a placeholder (line 403) and asserts it was removed — but a regression where the cleanup loop deleted the wrong path (e.g., the *committed* manifest list instead of the orphan) would slip through, because the committed path was never persisted either. Two ways to close this: (a) make `memIO.Create` persist on `Close()` so the test exercises the real flow; or (b) add `assert.NotEqual(t, originalManifestList, cat.metadata.CurrentSnapshot().ManifestList)` plus a positive assertion on the live path's existence. Same applies to the other three `TestDoCommit_Orphan*` tests.



-- 
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

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_r3202392992


##########
table/commit_retry_test.go:
##########
@@ -266,3 +307,244 @@ func TestReadRetryConfig_ClampsNegativeProperties(t *testing.T) {
 	assert.Equal(t, uint(CommitMaxRetryWaitMsDefault), cfg.maxWaitMs)
 	assert.Equal(t, uint(CommitTotalRetryTimeoutMsDefault), cfg.totalTimeoutMs)
 }
+
+// ---------------------------------------------------------------------------
+// Fix 5 — mandatory WriteFileIO check at top of doCommit
+// ---------------------------------------------------------------------------
+
+// TestDoCommit_NonWriteFileIOReturnsError verifies that doCommit fails
+// immediately when the table's file system does not implement WriteFileIO.
+// A silent skip would reuse the stale manifest list — exactly the bug
+// this PR was designed to fix.
+func TestDoCommit_NonWriteFileIOReturnsError(t *testing.T) {
+	cat := &flakyCatalog{}
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/rotest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	cat.metadata = meta
+
+	tbl := New(
+		Identifier{"db", "ro-test"},
+		meta, "file:///tmp/rotest/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return readOnlyIO{}, nil },
+		cat,
+	)
+
+	_, doErr := tbl.doCommit(t.Context(), nil, nil)
+	require.Error(t, doErr, "doCommit must fail when FS does not implement WriteFileIO")
+	assert.Contains(t, doErr.Error(), "WriteFileIO",
+		"error message must mention WriteFileIO so callers understand the requirement")
+	assert.Equal(t, int32(0), cat.attempts.Load(),
+		"CommitTable must not be called when FS check fails")
+}
+
+// ---------------------------------------------------------------------------
+// Fix 6 — orphan cleanup via defer
+// ---------------------------------------------------------------------------
+
+// newOCCTable creates a table that uses the given wfs for its FS and the given
+// catalog for commits. meta should include retry-config properties so that
+// doCommit's retry loop allows at least one retry.
+func newOCCTable(t *testing.T, meta Metadata, wfs iceio.WriteFileIO, cat CatalogIO) *Table {
+	t.Helper()
+
+	return New(
+		Identifier{"db", "occ-cleanup-test"},
+		meta,
+		"mem://default/table-location/metadata/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return wfs, nil },
+		cat,
+	)
+}
+
+// newMemIOWithRetryMeta creates a test memIO and a matching table Metadata that
+// includes retry-config properties, so doCommit's retry loop allows retries.
+// The location matches createTestTransactionWithMemIO so they share the same
+// memIO for writing manifest files.
+func newMemIOWithRetryMeta(t *testing.T, spec iceberg.PartitionSpec) (*memIO, Metadata) {
+	t.Helper()
+	wfs := newMemIO(1<<20, nil)
+	schema := simpleSchema()
+	meta, err := NewMetadata(schema, &spec, UnsortedSortOrder, "mem://default/table-location",
+		iceberg.Properties{
+			CommitNumRetriesKey:          "3",
+			CommitMinRetryWaitMsKey:      "1",
+			CommitMaxRetryWaitMsKey:      "2",
+			CommitTotalRetryTimeoutMsKey: "60000",
+		})
+	require.NoError(t, err, "new metadata")
+
+	return wfs, meta
+}
+
+// TestDoCommit_OrphanCleanedOnSuccess verifies that manifest-list files
+// orphaned by OCC retries are removed after a successful commit. These files
+// are written during rebuild and must not leak on the happy path.
+func TestDoCommit_OrphanCleanedOnSuccess(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	// Build a transaction from the retry-enabled meta and commit via the producer.
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	// Pre-populate the original manifest list path in the IO so that the
+	// defer's wfs.Remove call has a concrete entry to delete.
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	// Catalog: fail once with ErrCommitFailed (triggers rebuild that orphans
+	// originalManifestList), then succeed.
+	cat := &sequentialCatalog{
+		metadata: meta,
+		errs:     []error{ErrCommitFailed},
+	}
+	tbl = newOCCTable(t, meta, wfs, cat)
+
+	_, err = tbl.doCommit(t.Context(), updates, reqs, withCommitBranch(MainBranch))
+	require.NoError(t, err, "doCommit must succeed on the second attempt")
+
+	_, stillExists := wfs.files[originalManifestList]

Review Comment:
   [Important] The orphan tests don't actually exercise a real write→delete cycle. `memIO.Create` returns a `limitedWriteCloser` whose contents are not persisted into `wfs.files`, so the rebuilt manifest list path is never present in `wfs.files` to begin with. The test pre-populates the *original* path with a placeholder (line 403) and asserts it was removed — but a regression where the cleanup loop deleted the wrong path (e.g., the *committed* manifest list instead of the orphan) would slip through, because the committed path was never persisted either. Two ways to close this: (a) make `memIO.Create` persist on `Close()` so the test exercises the real flow; or (b) add `assert.NotEqual(t, originalManifestList, cat.metadata.CurrentSnapshot().ManifestList)` plus a positive assertion on the live path's existence. Same applies to the other three `TestDoCommit_Orphan*` tests.



-- 
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

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

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

["mzzz-zzm (via GitHub)" <gi...@apache.org>]

mzzz-zzm commented on PR #982:
URL: https://github.com/apache/iceberg-go/pull/982#issuecomment-4398306629

   **Heads-up — proactive note**
   
   While addressing the six review comments I noticed a fragility in `addSnapshotUpdate.Apply` that I think is worth flagging. It isn't covered by your review and I couldn't find an existing issue for it, so surfacing it here.
   
   The rebuild closure (`rebuildManifestList`) and `ownManifests` are runtime-only fields on `*addSnapshotUpdate`. When `Apply` runs, it calls `builder.AddSnapshot(u.Snapshot)`, which internally constructs a **new** `*addSnapshotUpdate` via `NewAddSnapshotUpdate(snapshot)` and appends it to `builder.updates`. That new object has the closure fields zeroed. To make the fix work, `Apply` reaches back into `builder.updates[n-1]` and copies the closure onto that fresh object.
   
   This works today, but it depends on `MetadataBuilder.AddSnapshot` always appending exactly **one** `*addSnapshotUpdate` as the last element. If `AddSnapshot` is ever extended to append a helper update afterward (e.g. for v3 row-lineage tracking or schema validation), `builder.updates[n-1]` is no longer our snapshot update — the type assertion silently fails, the closure is dropped, and the entire OCC rebuild stops running with no error or log. Same "silent fix-disable" shape as the `WriteFileIO` cast you flagged in Comment 5.
   
   The cleaner shape would be to have `MetadataBuilder.AddSnapshot` take `*addSnapshotUpdate` directly and store it as-is, so the back-door goes away and `Apply` becomes a one-liner. I kept the current approach to minimize blast radius, but happy to make the change in this PR or a follow-up


-- 
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

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_r3202392983


##########
table/rebuild_manifest_test.go:
##########
@@ -0,0 +1,476 @@
+// 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.
+
+package table
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	iceio "github.com/apache/iceberg-go/io"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// newMetadataWithLastSeqNum builds a v2 Metadata whose last-sequence-number
+// equals lastSeqNum by grafting a synthetic snapshot at that sequence number.
+// Used by tests that need to control freshMeta.LastSequenceNumber().
+func newMetadataWithLastSeqNum(t *testing.T, lastSeqNum int64) Metadata {
+	t.Helper()
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/seqtest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	builder, err := MetadataBuilderFromBase(meta, "")
+	require.NoError(t, err)
+	snap := Snapshot{
+		SnapshotID:     1000,
+		SequenceNumber: lastSeqNum,
+		TimestampMs:    meta.LastUpdatedMillis() + 1,
+		Summary:        &Summary{Operation: OpAppend},
+	}
+	require.NoError(t, builder.AddSnapshot(&snap))
+	require.NoError(t, builder.SetSnapshotRef(MainBranch, 1000, BranchRef))
+	out, err := builder.Build()
+	require.NoError(t, err)
+
+	return out
+}
+
+// newV3MetadataWithNextRowID builds a v3 Metadata whose NextRowID() returns
+// nextRowID by adding a synthetic snapshot that consumes that many rows.
+// Used by tests that need to control freshMeta.NextRowID() for v3 row lineage.
+func newV3MetadataWithNextRowID(t *testing.T, nextRowID int64) Metadata {
+	t.Helper()
+	spec := iceberg.NewPartitionSpec()
+	txn, _ := createTestTransactionWithMemIO(t, spec)
+	txn.meta.formatVersion = 3
+
+	firstRowID := int64(0)
+	addedRows := nextRowID
+	snap := Snapshot{
+		SnapshotID:     1000,
+		SequenceNumber: 1,
+		TimestampMs:    txn.meta.base.LastUpdatedMillis() + 1,
+		Summary:        &Summary{Operation: OpAppend},
+		FirstRowID:     &firstRowID,
+		AddedRows:      &addedRows,
+	}
+	require.NoError(t, txn.meta.AddSnapshot(&snap))
+	require.NoError(t, txn.meta.SetSnapshotRef(MainBranch, 1000, BranchRef))
+
+	meta, err := txn.meta.Build()
+	require.NoError(t, err)
+	require.Equal(t, nextRowID, meta.NextRowID())
+
+	return meta
+}
+
+// ptr returns a pointer to v, used in test helper expressions.
+// NOTE: ptr is also declared in pos_delete_partitioned_fanout_writer_test.go;
+// both files share the same package so only one declaration is needed.
+// This comment documents the shared usage — do not re-add a declaration here.
+
+// rebuildUpdate constructs an addSnapshotUpdate whose rebuildManifestList
+// closure simply records the freshParent it received and returns a new
+// snapshot whose ManifestList is the given newManifestList value.
+func rebuildUpdate(snap *Snapshot, newManifestList string, gotParent **Snapshot) *addSnapshotUpdate {
+	return &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, freshParent *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			*gotParent = freshParent
+			rebuilt := *snap
+			rebuilt.ManifestList = newManifestList
+
+			return &rebuilt, nil
+		},
+	}
+}
+
+// TestRebuildSnapshotUpdates_CallsClosureWithFreshParent verifies that
+// rebuildSnapshotUpdates invokes the rebuild closure and passes the fresh
+// branch head as freshParent.
+func TestRebuildSnapshotUpdates_CallsClosureWithFreshParent(t *testing.T) {
+	const oldManifest = "s3://bucket/old-manifest-list.avro"
+	const newManifest = "s3://bucket/new-manifest-list.avro"
+
+	// Build a fresh metadata that has a different snapshot than the one
+	// embedded in the update, so the parent-hasn't-changed guard is
+	// bypassed and the closure must be called.
+	freshHead := int64(42)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	parentID := int64(7) // original snap's parent — does NOT match freshHead (42)
+	snap := &Snapshot{
+		SnapshotID:       99,
+		ParentSnapshotID: &parentID,
+		ManifestList:     oldManifest,
+		Summary:          &Summary{Operation: OpAppend},
+	}
+
+	var receivedParent *Snapshot
+	upd := rebuildUpdate(snap, newManifest, &receivedParent)
+
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{upd},
+		freshMeta,
+		MainBranch,
+		iceio.LocalFS{},
+		1,
+	)
+	require.NoError(t, err)
+	require.Len(t, rebuilt, 1)
+	require.Len(t, orphaned, 1, "old manifest list must be recorded as orphaned")
+
+	// The closure should have been called with the branch's current head.
+	require.NotNil(t, receivedParent)
+	assert.Equal(t, freshHead, receivedParent.SnapshotID)
+
+	// The rebuilt update must carry the new manifest list.
+	addUpd, ok := rebuilt[0].(*addSnapshotUpdate)
+	require.True(t, ok)
+	assert.Equal(t, newManifest, addUpd.Snapshot.ManifestList)
+
+	// The superseded manifest list becomes an orphan.
+	assert.Equal(t, oldManifest, orphaned[0])
+}
+
+// TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged verifies that
+// rebuildSnapshotUpdates skips the rebuild when the update's snapshot
+// already has the fresh branch head as its parent (no-op retry).
+func TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged(t *testing.T) {
+	const manifest = "s3://bucket/manifest-list.avro"
+
+	freshHead := int64(42)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	// Parent already equals the fresh head — rebuild must be skipped.
+	snap := &Snapshot{
+		SnapshotID:       99,
+		ParentSnapshotID: &freshHead, // same as fresh head
+		ManifestList:     manifest,
+		Summary:          &Summary{Operation: OpAppend},
+	}
+
+	called := false
+	upd := &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, _ *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			called = true
+
+			return snap, nil
+		},
+	}
+
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{upd},
+		freshMeta,
+		MainBranch,
+		iceio.LocalFS{},
+		1,
+	)
+	require.NoError(t, err)
+	assert.False(t, called, "rebuild closure must not be called when parent is already up-to-date")
+	assert.Empty(t, orphaned, "no orphans when rebuild is skipped")
+	assert.Same(t, upd, rebuilt[0].(*addSnapshotUpdate), "original update must pass through unchanged")
+}
+
+// TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates verifies that
+// updates without a rebuildManifestList closure are returned unmodified.
+func TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates(t *testing.T) {
+	plainUpd := NewAddSnapshotUpdate(&Snapshot{
+		SnapshotID:   1,
+		ManifestList: "s3://bucket/no-rebuild.avro",
+		Summary:      &Summary{Operation: OpAppend},
+	})
+
+	// freshMeta may be nil — the plain update must not be touched.
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{plainUpd},
+		nil,
+		MainBranch,
+		iceio.LocalFS{},
+		0,
+	)
+	require.NoError(t, err)
+	assert.Empty(t, orphaned)
+	assert.Same(t, plainUpd, rebuilt[0].(*addSnapshotUpdate))
+}
+
+// TestRebuildSnapshotUpdates_PropagatesClosureError verifies that an error
+// returned by the rebuild closure surfaces as the function's return error.
+func TestRebuildSnapshotUpdates_PropagatesClosureError(t *testing.T) {
+	freshHead := int64(5)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	parent := int64(1)
+	snap := &Snapshot{
+		SnapshotID:       10,
+		ParentSnapshotID: &parent,
+		ManifestList:     "s3://bucket/old.avro",
+		Summary:          &Summary{Operation: OpAppend},
+	}
+	wantErr := errors.New("simulated S3 write failure")
+	upd := &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, _ *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			return nil, wantErr
+		},
+	}
+
+	_, _, err := rebuildSnapshotUpdates(t.Context(), []Update{upd}, freshMeta, MainBranch, iceio.LocalFS{}, 1)
+	assert.ErrorIs(t, err, wantErr)
+}
+
+// ---------------------------------------------------------------------------
+// Fix 3 — newSeq derived from freshMeta.LastSequenceNumber()
+// ---------------------------------------------------------------------------
+
+// TestRebuildFn_SeqNumDerivedFromFreshMeta verifies that the rebuilt snapshot's
+// SequenceNumber equals freshMeta.LastSequenceNumber()+1, NOT
+// freshParent.SequenceNumber+1. A concurrent writer on a different branch can
+// advance the table-wide last-sequence-number without advancing this branch's
+// parent, so using freshParent.SequenceNumber+1 would violate the spec invariant.
+func TestRebuildFn_SeqNumDerivedFromFreshMeta(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	txn, wfs := createTestTransactionWithMemIO(t, spec)
+
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, _, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	require.NotNil(t, addSnap.rebuildManifestList, "rebuildManifestList closure must be set")
+
+	// Simulate a concurrent writer on another branch that bumped the global
+	// last-sequence-number to 99 without advancing this branch's parent.
+	// old code: newSeq = capturedSnapshot.SequenceNumber (stale, ≤ 99 — spec violation)
+	// new code: newSeq = freshMeta.LastSequenceNumber() + 1 = 100
+	freshMeta := newMetadataWithLastSeqNum(t, 99)
+	require.Equal(t, int64(99), freshMeta.LastSequenceNumber())
+
+	rebuilt, err := addSnap.rebuildManifestList(context.Background(), freshMeta, nil, wfs, 1)
+	require.NoError(t, err)
+	require.Equal(t, int64(100), rebuilt.SequenceNumber,
+		"SequenceNumber must be freshMeta.LastSequenceNumber()+1; using freshParent.SequenceNumber+1 violates the spec when another branch advanced the global counter")
+}
+
+// TestRebuildFn_SeqNumV1TableIsZero verifies that v1 tables keep SequenceNumber == 0
+// regardless of what freshMeta reports (v1 does not use sequence numbers).
+func TestRebuildFn_SeqNumV1TableIsZero(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	txn, wfs := createTestTransactionWithMemIO(t, spec)
+	txn.meta.formatVersion = 1 // override to v1
+
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, _, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	require.NotNil(t, addSnap.rebuildManifestList)
+
+	freshMeta := newMetadataWithLastSeqNum(t, 99)
+	rebuilt, err := addSnap.rebuildManifestList(context.Background(), freshMeta, nil, wfs, 1)
+	require.NoError(t, err)
+	require.Equal(t, int64(0), rebuilt.SequenceNumber, "v1 tables must always have SequenceNumber == 0")
+}
+
+// ---------------------------------------------------------------------------
+// Fix 1 — firstRowID derived from freshMeta.NextRowID()
+// ---------------------------------------------------------------------------
+
+// TestRebuildFn_V3FirstRowIDDerivedFromFreshMeta verifies that on a v3 table
+// the rebuilt snapshot's FirstRowID equals freshMeta.NextRowID() rather than
+// the hardcoded 0. If two writers race and the peer commits first, the
+// catalog's nextRowID has already advanced; using 0 would produce a
+// first-row-id that disagrees with the catalog's view and fails row-lineage
+// validation.
+func TestRebuildFn_V3FirstRowIDDerivedFromFreshMeta(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	txn, wfs := createTestTransactionWithMemIO(t, spec)
+	txn.meta.formatVersion = 3
+
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, _, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	require.NotNil(t, addSnap.rebuildManifestList)
+
+	// Simulate a concurrent writer that added 50 rows, advancing nextRowID to 50.
+	// old code: firstRowID = 0 (hardcoded)
+	// new code: firstRowID = freshMeta.NextRowID() = 50
+	freshMeta := newV3MetadataWithNextRowID(t, 50)
+	require.Equal(t, int64(50), freshMeta.NextRowID(), "freshMeta.NextRowID() must be 50")
+
+	rebuilt, err := addSnap.rebuildManifestList(context.Background(), freshMeta, nil, wfs, 1)
+	require.NoError(t, err)
+	require.NotNil(t, rebuilt.FirstRowID, "v3 rebuilt snapshot must have FirstRowID")
+	require.Equal(t, int64(50), *rebuilt.FirstRowID,
+		"FirstRowID must equal freshMeta.NextRowID(); hardcoded 0 would conflict with catalog's advanced nextRowID")
+
+	// AddedRows must be recomputed from writer.NextRowID() - firstRowID for
+	// this attempt (NOT carried over from capturedSnapshot at attempt 0).
+	// The single test data file has record count 1, so the producer's own
+	// contribution this attempt is exactly 1 row.
+	require.NotNil(t, rebuilt.AddedRows, "v3 rebuilt snapshot must have AddedRows")
+	require.Equal(t, int64(1), *rebuilt.AddedRows,
+		"AddedRows must equal writer.NextRowID() - firstRowID (1 row from this producer's data file)")

Review Comment:
   [Important] AddedRows assertion is regression-insensitive. The producer's data file has record-count 1, so `capturedSnapshot.AddedRows == 1` from attempt 0 — and `*rebuilt.AddedRows == 1` is the same value. If the rebuild closure were reverted to propagate `capturedSnapshot.AddedRows` (the pre-fix path) instead of recomputing from `writer.NextRowID() - firstRowID`, this assertion still passes. The fixture symmetry shadows the very regression the comment names. To pin the recompute, set up a state where the captured value and the recomputed value disagree — e.g. multi-row data file with `freshMeta.NextRowID() = 50` and capture a producer whose attempt-0 AddedRows was different, or use two data files (captured AddedRows=2) against a freshMeta that yields a different count. Same fixture issue applies to `TestRebuildFn_V3FirstRowIDZeroWhenNilNextRowID`.



-- 
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

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_r3202392990


##########
table/rebuild_manifest_test.go:
##########
@@ -0,0 +1,476 @@
+// 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.
+
+package table
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	iceio "github.com/apache/iceberg-go/io"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// newMetadataWithLastSeqNum builds a v2 Metadata whose last-sequence-number
+// equals lastSeqNum by grafting a synthetic snapshot at that sequence number.
+// Used by tests that need to control freshMeta.LastSequenceNumber().
+func newMetadataWithLastSeqNum(t *testing.T, lastSeqNum int64) Metadata {
+	t.Helper()
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/seqtest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	builder, err := MetadataBuilderFromBase(meta, "")
+	require.NoError(t, err)
+	snap := Snapshot{
+		SnapshotID:     1000,
+		SequenceNumber: lastSeqNum,
+		TimestampMs:    meta.LastUpdatedMillis() + 1,
+		Summary:        &Summary{Operation: OpAppend},
+	}
+	require.NoError(t, builder.AddSnapshot(&snap))
+	require.NoError(t, builder.SetSnapshotRef(MainBranch, 1000, BranchRef))
+	out, err := builder.Build()
+	require.NoError(t, err)
+
+	return out
+}
+
+// newV3MetadataWithNextRowID builds a v3 Metadata whose NextRowID() returns
+// nextRowID by adding a synthetic snapshot that consumes that many rows.
+// Used by tests that need to control freshMeta.NextRowID() for v3 row lineage.
+func newV3MetadataWithNextRowID(t *testing.T, nextRowID int64) Metadata {
+	t.Helper()
+	spec := iceberg.NewPartitionSpec()
+	txn, _ := createTestTransactionWithMemIO(t, spec)
+	txn.meta.formatVersion = 3
+
+	firstRowID := int64(0)
+	addedRows := nextRowID
+	snap := Snapshot{
+		SnapshotID:     1000,
+		SequenceNumber: 1,
+		TimestampMs:    txn.meta.base.LastUpdatedMillis() + 1,
+		Summary:        &Summary{Operation: OpAppend},
+		FirstRowID:     &firstRowID,
+		AddedRows:      &addedRows,
+	}
+	require.NoError(t, txn.meta.AddSnapshot(&snap))
+	require.NoError(t, txn.meta.SetSnapshotRef(MainBranch, 1000, BranchRef))
+
+	meta, err := txn.meta.Build()
+	require.NoError(t, err)
+	require.Equal(t, nextRowID, meta.NextRowID())
+
+	return meta
+}
+
+// ptr returns a pointer to v, used in test helper expressions.
+// NOTE: ptr is also declared in pos_delete_partitioned_fanout_writer_test.go;
+// both files share the same package so only one declaration is needed.
+// This comment documents the shared usage — do not re-add a declaration here.
+
+// rebuildUpdate constructs an addSnapshotUpdate whose rebuildManifestList
+// closure simply records the freshParent it received and returns a new
+// snapshot whose ManifestList is the given newManifestList value.
+func rebuildUpdate(snap *Snapshot, newManifestList string, gotParent **Snapshot) *addSnapshotUpdate {
+	return &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, freshParent *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			*gotParent = freshParent
+			rebuilt := *snap
+			rebuilt.ManifestList = newManifestList
+
+			return &rebuilt, nil
+		},
+	}
+}
+
+// TestRebuildSnapshotUpdates_CallsClosureWithFreshParent verifies that
+// rebuildSnapshotUpdates invokes the rebuild closure and passes the fresh
+// branch head as freshParent.
+func TestRebuildSnapshotUpdates_CallsClosureWithFreshParent(t *testing.T) {
+	const oldManifest = "s3://bucket/old-manifest-list.avro"
+	const newManifest = "s3://bucket/new-manifest-list.avro"
+
+	// Build a fresh metadata that has a different snapshot than the one
+	// embedded in the update, so the parent-hasn't-changed guard is
+	// bypassed and the closure must be called.
+	freshHead := int64(42)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	parentID := int64(7) // original snap's parent — does NOT match freshHead (42)
+	snap := &Snapshot{
+		SnapshotID:       99,
+		ParentSnapshotID: &parentID,
+		ManifestList:     oldManifest,
+		Summary:          &Summary{Operation: OpAppend},
+	}
+
+	var receivedParent *Snapshot
+	upd := rebuildUpdate(snap, newManifest, &receivedParent)
+
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{upd},
+		freshMeta,
+		MainBranch,
+		iceio.LocalFS{},
+		1,
+	)
+	require.NoError(t, err)
+	require.Len(t, rebuilt, 1)
+	require.Len(t, orphaned, 1, "old manifest list must be recorded as orphaned")
+
+	// The closure should have been called with the branch's current head.
+	require.NotNil(t, receivedParent)
+	assert.Equal(t, freshHead, receivedParent.SnapshotID)
+
+	// The rebuilt update must carry the new manifest list.
+	addUpd, ok := rebuilt[0].(*addSnapshotUpdate)
+	require.True(t, ok)
+	assert.Equal(t, newManifest, addUpd.Snapshot.ManifestList)
+
+	// The superseded manifest list becomes an orphan.
+	assert.Equal(t, oldManifest, orphaned[0])
+}
+
+// TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged verifies that
+// rebuildSnapshotUpdates skips the rebuild when the update's snapshot
+// already has the fresh branch head as its parent (no-op retry).
+func TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged(t *testing.T) {
+	const manifest = "s3://bucket/manifest-list.avro"
+
+	freshHead := int64(42)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	// Parent already equals the fresh head — rebuild must be skipped.
+	snap := &Snapshot{
+		SnapshotID:       99,
+		ParentSnapshotID: &freshHead, // same as fresh head
+		ManifestList:     manifest,
+		Summary:          &Summary{Operation: OpAppend},
+	}
+
+	called := false
+	upd := &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, _ *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			called = true
+
+			return snap, nil
+		},
+	}
+
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{upd},
+		freshMeta,
+		MainBranch,
+		iceio.LocalFS{},
+		1,
+	)
+	require.NoError(t, err)
+	assert.False(t, called, "rebuild closure must not be called when parent is already up-to-date")
+	assert.Empty(t, orphaned, "no orphans when rebuild is skipped")
+	assert.Same(t, upd, rebuilt[0].(*addSnapshotUpdate), "original update must pass through unchanged")
+}
+
+// TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates verifies that
+// updates without a rebuildManifestList closure are returned unmodified.
+func TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates(t *testing.T) {
+	plainUpd := NewAddSnapshotUpdate(&Snapshot{
+		SnapshotID:   1,
+		ManifestList: "s3://bucket/no-rebuild.avro",
+		Summary:      &Summary{Operation: OpAppend},
+	})
+
+	// freshMeta may be nil — the plain update must not be touched.
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{plainUpd},
+		nil,
+		MainBranch,
+		iceio.LocalFS{},
+		0,
+	)
+	require.NoError(t, err)
+	assert.Empty(t, orphaned)
+	assert.Same(t, plainUpd, rebuilt[0].(*addSnapshotUpdate))
+}
+
+// TestRebuildSnapshotUpdates_PropagatesClosureError verifies that an error
+// returned by the rebuild closure surfaces as the function's return error.
+func TestRebuildSnapshotUpdates_PropagatesClosureError(t *testing.T) {
+	freshHead := int64(5)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	parent := int64(1)
+	snap := &Snapshot{
+		SnapshotID:       10,
+		ParentSnapshotID: &parent,
+		ManifestList:     "s3://bucket/old.avro",
+		Summary:          &Summary{Operation: OpAppend},
+	}
+	wantErr := errors.New("simulated S3 write failure")
+	upd := &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, _ *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			return nil, wantErr
+		},
+	}
+
+	_, _, err := rebuildSnapshotUpdates(t.Context(), []Update{upd}, freshMeta, MainBranch, iceio.LocalFS{}, 1)
+	assert.ErrorIs(t, err, wantErr)
+}
+
+// ---------------------------------------------------------------------------
+// Fix 3 — newSeq derived from freshMeta.LastSequenceNumber()
+// ---------------------------------------------------------------------------
+
+// TestRebuildFn_SeqNumDerivedFromFreshMeta verifies that the rebuilt snapshot's
+// SequenceNumber equals freshMeta.LastSequenceNumber()+1, NOT
+// freshParent.SequenceNumber+1. A concurrent writer on a different branch can
+// advance the table-wide last-sequence-number without advancing this branch's
+// parent, so using freshParent.SequenceNumber+1 would violate the spec invariant.
+func TestRebuildFn_SeqNumDerivedFromFreshMeta(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	txn, wfs := createTestTransactionWithMemIO(t, spec)
+
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, _, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	require.NotNil(t, addSnap.rebuildManifestList, "rebuildManifestList closure must be set")
+
+	// Simulate a concurrent writer on another branch that bumped the global
+	// last-sequence-number to 99 without advancing this branch's parent.
+	// old code: newSeq = capturedSnapshot.SequenceNumber (stale, ≤ 99 — spec violation)
+	// new code: newSeq = freshMeta.LastSequenceNumber() + 1 = 100
+	freshMeta := newMetadataWithLastSeqNum(t, 99)
+	require.Equal(t, int64(99), freshMeta.LastSequenceNumber())
+
+	rebuilt, err := addSnap.rebuildManifestList(context.Background(), freshMeta, nil, wfs, 1)

Review Comment:
   [Important] The headline scenario the fix targets — peer-branch advancement of `last-sequence-number` while this branch's parent stays put — is not actually exercised. The test passes `freshParent=nil`, so the buggy `freshParent != nil` branch (`newSeq = freshParent.SequenceNumber + 1`) never runs. An intermediate regression that reverts only the non-nil branch back to `freshParent.SequenceNumber + 1` would not be caught here, and `TestManifestListInheritedAfterConflict` doesn't catch it either because in that fixture both arithmetic paths happen to yield the same value. Strengthening: graft a peer-branch snapshot onto `freshMeta` with a higher `SequenceNumber` while keeping main's `freshParent.SequenceNumber` lower, then call the closure with both `freshMeta` and that lower `freshParent`, asserting `rebuilt.SequenceNumber == freshMeta.LastSequenceNumber()+1` — not `freshParent.SequenceNumber+1`. Alternatively, add `require.Equal(t, freshMeta.LastSequenceNumber()+1, rebuilt.
 SequenceNumber)` to the existing `TestRebuildFn_SummaryRebasedAgainstFreshParent` (which already passes both `freshParent` and `freshMeta`) — one-line strengthening that closes the gap.



-- 
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

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

["mzzz-zzm (via GitHub)" <gi...@apache.org>]

mzzz-zzm commented on code in PR #982:
URL: https://github.com/apache/iceberg-go/pull/982#discussion_r3202139760


##########
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:
   Inside `rebuildFn` (`table/snapshot_producers.go`), `firstRowID` is now derived from `freshMeta.NextRowID()` for v3 tables (only `freshMeta` is the source of truth — `capturedSnapshot` is never consulted for this value):
   
   ```go
   var firstRowID int64
   if formatVersion >= 3 {
       // Derive firstRowID from the fresh metadata so the manifest-list
       // first-row-id header matches the catalog's current nextRowID.
       firstRowID = freshMeta.NextRowID()
       ...
   }
   ```
   
   After `writer.AddManifests(combined)`, `rebuilt.FirstRowID` and `rebuilt.AddedRows` are updated to reflect the rows actually written this attempt:
   
   ```go
   rebuilt.FirstRowID = &firstRowID
   rebuilt.AddedRows = &addedRows   // writer.NextRowID() - firstRowID
   ```
   
   This mirrors what `commit()` does at the initial write site, keeping the snapshot record and the manifest-list header consistent.
   
   Tests added: `TestRebuildFn_V3FirstRowIDDerivedFromFreshMeta` (concurrent write advances `NextRowID` to 50; asserts rebuilt `FirstRowID == 50` AND `AddedRows == 1` reflecting only this attempt's contribution), `TestRebuildFn_V3FirstRowIDZeroWhenNilNextRowID` (freshMeta with `NextRowID() == 0` — the zero-equivalent for the `int64` API — asserts `FirstRowID == 0`, `AddedRows == 1`, no panic).



-- 
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

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

["mzzz-zzm (via GitHub)" <gi...@apache.org>]

mzzz-zzm commented on code in PR #982:
URL: https://github.com/apache/iceberg-go/pull/982#discussion_r3203217216


##########
table/rebuild_manifest_test.go:
##########
@@ -0,0 +1,476 @@
+// 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.
+
+package table
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	iceio "github.com/apache/iceberg-go/io"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// newMetadataWithLastSeqNum builds a v2 Metadata whose last-sequence-number
+// equals lastSeqNum by grafting a synthetic snapshot at that sequence number.
+// Used by tests that need to control freshMeta.LastSequenceNumber().
+func newMetadataWithLastSeqNum(t *testing.T, lastSeqNum int64) Metadata {
+	t.Helper()
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/seqtest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	builder, err := MetadataBuilderFromBase(meta, "")
+	require.NoError(t, err)
+	snap := Snapshot{
+		SnapshotID:     1000,
+		SequenceNumber: lastSeqNum,
+		TimestampMs:    meta.LastUpdatedMillis() + 1,
+		Summary:        &Summary{Operation: OpAppend},
+	}
+	require.NoError(t, builder.AddSnapshot(&snap))
+	require.NoError(t, builder.SetSnapshotRef(MainBranch, 1000, BranchRef))
+	out, err := builder.Build()
+	require.NoError(t, err)
+
+	return out
+}
+
+// newV3MetadataWithNextRowID builds a v3 Metadata whose NextRowID() returns
+// nextRowID by adding a synthetic snapshot that consumes that many rows.
+// Used by tests that need to control freshMeta.NextRowID() for v3 row lineage.
+func newV3MetadataWithNextRowID(t *testing.T, nextRowID int64) Metadata {
+	t.Helper()
+	spec := iceberg.NewPartitionSpec()
+	txn, _ := createTestTransactionWithMemIO(t, spec)
+	txn.meta.formatVersion = 3
+
+	firstRowID := int64(0)
+	addedRows := nextRowID
+	snap := Snapshot{
+		SnapshotID:     1000,
+		SequenceNumber: 1,
+		TimestampMs:    txn.meta.base.LastUpdatedMillis() + 1,
+		Summary:        &Summary{Operation: OpAppend},
+		FirstRowID:     &firstRowID,
+		AddedRows:      &addedRows,
+	}
+	require.NoError(t, txn.meta.AddSnapshot(&snap))
+	require.NoError(t, txn.meta.SetSnapshotRef(MainBranch, 1000, BranchRef))
+
+	meta, err := txn.meta.Build()
+	require.NoError(t, err)
+	require.Equal(t, nextRowID, meta.NextRowID())
+
+	return meta
+}
+
+// ptr returns a pointer to v, used in test helper expressions.
+// NOTE: ptr is also declared in pos_delete_partitioned_fanout_writer_test.go;
+// both files share the same package so only one declaration is needed.
+// This comment documents the shared usage — do not re-add a declaration here.
+
+// rebuildUpdate constructs an addSnapshotUpdate whose rebuildManifestList
+// closure simply records the freshParent it received and returns a new
+// snapshot whose ManifestList is the given newManifestList value.
+func rebuildUpdate(snap *Snapshot, newManifestList string, gotParent **Snapshot) *addSnapshotUpdate {
+	return &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, freshParent *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			*gotParent = freshParent
+			rebuilt := *snap
+			rebuilt.ManifestList = newManifestList
+
+			return &rebuilt, nil
+		},
+	}
+}
+
+// TestRebuildSnapshotUpdates_CallsClosureWithFreshParent verifies that
+// rebuildSnapshotUpdates invokes the rebuild closure and passes the fresh
+// branch head as freshParent.
+func TestRebuildSnapshotUpdates_CallsClosureWithFreshParent(t *testing.T) {
+	const oldManifest = "s3://bucket/old-manifest-list.avro"
+	const newManifest = "s3://bucket/new-manifest-list.avro"
+
+	// Build a fresh metadata that has a different snapshot than the one
+	// embedded in the update, so the parent-hasn't-changed guard is
+	// bypassed and the closure must be called.
+	freshHead := int64(42)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	parentID := int64(7) // original snap's parent — does NOT match freshHead (42)
+	snap := &Snapshot{
+		SnapshotID:       99,
+		ParentSnapshotID: &parentID,
+		ManifestList:     oldManifest,
+		Summary:          &Summary{Operation: OpAppend},
+	}
+
+	var receivedParent *Snapshot
+	upd := rebuildUpdate(snap, newManifest, &receivedParent)
+
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{upd},
+		freshMeta,
+		MainBranch,
+		iceio.LocalFS{},
+		1,
+	)
+	require.NoError(t, err)
+	require.Len(t, rebuilt, 1)
+	require.Len(t, orphaned, 1, "old manifest list must be recorded as orphaned")
+
+	// The closure should have been called with the branch's current head.
+	require.NotNil(t, receivedParent)
+	assert.Equal(t, freshHead, receivedParent.SnapshotID)
+
+	// The rebuilt update must carry the new manifest list.
+	addUpd, ok := rebuilt[0].(*addSnapshotUpdate)
+	require.True(t, ok)
+	assert.Equal(t, newManifest, addUpd.Snapshot.ManifestList)
+
+	// The superseded manifest list becomes an orphan.
+	assert.Equal(t, oldManifest, orphaned[0])
+}
+
+// TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged verifies that
+// rebuildSnapshotUpdates skips the rebuild when the update's snapshot
+// already has the fresh branch head as its parent (no-op retry).
+func TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged(t *testing.T) {
+	const manifest = "s3://bucket/manifest-list.avro"
+
+	freshHead := int64(42)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	// Parent already equals the fresh head — rebuild must be skipped.
+	snap := &Snapshot{
+		SnapshotID:       99,
+		ParentSnapshotID: &freshHead, // same as fresh head
+		ManifestList:     manifest,
+		Summary:          &Summary{Operation: OpAppend},
+	}
+
+	called := false
+	upd := &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, _ *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			called = true
+
+			return snap, nil
+		},
+	}
+
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{upd},
+		freshMeta,
+		MainBranch,
+		iceio.LocalFS{},
+		1,
+	)
+	require.NoError(t, err)
+	assert.False(t, called, "rebuild closure must not be called when parent is already up-to-date")
+	assert.Empty(t, orphaned, "no orphans when rebuild is skipped")
+	assert.Same(t, upd, rebuilt[0].(*addSnapshotUpdate), "original update must pass through unchanged")
+}
+
+// TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates verifies that
+// updates without a rebuildManifestList closure are returned unmodified.
+func TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates(t *testing.T) {
+	plainUpd := NewAddSnapshotUpdate(&Snapshot{
+		SnapshotID:   1,
+		ManifestList: "s3://bucket/no-rebuild.avro",
+		Summary:      &Summary{Operation: OpAppend},
+	})
+
+	// freshMeta may be nil — the plain update must not be touched.
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{plainUpd},
+		nil,
+		MainBranch,
+		iceio.LocalFS{},
+		0,
+	)
+	require.NoError(t, err)
+	assert.Empty(t, orphaned)
+	assert.Same(t, plainUpd, rebuilt[0].(*addSnapshotUpdate))
+}
+
+// TestRebuildSnapshotUpdates_PropagatesClosureError verifies that an error
+// returned by the rebuild closure surfaces as the function's return error.
+func TestRebuildSnapshotUpdates_PropagatesClosureError(t *testing.T) {
+	freshHead := int64(5)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	parent := int64(1)
+	snap := &Snapshot{
+		SnapshotID:       10,
+		ParentSnapshotID: &parent,
+		ManifestList:     "s3://bucket/old.avro",
+		Summary:          &Summary{Operation: OpAppend},
+	}
+	wantErr := errors.New("simulated S3 write failure")
+	upd := &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, _ *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			return nil, wantErr
+		},
+	}
+
+	_, _, err := rebuildSnapshotUpdates(t.Context(), []Update{upd}, freshMeta, MainBranch, iceio.LocalFS{}, 1)
+	assert.ErrorIs(t, err, wantErr)
+}
+
+// ---------------------------------------------------------------------------
+// Fix 3 — newSeq derived from freshMeta.LastSequenceNumber()
+// ---------------------------------------------------------------------------
+
+// TestRebuildFn_SeqNumDerivedFromFreshMeta verifies that the rebuilt snapshot's
+// SequenceNumber equals freshMeta.LastSequenceNumber()+1, NOT
+// freshParent.SequenceNumber+1. A concurrent writer on a different branch can
+// advance the table-wide last-sequence-number without advancing this branch's
+// parent, so using freshParent.SequenceNumber+1 would violate the spec invariant.
+func TestRebuildFn_SeqNumDerivedFromFreshMeta(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	txn, wfs := createTestTransactionWithMemIO(t, spec)
+
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, _, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	require.NotNil(t, addSnap.rebuildManifestList, "rebuildManifestList closure must be set")
+
+	// Simulate a concurrent writer on another branch that bumped the global
+	// last-sequence-number to 99 without advancing this branch's parent.
+	// old code: newSeq = capturedSnapshot.SequenceNumber (stale, ≤ 99 — spec violation)
+	// new code: newSeq = freshMeta.LastSequenceNumber() + 1 = 100
+	freshMeta := newMetadataWithLastSeqNum(t, 99)
+	require.Equal(t, int64(99), freshMeta.LastSequenceNumber())
+
+	rebuilt, err := addSnap.rebuildManifestList(context.Background(), freshMeta, nil, wfs, 1)
+	require.NoError(t, err)
+	require.Equal(t, int64(100), rebuilt.SequenceNumber,
+		"SequenceNumber must be freshMeta.LastSequenceNumber()+1; using freshParent.SequenceNumber+1 violates the spec when another branch advanced the global counter")
+}
+
+// TestRebuildFn_SeqNumV1TableIsZero verifies that v1 tables keep SequenceNumber == 0
+// regardless of what freshMeta reports (v1 does not use sequence numbers).
+func TestRebuildFn_SeqNumV1TableIsZero(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	txn, wfs := createTestTransactionWithMemIO(t, spec)
+	txn.meta.formatVersion = 1 // override to v1
+
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, _, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	require.NotNil(t, addSnap.rebuildManifestList)
+
+	freshMeta := newMetadataWithLastSeqNum(t, 99)
+	rebuilt, err := addSnap.rebuildManifestList(context.Background(), freshMeta, nil, wfs, 1)
+	require.NoError(t, err)
+	require.Equal(t, int64(0), rebuilt.SequenceNumber, "v1 tables must always have SequenceNumber == 0")
+}
+
+// ---------------------------------------------------------------------------
+// Fix 1 — firstRowID derived from freshMeta.NextRowID()
+// ---------------------------------------------------------------------------
+
+// TestRebuildFn_V3FirstRowIDDerivedFromFreshMeta verifies that on a v3 table
+// the rebuilt snapshot's FirstRowID equals freshMeta.NextRowID() rather than
+// the hardcoded 0. If two writers race and the peer commits first, the
+// catalog's nextRowID has already advanced; using 0 would produce a
+// first-row-id that disagrees with the catalog's view and fails row-lineage
+// validation.
+func TestRebuildFn_V3FirstRowIDDerivedFromFreshMeta(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	txn, wfs := createTestTransactionWithMemIO(t, spec)
+	txn.meta.formatVersion = 3
+
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, _, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	require.NotNil(t, addSnap.rebuildManifestList)
+
+	// Simulate a concurrent writer that added 50 rows, advancing nextRowID to 50.
+	// old code: firstRowID = 0 (hardcoded)
+	// new code: firstRowID = freshMeta.NextRowID() = 50
+	freshMeta := newV3MetadataWithNextRowID(t, 50)
+	require.Equal(t, int64(50), freshMeta.NextRowID(), "freshMeta.NextRowID() must be 50")
+
+	rebuilt, err := addSnap.rebuildManifestList(context.Background(), freshMeta, nil, wfs, 1)
+	require.NoError(t, err)
+	require.NotNil(t, rebuilt.FirstRowID, "v3 rebuilt snapshot must have FirstRowID")
+	require.Equal(t, int64(50), *rebuilt.FirstRowID,
+		"FirstRowID must equal freshMeta.NextRowID(); hardcoded 0 would conflict with catalog's advanced nextRowID")
+
+	// AddedRows must be recomputed from writer.NextRowID() - firstRowID for
+	// this attempt (NOT carried over from capturedSnapshot at attempt 0).
+	// The single test data file has record count 1, so the producer's own
+	// contribution this attempt is exactly 1 row.
+	require.NotNil(t, rebuilt.AddedRows, "v3 rebuilt snapshot must have AddedRows")
+	require.Equal(t, int64(1), *rebuilt.AddedRows,

Review Comment:
   `TestRebuildFn_V3FirstRowIDDerivedFromFreshMeta` and `TestRebuildFn_V3FirstRowIDZeroWhenNilNextRowID` now mutate `capturedSnapshot.AddedRows` to a sentinel that **cannot** equal a correct recomputation:
   
   ```go
   require.NotNil(t, addSnap.Snapshot.AddedRows, "captured snapshot must have AddedRows on v3")
   sentinel := int64(999_999) // 888_888 in the NextRowID==0 variant
   *addSnap.Snapshot.AddedRows = sentinel
   
   // ... run rebuild ...
   
   require.NotEqual(t, sentinel, *rebuilt.AddedRows,
       "AddedRows must be recomputed; carrying the captured value (mutated to sentinel) would be a regression")
   require.Equal(t, int64(1), *rebuilt.AddedRows,
       "AddedRows must equal writer.NextRowID()-firstRowID (this producer's 1 data row)")
   ```
   
   A regression that reverts the closure to propagate `capturedSnapshot.AddedRows` instead of recomputing from `writer.NextRowID() - firstRowID` would fail `NotEqual(sentinel)`. Verified by reverting the recompute path locally — both tests fail as expected.



##########
table/rebuild_manifest_test.go:
##########
@@ -0,0 +1,476 @@
+// 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.
+
+package table
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	iceio "github.com/apache/iceberg-go/io"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// newMetadataWithLastSeqNum builds a v2 Metadata whose last-sequence-number
+// equals lastSeqNum by grafting a synthetic snapshot at that sequence number.
+// Used by tests that need to control freshMeta.LastSequenceNumber().
+func newMetadataWithLastSeqNum(t *testing.T, lastSeqNum int64) Metadata {
+	t.Helper()
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/seqtest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	builder, err := MetadataBuilderFromBase(meta, "")
+	require.NoError(t, err)
+	snap := Snapshot{
+		SnapshotID:     1000,
+		SequenceNumber: lastSeqNum,
+		TimestampMs:    meta.LastUpdatedMillis() + 1,
+		Summary:        &Summary{Operation: OpAppend},
+	}
+	require.NoError(t, builder.AddSnapshot(&snap))
+	require.NoError(t, builder.SetSnapshotRef(MainBranch, 1000, BranchRef))
+	out, err := builder.Build()
+	require.NoError(t, err)
+
+	return out
+}
+
+// newV3MetadataWithNextRowID builds a v3 Metadata whose NextRowID() returns
+// nextRowID by adding a synthetic snapshot that consumes that many rows.
+// Used by tests that need to control freshMeta.NextRowID() for v3 row lineage.
+func newV3MetadataWithNextRowID(t *testing.T, nextRowID int64) Metadata {
+	t.Helper()
+	spec := iceberg.NewPartitionSpec()
+	txn, _ := createTestTransactionWithMemIO(t, spec)
+	txn.meta.formatVersion = 3
+
+	firstRowID := int64(0)
+	addedRows := nextRowID
+	snap := Snapshot{
+		SnapshotID:     1000,
+		SequenceNumber: 1,
+		TimestampMs:    txn.meta.base.LastUpdatedMillis() + 1,
+		Summary:        &Summary{Operation: OpAppend},
+		FirstRowID:     &firstRowID,
+		AddedRows:      &addedRows,
+	}
+	require.NoError(t, txn.meta.AddSnapshot(&snap))
+	require.NoError(t, txn.meta.SetSnapshotRef(MainBranch, 1000, BranchRef))
+
+	meta, err := txn.meta.Build()
+	require.NoError(t, err)
+	require.Equal(t, nextRowID, meta.NextRowID())
+
+	return meta
+}
+
+// ptr returns a pointer to v, used in test helper expressions.
+// NOTE: ptr is also declared in pos_delete_partitioned_fanout_writer_test.go;
+// both files share the same package so only one declaration is needed.
+// This comment documents the shared usage — do not re-add a declaration here.
+
+// rebuildUpdate constructs an addSnapshotUpdate whose rebuildManifestList
+// closure simply records the freshParent it received and returns a new
+// snapshot whose ManifestList is the given newManifestList value.
+func rebuildUpdate(snap *Snapshot, newManifestList string, gotParent **Snapshot) *addSnapshotUpdate {
+	return &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, freshParent *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			*gotParent = freshParent
+			rebuilt := *snap
+			rebuilt.ManifestList = newManifestList
+
+			return &rebuilt, nil
+		},
+	}
+}
+
+// TestRebuildSnapshotUpdates_CallsClosureWithFreshParent verifies that
+// rebuildSnapshotUpdates invokes the rebuild closure and passes the fresh
+// branch head as freshParent.
+func TestRebuildSnapshotUpdates_CallsClosureWithFreshParent(t *testing.T) {
+	const oldManifest = "s3://bucket/old-manifest-list.avro"
+	const newManifest = "s3://bucket/new-manifest-list.avro"
+
+	// Build a fresh metadata that has a different snapshot than the one
+	// embedded in the update, so the parent-hasn't-changed guard is
+	// bypassed and the closure must be called.
+	freshHead := int64(42)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	parentID := int64(7) // original snap's parent — does NOT match freshHead (42)
+	snap := &Snapshot{
+		SnapshotID:       99,
+		ParentSnapshotID: &parentID,
+		ManifestList:     oldManifest,
+		Summary:          &Summary{Operation: OpAppend},
+	}
+
+	var receivedParent *Snapshot
+	upd := rebuildUpdate(snap, newManifest, &receivedParent)
+
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{upd},
+		freshMeta,
+		MainBranch,
+		iceio.LocalFS{},
+		1,
+	)
+	require.NoError(t, err)
+	require.Len(t, rebuilt, 1)
+	require.Len(t, orphaned, 1, "old manifest list must be recorded as orphaned")
+
+	// The closure should have been called with the branch's current head.
+	require.NotNil(t, receivedParent)
+	assert.Equal(t, freshHead, receivedParent.SnapshotID)
+
+	// The rebuilt update must carry the new manifest list.
+	addUpd, ok := rebuilt[0].(*addSnapshotUpdate)
+	require.True(t, ok)
+	assert.Equal(t, newManifest, addUpd.Snapshot.ManifestList)
+
+	// The superseded manifest list becomes an orphan.
+	assert.Equal(t, oldManifest, orphaned[0])
+}
+
+// TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged verifies that
+// rebuildSnapshotUpdates skips the rebuild when the update's snapshot
+// already has the fresh branch head as its parent (no-op retry).
+func TestRebuildSnapshotUpdates_SkipsWhenParentUnchanged(t *testing.T) {
+	const manifest = "s3://bucket/manifest-list.avro"
+
+	freshHead := int64(42)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	// Parent already equals the fresh head — rebuild must be skipped.
+	snap := &Snapshot{
+		SnapshotID:       99,
+		ParentSnapshotID: &freshHead, // same as fresh head
+		ManifestList:     manifest,
+		Summary:          &Summary{Operation: OpAppend},
+	}
+
+	called := false
+	upd := &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, _ *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			called = true
+
+			return snap, nil
+		},
+	}
+
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{upd},
+		freshMeta,
+		MainBranch,
+		iceio.LocalFS{},
+		1,
+	)
+	require.NoError(t, err)
+	assert.False(t, called, "rebuild closure must not be called when parent is already up-to-date")
+	assert.Empty(t, orphaned, "no orphans when rebuild is skipped")
+	assert.Same(t, upd, rebuilt[0].(*addSnapshotUpdate), "original update must pass through unchanged")
+}
+
+// TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates verifies that
+// updates without a rebuildManifestList closure are returned unmodified.
+func TestRebuildSnapshotUpdates_PassesThroughNonRebuildUpdates(t *testing.T) {
+	plainUpd := NewAddSnapshotUpdate(&Snapshot{
+		SnapshotID:   1,
+		ManifestList: "s3://bucket/no-rebuild.avro",
+		Summary:      &Summary{Operation: OpAppend},
+	})
+
+	// freshMeta may be nil — the plain update must not be touched.
+	rebuilt, orphaned, err := rebuildSnapshotUpdates(
+		t.Context(),
+		[]Update{plainUpd},
+		nil,
+		MainBranch,
+		iceio.LocalFS{},
+		0,
+	)
+	require.NoError(t, err)
+	assert.Empty(t, orphaned)
+	assert.Same(t, plainUpd, rebuilt[0].(*addSnapshotUpdate))
+}
+
+// TestRebuildSnapshotUpdates_PropagatesClosureError verifies that an error
+// returned by the rebuild closure surfaces as the function's return error.
+func TestRebuildSnapshotUpdates_PropagatesClosureError(t *testing.T) {
+	freshHead := int64(5)
+	freshMeta := newConflictTestMetadata(t, &freshHead)
+
+	parent := int64(1)
+	snap := &Snapshot{
+		SnapshotID:       10,
+		ParentSnapshotID: &parent,
+		ManifestList:     "s3://bucket/old.avro",
+		Summary:          &Summary{Operation: OpAppend},
+	}
+	wantErr := errors.New("simulated S3 write failure")
+	upd := &addSnapshotUpdate{
+		baseUpdate: baseUpdate{ActionName: UpdateAddSnapshot},
+		Snapshot:   snap,
+		rebuildManifestList: func(_ context.Context, _ Metadata, _ *Snapshot, _ iceio.WriteFileIO, _ int) (*Snapshot, error) {
+			return nil, wantErr
+		},
+	}
+
+	_, _, err := rebuildSnapshotUpdates(t.Context(), []Update{upd}, freshMeta, MainBranch, iceio.LocalFS{}, 1)
+	assert.ErrorIs(t, err, wantErr)
+}
+
+// ---------------------------------------------------------------------------
+// Fix 3 — newSeq derived from freshMeta.LastSequenceNumber()
+// ---------------------------------------------------------------------------
+
+// TestRebuildFn_SeqNumDerivedFromFreshMeta verifies that the rebuilt snapshot's
+// SequenceNumber equals freshMeta.LastSequenceNumber()+1, NOT
+// freshParent.SequenceNumber+1. A concurrent writer on a different branch can
+// advance the table-wide last-sequence-number without advancing this branch's
+// parent, so using freshParent.SequenceNumber+1 would violate the spec invariant.
+func TestRebuildFn_SeqNumDerivedFromFreshMeta(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	txn, wfs := createTestTransactionWithMemIO(t, spec)
+
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, _, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	require.NotNil(t, addSnap.rebuildManifestList, "rebuildManifestList closure must be set")
+
+	// Simulate a concurrent writer on another branch that bumped the global
+	// last-sequence-number to 99 without advancing this branch's parent.
+	// old code: newSeq = capturedSnapshot.SequenceNumber (stale, ≤ 99 — spec violation)
+	// new code: newSeq = freshMeta.LastSequenceNumber() + 1 = 100
+	freshMeta := newMetadataWithLastSeqNum(t, 99)
+	require.Equal(t, int64(99), freshMeta.LastSequenceNumber())
+
+	rebuilt, err := addSnap.rebuildManifestList(context.Background(), freshMeta, nil, wfs, 1)

Review Comment:
   Fixed exactly as suggested (one-line strengthening of the existing summary test).
   
   `TestRebuildFn_SummaryRebasedAgainstFreshParent` already wired both a non-nil `freshParent` and a `freshMeta`. The fixture now sets distinct values so the seq-num source is observable, and a single assertion pins it:
   
   ```go
   freshParent := &Snapshot{
       SnapshotID:     freshParentID,
       SequenceNumber: 99,                       // would yield rebuilt.SequenceNumber == 100 under the buggy path
       ManifestList:   parentManifestListPath,
       Summary:        &Summary{Operation: OpAppend, Properties: freshParentSummary},
   }
   freshMeta := newMetadataWithLastSeqNum(t, 1)  // correct path yields rebuilt.SequenceNumber == 2
   require.NotEqual(t, freshParent.SequenceNumber, freshMeta.LastSequenceNumber(),
       "test fixture must keep these distinct so the seq-num source is observable")
   
   // ... run rebuild ...
   
   require.Equal(t, freshMeta.LastSequenceNumber()+1, rebuilt.SequenceNumber,
       "newSeq must derive from freshMeta.LastSequenceNumber(), not freshParent.SequenceNumber")
   ```
   
   `TestRebuildFn_SeqNumDerivedFromFreshMeta` (the `freshParent=nil` case) is kept unchanged — it remains the v2 nil-parent regression coverage.



##########
table/commit_retry_test.go:
##########
@@ -266,3 +307,244 @@ func TestReadRetryConfig_ClampsNegativeProperties(t *testing.T) {
 	assert.Equal(t, uint(CommitMaxRetryWaitMsDefault), cfg.maxWaitMs)
 	assert.Equal(t, uint(CommitTotalRetryTimeoutMsDefault), cfg.totalTimeoutMs)
 }
+
+// ---------------------------------------------------------------------------
+// Fix 5 — mandatory WriteFileIO check at top of doCommit
+// ---------------------------------------------------------------------------
+
+// TestDoCommit_NonWriteFileIOReturnsError verifies that doCommit fails
+// immediately when the table's file system does not implement WriteFileIO.
+// A silent skip would reuse the stale manifest list — exactly the bug
+// this PR was designed to fix.
+func TestDoCommit_NonWriteFileIOReturnsError(t *testing.T) {
+	cat := &flakyCatalog{}
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/rotest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	cat.metadata = meta
+
+	tbl := New(
+		Identifier{"db", "ro-test"},
+		meta, "file:///tmp/rotest/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return readOnlyIO{}, nil },
+		cat,
+	)
+
+	_, doErr := tbl.doCommit(t.Context(), nil, nil)
+	require.Error(t, doErr, "doCommit must fail when FS does not implement WriteFileIO")
+	assert.Contains(t, doErr.Error(), "WriteFileIO",
+		"error message must mention WriteFileIO so callers understand the requirement")
+	assert.Equal(t, int32(0), cat.attempts.Load(),
+		"CommitTable must not be called when FS check fails")
+}
+
+// ---------------------------------------------------------------------------
+// Fix 6 — orphan cleanup via defer
+// ---------------------------------------------------------------------------
+
+// newOCCTable creates a table that uses the given wfs for its FS and the given
+// catalog for commits. meta should include retry-config properties so that
+// doCommit's retry loop allows at least one retry.
+func newOCCTable(t *testing.T, meta Metadata, wfs iceio.WriteFileIO, cat CatalogIO) *Table {
+	t.Helper()
+
+	return New(
+		Identifier{"db", "occ-cleanup-test"},
+		meta,
+		"mem://default/table-location/metadata/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return wfs, nil },
+		cat,
+	)
+}
+
+// newMemIOWithRetryMeta creates a test memIO and a matching table Metadata that
+// includes retry-config properties, so doCommit's retry loop allows retries.
+// The location matches createTestTransactionWithMemIO so they share the same
+// memIO for writing manifest files.
+func newMemIOWithRetryMeta(t *testing.T, spec iceberg.PartitionSpec) (*memIO, Metadata) {
+	t.Helper()
+	wfs := newMemIO(1<<20, nil)
+	schema := simpleSchema()
+	meta, err := NewMetadata(schema, &spec, UnsortedSortOrder, "mem://default/table-location",
+		iceberg.Properties{
+			CommitNumRetriesKey:          "3",
+			CommitMinRetryWaitMsKey:      "1",
+			CommitMaxRetryWaitMsKey:      "2",
+			CommitTotalRetryTimeoutMsKey: "60000",
+		})
+	require.NoError(t, err, "new metadata")
+
+	return wfs, meta
+}
+
+// TestDoCommit_OrphanCleanedOnSuccess verifies that manifest-list files
+// orphaned by OCC retries are removed after a successful commit. These files
+// are written during rebuild and must not leak on the happy path.
+func TestDoCommit_OrphanCleanedOnSuccess(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	// Build a transaction from the retry-enabled meta and commit via the producer.
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	// Pre-populate the original manifest list path in the IO so that the
+	// defer's wfs.Remove call has a concrete entry to delete.
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	// Catalog: fail once with ErrCommitFailed (triggers rebuild that orphans
+	// originalManifestList), then succeed.
+	cat := &sequentialCatalog{
+		metadata: meta,
+		errs:     []error{ErrCommitFailed},
+	}
+	tbl = newOCCTable(t, meta, wfs, cat)
+
+	_, err = tbl.doCommit(t.Context(), updates, reqs, withCommitBranch(MainBranch))
+	require.NoError(t, err, "doCommit must succeed on the second attempt")
+
+	_, stillExists := wfs.files[originalManifestList]

Review Comment:
   `limitedWriteCloser.Close()` now persists its buffered payload back into the parent `memIO.files` under the existing mutex, so the rebuilt manifest list path is genuinely present in `wfs.files` after `doCommit` runs (no more pre-population placeholder). All four `TestDoCommit_Orphan*` tests now assert the real contract:
   
   ```go
   // Orphan tests now do BOTH:
   //   (a) require the original ML was actually persisted before doCommit (real flow, not placeholder)
   //   (b) prove the live committed path was preserved
   require.Contains(t, wfs.files, originalManifestList,
       "producer.commit() must persist the manifest list via WriteFileIO before doCommit runs")
   
   // after doCommit success:
   committedSnap := cat.metadata.CurrentSnapshot()
   committedManifestList := committedSnap.ManifestList
   require.NotEqual(t, originalManifestList, committedManifestList,
       "committed manifest list must be the rebuilt path, not the original")
   require.Contains(t, wfs.files, committedManifestList,
       "live committed manifest list must be preserved by the cleanup defer")
   ```
   
   For `TestDoCommit_OrphanNotCleanedOnUnknownError` the assertion is symmetrical: ≥2 distinct manifest-list entries remain in `wfs.files` so callers can recover whichever the catalog silently accepted. A regression that flipped the cleanup loop to delete the committed path instead of the orphan now fails (b).



##########
table/commit_retry_test.go:
##########
@@ -266,3 +307,244 @@ func TestReadRetryConfig_ClampsNegativeProperties(t *testing.T) {
 	assert.Equal(t, uint(CommitMaxRetryWaitMsDefault), cfg.maxWaitMs)
 	assert.Equal(t, uint(CommitTotalRetryTimeoutMsDefault), cfg.totalTimeoutMs)
 }
+
+// ---------------------------------------------------------------------------
+// Fix 5 — mandatory WriteFileIO check at top of doCommit
+// ---------------------------------------------------------------------------
+
+// TestDoCommit_NonWriteFileIOReturnsError verifies that doCommit fails
+// immediately when the table's file system does not implement WriteFileIO.
+// A silent skip would reuse the stale manifest list — exactly the bug
+// this PR was designed to fix.
+func TestDoCommit_NonWriteFileIOReturnsError(t *testing.T) {
+	cat := &flakyCatalog{}
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/rotest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	cat.metadata = meta
+
+	tbl := New(
+		Identifier{"db", "ro-test"},
+		meta, "file:///tmp/rotest/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return readOnlyIO{}, nil },
+		cat,
+	)
+
+	_, doErr := tbl.doCommit(t.Context(), nil, nil)
+	require.Error(t, doErr, "doCommit must fail when FS does not implement WriteFileIO")
+	assert.Contains(t, doErr.Error(), "WriteFileIO",
+		"error message must mention WriteFileIO so callers understand the requirement")
+	assert.Equal(t, int32(0), cat.attempts.Load(),
+		"CommitTable must not be called when FS check fails")
+}
+
+// ---------------------------------------------------------------------------
+// Fix 6 — orphan cleanup via defer
+// ---------------------------------------------------------------------------
+
+// newOCCTable creates a table that uses the given wfs for its FS and the given
+// catalog for commits. meta should include retry-config properties so that
+// doCommit's retry loop allows at least one retry.
+func newOCCTable(t *testing.T, meta Metadata, wfs iceio.WriteFileIO, cat CatalogIO) *Table {
+	t.Helper()
+
+	return New(
+		Identifier{"db", "occ-cleanup-test"},
+		meta,
+		"mem://default/table-location/metadata/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return wfs, nil },
+		cat,
+	)
+}
+
+// newMemIOWithRetryMeta creates a test memIO and a matching table Metadata that
+// includes retry-config properties, so doCommit's retry loop allows retries.
+// The location matches createTestTransactionWithMemIO so they share the same
+// memIO for writing manifest files.
+func newMemIOWithRetryMeta(t *testing.T, spec iceberg.PartitionSpec) (*memIO, Metadata) {
+	t.Helper()
+	wfs := newMemIO(1<<20, nil)
+	schema := simpleSchema()
+	meta, err := NewMetadata(schema, &spec, UnsortedSortOrder, "mem://default/table-location",
+		iceberg.Properties{
+			CommitNumRetriesKey:          "3",
+			CommitMinRetryWaitMsKey:      "1",
+			CommitMaxRetryWaitMsKey:      "2",
+			CommitTotalRetryTimeoutMsKey: "60000",
+		})
+	require.NoError(t, err, "new metadata")
+
+	return wfs, meta
+}
+
+// TestDoCommit_OrphanCleanedOnSuccess verifies that manifest-list files
+// orphaned by OCC retries are removed after a successful commit. These files
+// are written during rebuild and must not leak on the happy path.
+func TestDoCommit_OrphanCleanedOnSuccess(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	// Build a transaction from the retry-enabled meta and commit via the producer.
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	// Pre-populate the original manifest list path in the IO so that the
+	// defer's wfs.Remove call has a concrete entry to delete.
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	// Catalog: fail once with ErrCommitFailed (triggers rebuild that orphans
+	// originalManifestList), then succeed.
+	cat := &sequentialCatalog{
+		metadata: meta,
+		errs:     []error{ErrCommitFailed},
+	}
+	tbl = newOCCTable(t, meta, wfs, cat)
+
+	_, err = tbl.doCommit(t.Context(), updates, reqs, withCommitBranch(MainBranch))
+	require.NoError(t, err, "doCommit must succeed on the second attempt")
+
+	_, stillExists := wfs.files[originalManifestList]
+	assert.False(t, stillExists,
+		"orphaned manifest list must be removed after successful commit")
+}
+
+// TestDoCommit_OrphanNotCleanedOnUnknownError verifies that manifest-list
+// files are NOT removed when CommitTable returns an unknown non-ErrCommitFailed
+// error (5xx / gateway timeout). In that case the catalog may have silently
+// accepted the commit, meaning one of the "orphaned" files is actually the
+// live snapshot. Deleting it would permanently corrupt the table.
+func TestDoCommit_OrphanNotCleanedOnUnknownError(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	unknown5xxErr := errors.New("simulated 5xx: internal server error")
+	// Catalog: fail once (ErrCommitFailed → rebuild → orphan created),
+	// then return a 5xx (non-ErrCommitFailed → cleanupOrphans=false).
+	cat := &sequentialCatalog{
+		metadata: meta,
+		errs:     []error{ErrCommitFailed, unknown5xxErr},
+	}
+	tbl = newOCCTable(t, meta, wfs, cat)
+
+	_, err = tbl.doCommit(t.Context(), updates, reqs, withCommitBranch(MainBranch))
+	require.ErrorIs(t, err, unknown5xxErr, "5xx error must propagate")
+
+	_, stillExists := wfs.files[originalManifestList]
+	assert.True(t, stillExists,
+		"orphaned manifest list must NOT be removed when commit outcome is unknown (5xx)")
+}
+
+// TestDoCommit_OrphanCleanedOnCommitDiverged verifies that manifest-list files
+// orphaned by rebuild attempts are removed when ErrCommitDiverged is returned
+// by a conflict validator. Diverged commits are terminal (no retry), and since
+// neither of the orphaned files was ever accepted by the catalog, they are safe
+// to delete. The defer cleanup runs with cleanupOrphans=true on this path.
+func TestDoCommit_OrphanCleanedOnCommitDiverged(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	// freshMeta has a snapshot on MainBranch so validators run on the retry attempt.
+	freshID := int64(42)
+	freshMeta := newConflictTestMetadataWithProps(t, &freshID, iceberg.Properties{
+		CommitNumRetriesKey:          "3",
+		CommitMinRetryWaitMsKey:      "1",
+		CommitMaxRetryWaitMsKey:      "2",
+		CommitTotalRetryTimeoutMsKey: "60000",
+	})
+
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	// Catalog: fail once (ErrCommitFailed → rebuild → orphan created).
+	// LoadTable returns freshMeta (has branch snapshot → validators run on retry).
+	// Validator returns ErrCommitDiverged immediately.
+	cat := &sequentialCatalog{
+		metadata: meta,
+		loadMeta: freshMeta,
+		errs:     []error{ErrCommitFailed},
+	}
+	tbl = newOCCTable(t, meta, wfs, cat)
+
+	divergedValidator := func(*conflictContext) error { return ErrCommitDiverged }
+
+	_, err = tbl.doCommit(t.Context(), updates, reqs,
+		withCommitBranch(MainBranch),
+		withCommitValidators(divergedValidator),
+	)
+	require.ErrorIs(t, err, ErrCommitDiverged)
+
+	// The defer must have fired with cleanupOrphans=true and removed the orphan.
+	_, stillExists := wfs.files[originalManifestList]
+	assert.False(t, stillExists,
+		"orphaned manifest list must be removed even on ErrCommitDiverged: "+
+			"the file was never accepted by the catalog so it is safe to delete")
+}
+
+// TestDoCommit_OrphanCleanedOnRetriesExhausted verifies that when every retry
+// attempt fails with ErrCommitFailed and the loop exits with the budget
+// exhausted, the defer still fires with cleanupOrphans=true. None of the
+// orphaned manifest-list files were ever accepted by the catalog, so they are
+// safe to delete on this terminal exit.
+func TestDoCommit_OrphanCleanedOnRetriesExhausted(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	// numRetries=3 → 4 attempts; every attempt fails with ErrCommitFailed.
+	cat := &sequentialCatalog{
+		metadata: meta,
+		errs:     []error{ErrCommitFailed, ErrCommitFailed, ErrCommitFailed, ErrCommitFailed},

Review Comment:
   Added `TestDoCommit_RetryProgressesFreshMeta` (in `commit_retry_test.go`) that drives a **real producer commit** with `rebuildManifestList` closure attached through 2 `ErrCommitFailed` retries + 1 success. The new `progressingRebuildCatalog`:
   
   - writes a real (empty) manifest-list file at a unique path on every peer graft, so the rebuild closure's `freshParent.Manifests(fio)` succeeds on each retry;
   - publishes a `Summary` with cumulative `total-data-files` on each peer, so the rebuild's summary recomputation observes an evolving parent;
   - applies the accepted updates through `MetadataBuilder` so the test can read the live committed snapshot.
   
   A single test now exercises every Round 1 fix at once:
   
   ```go
   // (a) commitTableCalls: 2 fails + 1 success.
   require.Equal(t, int32(3), cat.commitTableCalls.Load())
   
   // (b) Orphan list grows by N — every retry submits a UNIQUE manifest list path.
   seen := make(map[string]struct{}, 3)
   for i, p := range cat.observedManifestLists { /* assert distinct */ }
   require.Equal(t, originalManifestList, cat.observedManifestLists[0])
   
   // (c) freshMeta.LastSequenceNumber() observed at LoadTable strictly advances.
   require.Greater(t, cat.loadedLastSeqNums[1], cat.loadedLastSeqNums[0])
   
   // (d) Fix 3 — submitted seq numbers equal freshMeta.LastSequenceNumber()+1 per retry.
   require.Equal(t, cat.loadedLastSeqNums[0]+1, cat.observedSeqNums[1])
   require.Equal(t, cat.loadedLastSeqNums[1]+1, cat.observedSeqNums[2])
   
   // (e) Fix 2 — summary chains on top of evolving freshParent.
   require.Equal(t, 3, gotDataFiles,
       "committed total-data-files must = peer cumulative (2) + producer (1) = 3; "+
           "a regression that ignored freshParent.Summary would publish 1")
   
   // (f) Fix 6 — defer cleanup: live committed ML preserved; orphans removed.
   require.Contains(t, wfs.files, committedML)
   require.NotContains(t, wfs.files, originalManifestList)
   require.NotContains(t, wfs.files, cat.observedManifestLists[1])
   ```
   
   Note on helper choice: `headTrackingCatalog` advances metadata only on a *successful* `CommitTable`; the ≥2-retry-progression scenario requires the catalog state to advance **between failed retries**. `progressingRebuildCatalog` adds exactly that capability while keeping the validate-and-track invariants minimal.
   
   (An earlier version of this test passed `nil` updates to `doCommit`, which never invoked `rebuildSnapshotUpdates` — only the retry-loop refresh path was exercised. `e35d2bd` corrects this so the rebuild closure is exercised for real on every retry, which is what was actually requested.)



##########
table/commit_retry_test.go:
##########
@@ -266,3 +307,244 @@ func TestReadRetryConfig_ClampsNegativeProperties(t *testing.T) {
 	assert.Equal(t, uint(CommitMaxRetryWaitMsDefault), cfg.maxWaitMs)
 	assert.Equal(t, uint(CommitTotalRetryTimeoutMsDefault), cfg.totalTimeoutMs)
 }
+
+// ---------------------------------------------------------------------------
+// Fix 5 — mandatory WriteFileIO check at top of doCommit
+// ---------------------------------------------------------------------------
+
+// TestDoCommit_NonWriteFileIOReturnsError verifies that doCommit fails
+// immediately when the table's file system does not implement WriteFileIO.
+// A silent skip would reuse the stale manifest list — exactly the bug
+// this PR was designed to fix.
+func TestDoCommit_NonWriteFileIOReturnsError(t *testing.T) {
+	cat := &flakyCatalog{}
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/rotest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	cat.metadata = meta
+
+	tbl := New(
+		Identifier{"db", "ro-test"},
+		meta, "file:///tmp/rotest/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return readOnlyIO{}, nil },
+		cat,
+	)
+
+	_, doErr := tbl.doCommit(t.Context(), nil, nil)
+	require.Error(t, doErr, "doCommit must fail when FS does not implement WriteFileIO")
+	assert.Contains(t, doErr.Error(), "WriteFileIO",

Review Comment:
   `table/table.go` now exposes a sentinel:
   
   ```go
   // ErrWriteIORequired is returned by doCommit when the table's file system
   // does not implement icebergio.WriteFileIO. Callers can detect this via
   // errors.Is so the test does not depend on the rendered error message.
   var ErrWriteIORequired = errors.New("commit: file system does not implement WriteFileIO")
   
   // ... in doCommit:
   return nil, fmt.Errorf("%w: manifest list rebuild requires write access", ErrWriteIORequired)
   ```
   
   `TestDoCommit_NonWriteFileIOReturnsError` uses `require.ErrorIs(t, doErr, ErrWriteIORequired)` and the substring assertion is removed. The two `TestComputeOwnManifests_*Error` tests received the same treatment (you flagged them as the same shape):
   
   ```go
   require.ErrorIs(t, err, ErrSnapshotNotFound,
       "production wraps ErrSnapshotNotFound; pin meaning via errors.Is so a regression "+
           "that swallows the lookup error and returns a programming-bug error would fail this test")
   
   require.ErrorIs(t, err, fs.ErrNotExist,
       "production wraps the underlying IO error; pin meaning via errors.Is so a regression that "+
           "swallows the IO error and returns a programming-bug error would fail this test")
   ```
   
   `ErrSnapshotNotFound` is the sentinel already wrapped by `MetadataBuilder.SnapshotByID`, so no new sentinel was needed for `_SnapshotByIDError`.



-- 
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

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

["mzzz-zzm (via GitHub)" <gi...@apache.org>]

mzzz-zzm commented on code in PR #982:
URL: https://github.com/apache/iceberg-go/pull/982#discussion_r3202200437


##########
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:
   The signature is now `([]iceberg.ManifestFile, error)`. All three error paths return `nil, err` (or a wrapped error for the missing-parent case) rather than `allManifests`:
   
   ```go
   parent, err := sp.txn.meta.SnapshotByID(sp.parentSnapshotID)
   if err != nil {
       return nil, fmt.Errorf("computeOwnManifests: lookup parent snapshot %d: %w", sp.parentSnapshotID, err)
   }
   if parent == nil {
       return nil, fmt.Errorf("computeOwnManifests: parent snapshot %d not found", sp.parentSnapshotID)
   }
   parentManifests, err := parent.Manifests(sp.io)
   if err != nil {
       return nil, fmt.Errorf("computeOwnManifests: read parent manifests: %w", err)
   }
   ```
   
   The `parentSnapshotID <= 0` early return (`return allManifests, nil`) is unchanged — that is the valid all-new case for a new table.
   
   The call site in `commit()` now checks the error and aborts:
   
   ```go
   ownManifests, err := sp.computeOwnManifests(newManifests)
   if err != nil {
       return nil, nil, err
   }
   ```
   
   Tests added: `TestComputeOwnManifests_NewTable`, `TestComputeOwnManifests_SnapshotByIDError`, `TestComputeOwnManifests_ParentManifestsIOError`.



-- 
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

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

["mzzz-zzm (via GitHub)" <gi...@apache.org>]

mzzz-zzm commented on code in PR #982:
URL: https://github.com/apache/iceberg-go/pull/982#discussion_r3202207361


##########
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:
   The `WriteFileIO` type assertion is now hoisted to the top of `doCommit`, before the retry loop. A non-`WriteFileIO` filesystem returns an error immediately — there is no silent fallback:
   
   ```go
   wfs, ok := fs.(icebergio.WriteFileIO)
   if !ok {
       return nil, fmt.Errorf("commit: file system does not implement WriteFileIO: "+
           "manifest list rebuild requires write access")
   }
   ```
   
   Both inline `if wfs, ok := fs.(icebergio.WriteFileIO); ok { ... }` guards in the original code are removed. The rest of `doCommit` uses `wfs` directly.
   
   Test added: `TestDoCommit_NonWriteFileIOReturnsError` — passes a read-only FS that does not implement `WriteFileIO`; asserts an immediate error before any commit attempt.



-- 
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

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

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

laskoviymishka merged PR #982:
URL: https://github.com/apache/iceberg-go/pull/982


-- 
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

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_r3202392999


##########
table/commit_retry_test.go:
##########
@@ -266,3 +307,244 @@ func TestReadRetryConfig_ClampsNegativeProperties(t *testing.T) {
 	assert.Equal(t, uint(CommitMaxRetryWaitMsDefault), cfg.maxWaitMs)
 	assert.Equal(t, uint(CommitTotalRetryTimeoutMsDefault), cfg.totalTimeoutMs)
 }
+
+// ---------------------------------------------------------------------------
+// Fix 5 — mandatory WriteFileIO check at top of doCommit
+// ---------------------------------------------------------------------------
+
+// TestDoCommit_NonWriteFileIOReturnsError verifies that doCommit fails
+// immediately when the table's file system does not implement WriteFileIO.
+// A silent skip would reuse the stale manifest list — exactly the bug
+// this PR was designed to fix.
+func TestDoCommit_NonWriteFileIOReturnsError(t *testing.T) {
+	cat := &flakyCatalog{}
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+	)
+	meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "file:///tmp/rotest",
+		iceberg.Properties{PropertyFormatVersion: "2"})
+	require.NoError(t, err)
+	cat.metadata = meta
+
+	tbl := New(
+		Identifier{"db", "ro-test"},
+		meta, "file:///tmp/rotest/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return readOnlyIO{}, nil },
+		cat,
+	)
+
+	_, doErr := tbl.doCommit(t.Context(), nil, nil)
+	require.Error(t, doErr, "doCommit must fail when FS does not implement WriteFileIO")
+	assert.Contains(t, doErr.Error(), "WriteFileIO",
+		"error message must mention WriteFileIO so callers understand the requirement")
+	assert.Equal(t, int32(0), cat.attempts.Load(),
+		"CommitTable must not be called when FS check fails")
+}
+
+// ---------------------------------------------------------------------------
+// Fix 6 — orphan cleanup via defer
+// ---------------------------------------------------------------------------
+
+// newOCCTable creates a table that uses the given wfs for its FS and the given
+// catalog for commits. meta should include retry-config properties so that
+// doCommit's retry loop allows at least one retry.
+func newOCCTable(t *testing.T, meta Metadata, wfs iceio.WriteFileIO, cat CatalogIO) *Table {
+	t.Helper()
+
+	return New(
+		Identifier{"db", "occ-cleanup-test"},
+		meta,
+		"mem://default/table-location/metadata/v1.metadata.json",
+		func(context.Context) (iceio.IO, error) { return wfs, nil },
+		cat,
+	)
+}
+
+// newMemIOWithRetryMeta creates a test memIO and a matching table Metadata that
+// includes retry-config properties, so doCommit's retry loop allows retries.
+// The location matches createTestTransactionWithMemIO so they share the same
+// memIO for writing manifest files.
+func newMemIOWithRetryMeta(t *testing.T, spec iceberg.PartitionSpec) (*memIO, Metadata) {
+	t.Helper()
+	wfs := newMemIO(1<<20, nil)
+	schema := simpleSchema()
+	meta, err := NewMetadata(schema, &spec, UnsortedSortOrder, "mem://default/table-location",
+		iceberg.Properties{
+			CommitNumRetriesKey:          "3",
+			CommitMinRetryWaitMsKey:      "1",
+			CommitMaxRetryWaitMsKey:      "2",
+			CommitTotalRetryTimeoutMsKey: "60000",
+		})
+	require.NoError(t, err, "new metadata")
+
+	return wfs, meta
+}
+
+// TestDoCommit_OrphanCleanedOnSuccess verifies that manifest-list files
+// orphaned by OCC retries are removed after a successful commit. These files
+// are written during rebuild and must not leak on the happy path.
+func TestDoCommit_OrphanCleanedOnSuccess(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	// Build a transaction from the retry-enabled meta and commit via the producer.
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	// Pre-populate the original manifest list path in the IO so that the
+	// defer's wfs.Remove call has a concrete entry to delete.
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	// Catalog: fail once with ErrCommitFailed (triggers rebuild that orphans
+	// originalManifestList), then succeed.
+	cat := &sequentialCatalog{
+		metadata: meta,
+		errs:     []error{ErrCommitFailed},
+	}
+	tbl = newOCCTable(t, meta, wfs, cat)
+
+	_, err = tbl.doCommit(t.Context(), updates, reqs, withCommitBranch(MainBranch))
+	require.NoError(t, err, "doCommit must succeed on the second attempt")
+
+	_, stillExists := wfs.files[originalManifestList]
+	assert.False(t, stillExists,
+		"orphaned manifest list must be removed after successful commit")
+}
+
+// TestDoCommit_OrphanNotCleanedOnUnknownError verifies that manifest-list
+// files are NOT removed when CommitTable returns an unknown non-ErrCommitFailed
+// error (5xx / gateway timeout). In that case the catalog may have silently
+// accepted the commit, meaning one of the "orphaned" files is actually the
+// live snapshot. Deleting it would permanently corrupt the table.
+func TestDoCommit_OrphanNotCleanedOnUnknownError(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	unknown5xxErr := errors.New("simulated 5xx: internal server error")
+	// Catalog: fail once (ErrCommitFailed → rebuild → orphan created),
+	// then return a 5xx (non-ErrCommitFailed → cleanupOrphans=false).
+	cat := &sequentialCatalog{
+		metadata: meta,
+		errs:     []error{ErrCommitFailed, unknown5xxErr},
+	}
+	tbl = newOCCTable(t, meta, wfs, cat)
+
+	_, err = tbl.doCommit(t.Context(), updates, reqs, withCommitBranch(MainBranch))
+	require.ErrorIs(t, err, unknown5xxErr, "5xx error must propagate")
+
+	_, stillExists := wfs.files[originalManifestList]
+	assert.True(t, stillExists,
+		"orphaned manifest list must NOT be removed when commit outcome is unknown (5xx)")
+}
+
+// TestDoCommit_OrphanCleanedOnCommitDiverged verifies that manifest-list files
+// orphaned by rebuild attempts are removed when ErrCommitDiverged is returned
+// by a conflict validator. Diverged commits are terminal (no retry), and since
+// neither of the orphaned files was ever accepted by the catalog, they are safe
+// to delete. The defer cleanup runs with cleanupOrphans=true on this path.
+func TestDoCommit_OrphanCleanedOnCommitDiverged(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	// freshMeta has a snapshot on MainBranch so validators run on the retry attempt.
+	freshID := int64(42)
+	freshMeta := newConflictTestMetadataWithProps(t, &freshID, iceberg.Properties{
+		CommitNumRetriesKey:          "3",
+		CommitMinRetryWaitMsKey:      "1",
+		CommitMaxRetryWaitMsKey:      "2",
+		CommitTotalRetryTimeoutMsKey: "60000",
+	})
+
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	// Catalog: fail once (ErrCommitFailed → rebuild → orphan created).
+	// LoadTable returns freshMeta (has branch snapshot → validators run on retry).
+	// Validator returns ErrCommitDiverged immediately.
+	cat := &sequentialCatalog{
+		metadata: meta,
+		loadMeta: freshMeta,
+		errs:     []error{ErrCommitFailed},
+	}
+	tbl = newOCCTable(t, meta, wfs, cat)
+
+	divergedValidator := func(*conflictContext) error { return ErrCommitDiverged }
+
+	_, err = tbl.doCommit(t.Context(), updates, reqs,
+		withCommitBranch(MainBranch),
+		withCommitValidators(divergedValidator),
+	)
+	require.ErrorIs(t, err, ErrCommitDiverged)
+
+	// The defer must have fired with cleanupOrphans=true and removed the orphan.
+	_, stillExists := wfs.files[originalManifestList]
+	assert.False(t, stillExists,
+		"orphaned manifest list must be removed even on ErrCommitDiverged: "+
+			"the file was never accepted by the catalog so it is safe to delete")
+}
+
+// TestDoCommit_OrphanCleanedOnRetriesExhausted verifies that when every retry
+// attempt fails with ErrCommitFailed and the loop exits with the budget
+// exhausted, the defer still fires with cleanupOrphans=true. None of the
+// orphaned manifest-list files were ever accepted by the catalog, so they are
+// safe to delete on this terminal exit.
+func TestDoCommit_OrphanCleanedOnRetriesExhausted(t *testing.T) {
+	spec := iceberg.NewPartitionSpec()
+	wfs, meta := newMemIOWithRetryMeta(t, spec)
+
+	tbl := newOCCTable(t, meta, wfs, nil)
+	txn := tbl.NewTransaction()
+	sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+	sp.appendDataFile(newTestDataFile(t, spec, "mem://default/table-location/data/f.parquet", nil))
+
+	updates, reqs, err := sp.commit(context.Background())
+	require.NoError(t, err)
+	addSnap := updates[0].(*addSnapshotUpdate)
+	originalManifestList := addSnap.Snapshot.ManifestList
+
+	wfs.files[originalManifestList] = []byte("placeholder")
+
+	// numRetries=3 → 4 attempts; every attempt fails with ErrCommitFailed.
+	cat := &sequentialCatalog{
+		metadata: meta,
+		errs:     []error{ErrCommitFailed, ErrCommitFailed, ErrCommitFailed, ErrCommitFailed},

Review Comment:
   [Non-blocking] None of the `TestDoCommit_Orphan*` tests progress `freshMeta` between retries — `sequentialCatalog.metadata` is the same object on every `LoadTable` call, so on each retry the rebuild closure sees a `freshHead` that equals the original parent (or is nil for an empty table). The skip-if-parent-unchanged predicate could short-circuit, and even when it doesn't, the closure runs in degenerate mode where every per-retry recompute (newSeq, firstRowID, summary) reproduces the captured-attempt-0 values. So this test exercises the orphan-cleanup state machine, but not the actual *retry* semantics the fixes target. Worth one end-to-end test that drives ≥2 ErrCommitFailed retries with a `freshMeta` that progresses between attempts (the existing `headTrackingCatalog` in `commit_refresh_replay_test.go` is most of the way there). That single test would close more coverage than the four orphan-cleanup variants combined — every fix in this PR is exercised at once: orphan list
  grows by N, each rebuild sees a different `LastSequenceNumber`/`NextRowID`, summary chains on top of an evolving parent.



-- 
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

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

["mzzz-zzm (via GitHub)" <gi...@apache.org>]

mzzz-zzm commented on code in PR #982:
URL: https://github.com/apache/iceberg-go/pull/982#discussion_r3202156649


##########
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:
   Inside `rebuildFn`, the rebuilt summary is now recomputed against the fresh parent instead of copied from `capturedSnapshot`. `deltaSummary` is derived from `capturedSnapshot.Summary` minus the fresh parent's counts — it represents only the files this producer added/removed, which is invariant across retries:
   
   ```go
   if freshParent != nil && freshParent.Summary != nil && capturedSnapshot.Summary != nil {
       deltaSummary := computeDeltaSummary(capturedSnapshot.Summary, freshParent.Summary)
       if s, sumErr := updateSnapshotSummaries(deltaSummary, freshParent.Summary.Properties); sumErr == nil {
           rebuilt.Summary = s
       }
   }
   ```
   
   When `freshParent == nil` (first snapshot on the table), `rebuilt.Summary` stays as `capturedSnapshot.Summary` — there is no prior state to rebase against, so the attempt-0 summary is correct.
   
   Tests added: `TestRebuildFn_SummaryRebasedAgainstFreshParent` (concurrent writer adds 5 files / 500 records / 5000 bytes; this producer adds 1 / 1 / 1 → asserts rebuilt `total-data-files == 6`, `total-records == 501`, `total-files-size == 5001`), `TestRebuildFn_SummaryFreshParentNilKeepsCapturedSummary` (no parent to rebase against → rebuilt summary equals captured summary).



-- 
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

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

["mzzz-zzm (via GitHub)" <gi...@apache.org>]

mzzz-zzm commented on code in PR #982:
URL: https://github.com/apache/iceberg-go/pull/982#discussion_r3202177808


##########
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:
   `newSeq` is now always derived from the table-wide `LastSequenceNumber` in the fresh metadata, not from the branch-local parent snapshot's `SequenceNumber`:
   
   ```go
   var newSeq int64
   if formatVersion >= 2 {
       newSeq = freshMeta.LastSequenceNumber() + 1
   }
   ```
   
   Both the `freshParent != nil` and `freshParent == nil` branches collapse to this single expression. For v1 tables, `newSeq` stays `0` per spec.
   
   `freshMeta` is threaded into the closure as a parameter (same change that also fixes Comments 1 and 2), so it reflects the metadata state at retry time, never at attempt 0.
   
   Tests added: `TestRebuildFn_SeqNumDerivedFromFreshMeta` (peer bumps `lastSeq` to 100 on another branch; asserts rebuilt `SequenceNumber == 101`), `TestRebuildFn_SeqNumV1TableIsZero` (v1 table; asserts `SequenceNumber == 0`).



-- 
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