This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 4acc0415c fix(table): delete manifests that superseded commit attempts
wrote (#2343)
4acc0415c is described below
commit 4acc0415c28b57caee589188f6f66c4b95dbda0c
Author: Harrison Crosse <[email protected]>
AuthorDate: Wed Oct 28 15:47:34 2026 -0400
fix(table): delete manifests that superseded commit attempts wrote (#2343)
Merge-append and overwrite rebuild their manifests against the fresh
parent on every retry and every resumed Commit. A merge writes new
merged manifests, and an overwrite writes rewritten parent manifests
and tombstones. doCommit tracked only the manifest lists those passes
wrote, so a retried commit that landed left attempt 0's manifests and
every superseded rebuild's manifests unreferenced on storage, and a
failed commit left the manifests its rebuilds wrote.
Every snapshot producer now records the manifests each pass writes,
replacing the rewrite producer's private superseded list, and doCommit
treats them like manifest lists. Once a commit lands it deletes the
recorded manifests no committed snapshot references, including attempt
0's. After a known failure it deletes only those this call's rebuilds
wrote, and after an unknown commit state it deletes nothing. Before
deleting, it reads every committed snapshot's manifest list and keeps
the files they reference.
A merge that consumes its added-content manifest now deletes it once
the commit lands, with or without a retry, as Java's
MergingSnapshotProducer does. The rewrite producer still keeps its
merged manifests after a failed commit, since a resume can reuse them.
Signed-off-by: Harrison Crosse <[email protected]>
---
table/retry_manifest_cleanup_test.go | 436 +++++++++++++++++++++++++++++++
table/rewrite_manifests.go | 20 +-
table/rewrite_manifests_internal_test.go | 3 +-
table/snapshot_producers.go | 86 +++++-
table/table.go | 53 +++-
table/updates.go | 8 +-
6 files changed, 563 insertions(+), 43 deletions(-)
diff --git a/table/retry_manifest_cleanup_test.go
b/table/retry_manifest_cleanup_test.go
new file mode 100644
index 000000000..864678751
--- /dev/null
+++ b/table/retry_manifest_cleanup_test.go
@@ -0,0 +1,436 @@
+// 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"
+ "maps"
+ "path/filepath"
+ "slices"
+ "strings"
+ "sync"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+const retryCleanupLocation = "mem://default/table-location"
+
+// peerHeadsCatalog serves heads in order, one per LoadTable, and stays on the
+// last. CommitTable returns errs[n] for its nth call and otherwise applies the
+// updates to the head it last served.
+type peerHeadsCatalog struct {
+ heads []Metadata
+ loads int
+ current Metadata
+ errs []error
+ attempts int
+}
+
+func (c *peerHeadsCatalog) LoadTable(_ context.Context, ident Identifier)
(*Table, error) {
+ c.current = c.heads[min(c.loads, len(c.heads)-1)]
+ c.loads++
+
+ return New(ident, c.current, "", nil, c), nil
+}
+
+func (c *peerHeadsCatalog) CommitTable(_ context.Context, _ Identifier, _
[]Requirement, updates []Update) (Metadata, string, error) {
+ n := c.attempts
+ c.attempts++
+ if n < len(c.errs) && c.errs[n] != nil {
+ return nil, "", c.errs[n]
+ }
+ meta, err := UpdateTableMetadata(c.current, updates, "")
+ if err != nil {
+ return nil, "", err
+ }
+ c.current = meta
+
+ return meta, "", nil
+}
+
+// landingCatalog calls landed with the updates of the commit it accepts.
+type landingCatalog struct {
+ *peerHeadsCatalog
+ landed func([]Update)
+}
+
+func (c *landingCatalog) CommitTable(ctx context.Context, ident Identifier,
reqs []Requirement, updates []Update) (Metadata, string, error) {
+ meta, loc, err := c.peerHeadsCatalog.CommitTable(ctx, ident, reqs,
updates)
+ if err == nil {
+ c.landed(updates)
+ }
+
+ return meta, loc, err
+}
+
+// unreadableListIO fails Open for the path in unreadable and records every
+// Remove.
+type unreadableListIO struct {
+ *memIO
+ mu sync.Mutex
+ unreadable string
+ removed []string
+}
+
+func (u *unreadableListIO) Open(ctx context.Context, name string) (iceio.File,
error) {
+ u.mu.Lock()
+ unreadable := u.unreadable
+ u.mu.Unlock()
+ if name == unreadable {
+ return nil, errors.New("simulated read failure")
+ }
+
+ return u.memIO.Open(ctx, name)
+}
+
+func (u *unreadableListIO) Remove(ctx context.Context, name string) error {
+ u.mu.Lock()
+ u.removed = append(u.removed, name)
+ u.mu.Unlock()
+
+ return u.memIO.Remove(ctx, name)
+}
+
+// retryCleanupFixture is a table on memIO whose base head h0 holds f1 and f2
+// in one manifest. Peers then appended f3 (h1) and f5 (h2).
+type retryCleanupFixture struct {
+ wfs *memIO
+ h0, h1, h2 Metadata
+ preExisting []string
+}
+
+func newRetryCleanupFixture(t *testing.T, numRetries string, props
iceberg.Properties) retryCleanupFixture {
+ t.Helper()
+ spec := iceberg.NewPartitionSpec()
+ all := iceberg.Properties{
+ PropertyFormatVersion: "2",
+ CommitNumRetriesKey: numRetries,
+ CommitMinRetryWaitMsKey: "1",
+ CommitMaxRetryWaitMsKey: "2",
+ CommitTotalRetryTimeoutMsKey: "60000",
+ }
+ maps.Copy(all, props)
+ meta, err := NewMetadata(simpleSchema(), &spec, UnsortedSortOrder,
retryCleanupLocation, all)
+ require.NoError(t, err)
+
+ f := retryCleanupFixture{wfs: newMemIO(1<<20, nil)}
+ tbl := newOCCTable(t, meta, f.wfs, &sequentialCatalog{metadata: meta})
+ tbl = appendRetryCleanupFiles(t, tbl, "f1", "f2")
+ f.h0 = tbl.Metadata()
+ tbl = appendRetryCleanupFiles(t, tbl, "f3")
+ f.h1 = tbl.Metadata()
+ f.h2 = appendRetryCleanupFiles(t, tbl, "f5").Metadata()
+ f.preExisting = slices.Sorted(maps.Keys(f.wfs.files))
+
+ return f
+}
+
+func retryCleanupFile(t *testing.T, name string) iceberg.DataFile {
+ t.Helper()
+
+ return newTestDataFile(t, iceberg.NewPartitionSpec(),
retryCleanupLocation+"/data/"+name+".parquet", nil)
+}
+
+func appendRetryCleanupFiles(t *testing.T, tbl *Table, names ...string) *Table
{
+ t.Helper()
+ files := make([]iceberg.DataFile, 0, len(names))
+ for _, name := range names {
+ files = append(files, retryCleanupFile(t, name))
+ }
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFiles(t.Context(), files, nil))
+ out, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ return out
+}
+
+func (f retryCleanupFixture) newTable(t *testing.T, heads []Metadata, errs
...error) (*Table, *peerHeadsCatalog) {
+ t.Helper()
+ cat := &peerHeadsCatalog{current: f.h0, heads: heads, errs: errs}
+
+ return newOCCTable(t, f.h0, f.wfs, cat), cat
+}
+
+func (f retryCleanupFixture) newFiles() []string {
+ files := make([]string, 0, len(f.wfs.files))
+ for path := range f.wfs.files {
+ if !slices.Contains(f.preExisting, path) {
+ files = append(files, path)
+ }
+ }
+ slices.Sort(files)
+
+ return files
+}
+
+// assertOnlyCommittedFilesAdded checks that the files written since the
+// fixture was built are exactly the manifest lists and manifests of the
+// snapshots committed on top of the peers, and that each of them is readable.
+func (f retryCleanupFixture) assertOnlyCommittedFilesAdded(t *testing.T,
committed *Table) {
+ t.Helper()
+ prior := make(map[int64]struct{})
+ for _, s := range f.h2.Snapshots() {
+ prior[s.SnapshotID] = struct{}{}
+ }
+
+ want := make(map[string]struct{})
+ for _, path := range f.preExisting {
+ want[path] = struct{}{}
+ }
+ var added int
+ for _, s := range committed.Metadata().Snapshots() {
+ if _, ok := prior[s.SnapshotID]; ok {
+ continue
+ }
+ added++
+ want[s.ManifestList] = struct{}{}
+ manifests, err := s.Manifests(f.wfs)
+ require.NoError(t, err, "committed manifest list %s must be
readable", s.ManifestList)
+ for _, m := range manifests {
+ want[m.FilePath()] = struct{}{}
+ for _, err := range m.Entries(f.wfs, false) {
+ require.NoError(t, err, "committed manifest %s
must be readable", m.FilePath())
+ }
+ }
+ }
+ require.NotZero(t, added, "the commit must add a snapshot")
+ assert.Equal(t, slices.Sorted(maps.Keys(want)),
slices.Sorted(maps.Keys(f.wfs.files)),
+ "storage must hold the pre-existing files and the committed
snapshots' files, nothing else")
+}
+
+func headManifests(t *testing.T, tbl *Table, fs *memIO) (*Snapshot,
[]iceberg.ManifestFile) {
+ t.Helper()
+ head := tbl.CurrentSnapshot()
+ require.NotNil(t, head)
+ manifests, err := head.Manifests(fs)
+ require.NoError(t, err)
+
+ return head, manifests
+}
+
+var retryCleanupProducers = []struct {
+ name string
+ props iceberg.Properties
+ stage func(*testing.T, *Transaction)
+ checkHead func(*testing.T, *Snapshot, []iceberg.ManifestFile)
+}{
+ {
+ name: "merge append",
+ props: iceberg.Properties{ManifestMergeEnabledKey: "true",
ManifestMinMergeCountKey: "2"},
+ stage: func(t *testing.T, tx *Transaction) {
+ require.NoError(t, tx.AddDataFiles(t.Context(),
[]iceberg.DataFile{retryCleanupFile(t, "f4")}, nil))
+ },
+ checkHead: func(t *testing.T, head *Snapshot, manifests
[]iceberg.ManifestFile) {
+ require.Len(t, manifests, 1, "every manifest must merge
into one")
+ assert.Equal(t, head.SnapshotID,
manifests[0].SnapshotID())
+ assert.EqualValues(t, 1, manifests[0].AddedDataFiles())
+ },
+ },
+ {
+ name: "overwrite with deletes",
+ props: iceberg.Properties{
+ ManifestMergeEnabledKey: "false",
+ WriteUpdateIsolationLevelKey: string(IsolationSnapshot),
+ },
+ stage: func(t *testing.T, tx *Transaction) {
+ require.NoError(t, tx.ReplaceDataFiles(t.Context(),
+ []iceberg.DataFile{retryCleanupFile(t, "f1")},
[]iceberg.DataFile{retryCleanupFile(t, "f4")}, nil))
+ },
+ checkHead: func(t *testing.T, head *Snapshot, manifests
[]iceberg.ManifestFile) {
+ var tombstones, rewritten int
+ for _, m := range manifests {
+ if m.SnapshotID() != head.SnapshotID {
+ continue
+ }
+ if m.DeletedDataFiles() > 0 {
+ tombstones++
+ }
+ if m.ExistingDataFiles() > 0 {
+ rewritten++
+ }
+ }
+ assert.Equal(t, 1, tombstones, "the removal of f1 must
be written as a tombstone manifest")
+ assert.Equal(t, 1, rewritten, "the manifest holding f1
must be rewritten to keep f2")
+ },
+ },
+}
+
+func TestCommitLeavesOnlyCommittedFiles(t *testing.T) {
+ runs := []struct {
+ name string
+ errs []error
+ }{
+ {name: "first attempt"},
+ {name: "after two conflicts", errs: []error{ErrCommitFailed,
ErrCommitFailed}},
+ }
+ for _, p := range retryCleanupProducers {
+ for _, run := range runs {
+ t.Run(p.name+"/"+run.name, func(t *testing.T) {
+ f := newRetryCleanupFixture(t, "2", p.props)
+ tbl, cat := f.newTable(t, []Metadata{f.h1,
f.h2}, run.errs...)
+ tx := tbl.NewTransaction()
+ p.stage(t, tx)
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ require.Equal(t, len(run.errs)+1, cat.attempts)
+ head, manifests := headManifests(t, committed,
f.wfs)
+ p.checkHead(t, head, manifests)
+ f.assertOnlyCommittedFilesAdded(t, committed)
+ })
+ }
+ }
+}
+
+func TestFailedCommitKeepsOnlyStagedFilesUntilResumeLands(t *testing.T) {
+ for _, p := range retryCleanupProducers {
+ t.Run(p.name, func(t *testing.T) {
+ f := newRetryCleanupFixture(t, "2", p.props)
+ tbl, cat := f.newTable(t, []Metadata{f.h1, f.h2},
ErrCommitFailed, ErrCommitFailed, ErrCommitFailed)
+ tx := tbl.NewTransaction()
+ p.stage(t, tx)
+ staged := f.newFiles()
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
+ require.NotErrorIs(t, err, ErrTransactionUnusable)
+ require.Equal(t, 3, cat.attempts)
+ assert.Equal(t, staged, f.newFiles(), "only the staged
files may remain, and all of them")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ head, manifests := headManifests(t, committed, f.wfs)
+ p.checkHead(t, head, manifests)
+ f.assertOnlyCommittedFilesAdded(t, committed)
+ })
+ }
+}
+
+// Cleanup must not guess which files a landed commit references when it
+// cannot read a committed manifest list.
+func TestCommitCleanupSkippedWhenCommittedListUnreadable(t *testing.T) {
+ for _, p := range retryCleanupProducers {
+ t.Run(p.name, func(t *testing.T) {
+ f := newRetryCleanupFixture(t, "2", p.props)
+ fs := &unreadableListIO{memIO: f.wfs}
+ cat := &landingCatalog{
+ peerHeadsCatalog: &peerHeadsCatalog{
+ current: f.h0,
+ heads: []Metadata{f.h1, f.h2},
+ errs: []error{ErrCommitFailed,
ErrCommitFailed},
+ },
+ landed: func(updates []Update) {
+ for _, u := range updates {
+ if su, ok :=
u.(*addSnapshotUpdate); ok {
+ fs.mu.Lock()
+ fs.unreadable =
su.Snapshot.ManifestList
+ fs.mu.Unlock()
+ }
+ }
+ },
+ }
+ tx := newOCCTable(t, f.h0, fs, cat).NewTransaction()
+ p.stage(t, tx)
+ staged := f.newFiles()
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ require.Equal(t, 3, cat.attempts)
+ require.NotNil(t, committed.CurrentSnapshot())
+ require.NotEmpty(t, fs.unreadable)
+ assert.Empty(t, fs.removed, "nothing may be deleted
without reading what the commit references")
+ after := f.newFiles()
+ for _, path := range staged {
+ assert.Contains(t, after, path, "superseded
staged files must be kept too")
+ }
+ })
+ }
+}
+
+func TestUnknownCommitStateDeletesNoManifests(t *testing.T) {
+ for _, p := range retryCleanupProducers {
+ t.Run(p.name, func(t *testing.T) {
+ f := newRetryCleanupFixture(t, "2", p.props)
+ unknown := errors.New("simulated 5xx: internal server
error")
+ tbl, _ := f.newTable(t, []Metadata{f.h1},
ErrCommitFailed, unknown)
+ tx := tbl.NewTransaction()
+ p.stage(t, tx)
+ before := f.newFiles()
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, unknown)
+
+ after := f.newFiles()
+ for _, path := range before {
+ assert.Contains(t, after, path, "nothing may be
deleted when the commit state is unknown")
+ }
+ var rebuiltManifests int
+ for _, path := range after {
+ if !slices.Contains(before, path) &&
!strings.HasPrefix(filepath.Base(path), "snap-") {
+ rebuiltManifests++
+ }
+ }
+ assert.NotZero(t, rebuiltManifests, "the rebuild must
have written manifests, all of them kept")
+ })
+ }
+}
+
+// A transaction staging a fast append and then a merging append commits both
+// snapshots. The merge consumes the first snapshot's added manifest, which
only
+// the first committed snapshot still references.
+func TestChainedSnapshotsKeepEveryCommittedManifest(t *testing.T) {
+ f := newRetryCleanupFixture(t, "1", iceberg.Properties{
+ ManifestMergeEnabledKey: "true",
+ ManifestMinMergeCountKey: "2",
+ })
+ tbl, cat := f.newTable(t, []Metadata{f.h1}, ErrCommitFailed)
+ tx := tbl.NewTransaction()
+
+ fastAppend := newFastAppendFilesProducer(OpAppend, tx, f.wfs, nil, nil)
+ fastAppend.appendDataFile(retryCleanupFile(t, "f4"))
+ updates, reqs, err := fastAppend.commit(t.Context())
+ require.NoError(t, err)
+ require.NoError(t, tx.apply(updates, reqs))
+ require.NoError(t, tx.AddDataFiles(t.Context(),
[]iceberg.DataFile{retryCleanupFile(t, "f6")}, nil))
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ require.Equal(t, 2, cat.attempts)
+
+ head, manifests := headManifests(t, committed, f.wfs)
+ require.Len(t, manifests, 1, "the merging append must merge every
manifest")
+ require.NotNil(t, head.ParentSnapshotID)
+ first := committed.Metadata().SnapshotByID(*head.ParentSnapshotID)
+ require.NotNil(t, first)
+ firstManifests, err := first.Manifests(f.wfs)
+ require.NoError(t, err)
+ var firstAdded int
+ for _, m := range firstManifests {
+ if m.SnapshotID() == first.SnapshotID {
+ firstAdded++
+ }
+ }
+ require.Equal(t, 1, firstAdded, "the first snapshot must reference its
own added manifest")
+ f.assertOnlyCommittedFilesAdded(t, committed)
+}
diff --git a/table/rewrite_manifests.go b/table/rewrite_manifests.go
index e2e7dade0..3cdb77589 100644
--- a/table/rewrite_manifests.go
+++ b/table/rewrite_manifests.go
@@ -136,11 +136,6 @@ type rewriteManifests struct {
rewritten []iceberg.ManifestFile
added []iceberg.ManifestFile
- // superseded accumulates merged manifest files written by a rebuild on
a
- // prior OCC attempt that a later attempt replaced. doCommit removes
them
- // after a successful commit so retries don't leak orphaned manifests.
- superseded []string
-
// result is the value returned to the caller. record() rewrites its
// fields on every pass, including OCC retries, so the pointer the
caller
// holds reflects the manifests actually committed, not attempt 0's.
@@ -150,17 +145,12 @@ type rewriteManifests struct {
func newRewriteManifestsProducer(txn *Transaction, fs iceio.WriteFileIO, props
iceberg.Properties, cfg rewriteManifestsCfg) *snapshotProducer {
prod := createSnapshotProducer(OpReplace, txn, fs, nil, props)
prod.producerImpl = &rewriteManifests{base: prod, cfg: cfg, result:
&RewriteManifestsResult{}}
+ // A resumed commit can reuse the manifests the last rebuild merged.
+ prod.written.keepRebuilt = true
return prod
}
-// supersededManifests returns the merged manifests earlier passes wrote and a
-// later pass replaced. The committed snapshot references only the last pass's
-// manifests, so these are safe to delete once the commit lands.
-func (r *rewriteManifests) supersededManifests() []string {
- return r.superseded
-}
-
func (r *rewriteManifests) parentDependentManifests(_ context.Context, parent
*Snapshot) ([]iceberg.ManifestFile, error) {
if parent == nil {
return nil, nil
@@ -356,12 +346,6 @@ func (r *rewriteManifests) record(toRewrite, merged, kept
[]iceberg.ManifestFile
// Publish only after the guard passes. On failure the caller deletes
this
// pass's merged files, so the producer must keep the previous pass's
state.
- // The previous pass's merged manifests were never committed. Carry them
- // into the superseded set so a successful commit deletes them, or every
- // retry leaks a full set of merged files.
- for _, m := range r.added {
- r.superseded = append(r.superseded, m.FilePath())
- }
r.added, r.rewritten = added, rewritten
// Publish into the shared result the caller holds so a retry's counts
win.
diff --git a/table/rewrite_manifests_internal_test.go
b/table/rewrite_manifests_internal_test.go
index 4c261ea25..33131f5f5 100644
--- a/table/rewrite_manifests_internal_test.go
+++ b/table/rewrite_manifests_internal_test.go
@@ -419,7 +419,7 @@ func TestRewriteManifestsFailedPassKeepsPreviousState(t
*testing.T) {
_, err := r.processManifests(t.Context(), inputs("good", 1))
require.NoError(t, err)
- added, rewritten, superseded := slices.Clone(r.added),
slices.Clone(r.rewritten), slices.Clone(r.superseded)
+ added, rewritten := slices.Clone(r.added), slices.Clone(r.rewritten)
require.NotEmpty(t, added)
// A header count that lies trips the guard after the merged file is
written.
@@ -428,7 +428,6 @@ func TestRewriteManifestsFailedPassKeepsPreviousState(t
*testing.T) {
assert.Equal(t, added, r.added)
assert.Equal(t, rewritten, r.rewritten)
- assert.Equal(t, superseded, r.superseded)
assert.Equal(t, added, r.result.AddedManifests)
assert.Equal(t, rewritten, r.result.RewrittenManifests)
files := memKeys(mem)
diff --git a/table/snapshot_producers.go b/table/snapshot_producers.go
index d856a6958..56e76e290 100644
--- a/table/snapshot_producers.go
+++ b/table/snapshot_producers.go
@@ -64,12 +64,64 @@ type producerImpl interface {
needsValidation() bool
}
-// supersededAccumulator marks a producer that writes merged manifests OCC
-// retries orphan. Only rewrite implements it.
-type supersededAccumulator interface {
- // supersededManifests returns paths of merged manifests that earlier
- // passes wrote and a later pass replaced, to delete once the commit
lands.
- supersededManifests() []string
+// writtenManifests records the manifests a producer wrote for its staged
+// snapshot and each rebuild. Staged manifests are deleted only after a commit
+// lands, and a failed commit deletes only what its rebuilds wrote.
+type writtenManifests struct {
+ stagedList string
+ staged map[string]struct{}
+ unreferenced []string
+ rebuilt map[string]struct{}
+ keepRebuilt bool
+}
+
+func (w *writtenManifests) stage(list string, referenced, added []string) {
+ w.stagedList = list
+ w.staged = make(map[string]struct{}, len(referenced)+len(added))
+ w.unreferenced = nil
+ for _, path := range referenced {
+ w.staged[path] = struct{}{}
+ }
+ for _, path := range added {
+ if _, ok := w.staged[path]; !ok {
+ w.staged[path] = struct{}{}
+ w.unreferenced = append(w.unreferenced, path)
+ }
+ }
+}
+
+func (w *writtenManifests) rebuild(referenced []string) {
+ for _, path := range referenced {
+ if _, ok := w.staged[path]; ok {
+ continue
+ }
+ if w.rebuilt == nil {
+ w.rebuilt = make(map[string]struct{})
+ }
+ w.rebuilt[path] = struct{}{}
+ }
+}
+
+// uncommitted returns the recorded manifests that the committed snapshot,
+// whose manifest list is committedList, might not reference. The caller must
+// still drop the ones a committed snapshot references.
+func (w *writtenManifests) uncommitted(committedList string) []string {
+ paths := slices.Collect(maps.Keys(w.rebuilt))
+ if committedList == w.stagedList {
+ return append(paths, w.unreferenced...)
+ }
+
+ return append(paths, slices.Collect(maps.Keys(w.staged))...)
+}
+
+func (w *writtenManifests) discardRebuilt() []string {
+ if w.keepRebuilt {
+ return nil
+ }
+ paths := slices.Collect(maps.Keys(w.rebuilt))
+ w.rebuilt = nil
+
+ return paths
}
// retryManifestRebuilder lets a producer reuse parent-independent manifest
@@ -759,6 +811,7 @@ type snapshotProducer struct {
deletedDeleteFiles map[string]iceberg.DataFile
deletedDVsByRef map[string]iceberg.DataFile
snapshotProps iceberg.Properties
+ written writtenManifests
}
// deleteFileAddition carries the data sequence number of a rewritten delete
@@ -915,6 +968,19 @@ func (sp *snapshotProducer) iterManifestEntries(m
iceberg.ManifestFile, discardD
return m.Entries(sp.io, discardDeleted)
}
+// writtenPaths returns the manifests this producer wrote, the ones carrying
its
+// snapshot ID.
+func (sp *snapshotProducer) writtenPaths(manifests []iceberg.ManifestFile)
[]string {
+ paths := make([]string, 0, len(manifests))
+ for _, m := range manifests {
+ if m.SnapshotID() == sp.snapshotID {
+ paths = append(paths, m.FilePath())
+ }
+ }
+
+ return paths
+}
+
// parentSnapshot resolves the snapshot this producer is layered on (the branch
// head captured when the producer was built), or nil for the first snapshot.
func (sp *snapshotProducer) parentSnapshot() (*Snapshot, error) {
@@ -1614,6 +1680,7 @@ func (sp *snapshotProducer) commitManifests(ctx
context.Context, newManifests, a
if err != nil {
return nil, nil, err
}
+ sp.written.stage(manifestListFilePath, sp.writtenPaths(newManifests),
sp.writtenPaths(addedContent))
snapshot := Snapshot{
SnapshotID: sp.snapshotID,
@@ -1699,6 +1766,7 @@ func (sp *snapshotProducer) commitManifests(ctx
context.Context, newManifests, a
if procErr != nil {
return nil, fmt.Errorf("rebuild manifest list: assemble
manifests: %w", procErr)
}
+ sp.written.rebuild(sp.writtenPaths(combined))
// Derive the sequence number from the fresh table-wide
last-sequence-number.
// Using freshParent.SequenceNumber + 1 alone would violate the
spec when a
@@ -1794,11 +1862,7 @@ func (sp *snapshotProducer) commitManifests(ctx
context.Context, newManifests, a
addSnap := NewAddSnapshotUpdate(&snapshot)
addSnap.ownManifests = addedContent
addSnap.rebuildManifestList = rebuildFn
- // Hold the producer so doCommit can read the manifests orphaned by OCC
- // retries and clean them up. Only rewrite producers implement this.
- if acc, ok := sp.producerImpl.(supersededAccumulator); ok {
- addSnap.supersededSource = acc
- }
+ addSnap.written = &sp.written
// Build the assertion from the base table's branch head, not the
// staged metadata's current snapshot: a staged intermediate snapshot
diff --git a/table/table.go b/table/table.go
index db12ced2f..4033709c3 100644
--- a/table/table.go
+++ b/table/table.go
@@ -684,18 +684,37 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
remove := slices.Collect(maps.Keys(rebuiltLists))
if committed {
remove = orphanedManifests
- for _, u := range updates {
- if su, ok := u.(*addSnapshotUpdate); ok &&
su.supersededSource != nil {
- remove = append(remove,
su.supersededSource.supersededManifests()...)
- }
+ }
+ for _, u := range updates {
+ su, ok := u.(*addSnapshotUpdate)
+ if !ok || su.written == nil {
+ continue
}
+ if committed {
+ remove = append(remove,
su.written.uncommitted(su.Snapshot.ManifestList)...)
+ } else {
+ remove = append(remove,
su.written.discardRebuilt()...)
+ }
+ }
+ if committed && len(remove) > 0 {
+ referenced, err := committedFiles(wfs, updates)
+ if err != nil {
+ log.Printf("Warning: skipping cleanup of
uncommitted files: %v", err)
+
+ return
+ }
+ remove = slices.DeleteFunc(remove, func(path string)
bool {
+ _, ok := referenced[path]
+
+ return ok
+ })
}
// The commit may have failed because ctx was canceled; remove
the
// orphans without that cancellation so they are not left
behind.
cleanupCtx := context.WithoutCancel(ctx)
for _, path := range remove {
if removeErr := wfs.Remove(cleanupCtx, path); removeErr
!= nil {
- log.Printf("Warning: failed to delete orphaned
manifest list %s: %v", path, removeErr)
+ log.Printf("Warning: failed to delete
uncommitted file %s: %v", path, removeErr)
}
}
}()
@@ -1034,6 +1053,28 @@ func stagedManifestLists(updates []Update)
map[string]struct{} {
return lists
}
+// committedFiles returns the manifest lists and manifests of every snapshot
+// updates add.
+func committedFiles(fs icebergio.IO, updates []Update) (map[string]struct{},
error) {
+ files := make(map[string]struct{})
+ for _, u := range updates {
+ su, ok := u.(*addSnapshotUpdate)
+ if !ok || su.Snapshot == nil {
+ continue
+ }
+ files[su.Snapshot.ManifestList] = struct{}{}
+ manifests, err := su.Snapshot.Manifests(fs)
+ if err != nil {
+ return nil, fmt.Errorf("read manifests of committed
snapshot %d: %w", su.Snapshot.SnapshotID, err)
+ }
+ for _, m := range manifests {
+ files[m.FilePath()] = struct{}{}
+ }
+ }
+
+ return files, nil
+}
+
func validateBranchRequirement(reqs []Requirement, branch string, meta
Metadata) error {
for _, req := range reqs {
if ref, ok := req.(*assertRefSnapshotID); ok &&
ref.requireBranch && ref.Ref == branch {
@@ -1153,7 +1194,7 @@ func rebuildSnapshotUpdates(ctx context.Context, updates
[]Update, freshMeta Met
Snapshot: newSnap,
ownManifests: su.ownManifests,
rebuildManifestList: su.rebuildManifestList,
- supersededSource: su.supersededSource,
+ written: su.written,
}
// The old manifest list is now an orphaned object in object
storage.
diff --git a/table/updates.go b/table/updates.go
index 4c4783ad8..511f556d5 100644
--- a/table/updates.go
+++ b/table/updates.go
@@ -505,12 +505,8 @@ type addSnapshotUpdate struct {
// concurrent writers since the original build.
rebuildManifestList func(ctx context.Context, freshMeta Metadata,
freshParent *Snapshot, fio io.WriteFileIO, attempt int) (*Snapshot, error)
- // supersededSource, when non-nil, is the producer whose merged
manifests a
- // rebuild on a prior attempt wrote and a later attempt replaced.
doCommit
- // reads its accumulated paths and removes them once the commit
resolves.
- // Distinct from the manifest-LIST orphans rebuildSnapshotUpdates
returns:
- // these are the inner data manifests a rewrite re-merges on every
retry.
- supersededSource supersededAccumulator
+ // written records the manifests the producer wrote for this snapshot.
+ written *writtenManifests
}
// NewAddSnapshotUpdate creates a new update that adds the given snapshot to
the table metadata.
