This is an automated email from the ASF dual-hosted git repository.
laskoviymishka 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 94102c5cb feat(table): replan snapshot expiration on commit retry
(#2354)
94102c5cb is described below
commit 94102c5cb24f437a7c33e6bf53e2acb4efacdcc5
Author: Harrison Crosse <[email protected]>
AuthorDate: Sun Nov 8 16:17:02 2026 -0500
feat(table): replan snapshot expiration on commit retry (#2354)
---
catalog/sql/sql_test.go | 261 ++++++++-----
table/commit.go | 2 +
table/commit_retry_test.go | 7 +-
table/expire_replay.go | 191 +++++++++
table/expire_replay_test.go | 824 +++++++++++++++++++++++++++++++++++++++
table/table.go | 137 ++++---
table/table_test.go | 226 +++++------
table/transaction.go | 134 +++++--
website/src/concurrent-writes.md | 44 ++-
website/src/releases.md | 1 +
10 files changed, 1514 insertions(+), 313 deletions(-)
diff --git a/catalog/sql/sql_test.go b/catalog/sql/sql_test.go
index 2025dc386..f44304536 100644
--- a/catalog/sql/sql_test.go
+++ b/catalog/sql/sql_test.go
@@ -36,6 +36,7 @@ import (
"sync"
"sync/atomic"
"testing"
+ "testing/synctest"
"time"
"github.com/apache/arrow-go/v18/arrow"
@@ -1526,33 +1527,26 @@ func (s *SqliteCatalogTestSuite) appendStringRow(ctx
context.Context, tbl *table
return out
}
-// waitPastSnapshots blocks until the wall clock in milliseconds is past
-// every snapshot timestamp of meta, so an expiry with WithOlderThan(0)
-// treats them all as old enough.
-func (s *SqliteCatalogTestSuite) waitPastSnapshots(meta table.Metadata) {
- var newest int64
- for _, snap := range meta.Snapshots() {
- newest = max(newest, snap.TimestampMs)
- }
- s.Require().Eventually(func() bool { return time.Now().UnixMilli() >
newest },
- time.Second, time.Millisecond)
-}
-
// createRetryingStringTable creates a single-column string table that
// retries conflicting commits numRetries times with millisecond waits,
// then appends one row per value.
func (s *SqliteCatalogTestSuite) createRetryingStringTable(ctx
context.Context, cat *sqlcat.Catalog, numRetries string, values ...string)
table.Identifier {
+ return s.createStringTable(ctx, cat,
iceberg.Properties{table.CommitNumRetriesKey: numRetries}, values...)
+}
+
+func (s *SqliteCatalogTestSuite) createStringTable(ctx context.Context, cat
*sqlcat.Catalog, props iceberg.Properties, values ...string) table.Identifier {
schema := iceberg.NewSchema(1, iceberg.NestedField{
ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.String,
Required: true,
})
tblID := s.randomTableIdentifier()
s.Require().NoError(cat.CreateNamespace(ctx,
catalog.NamespaceFromIdent(tblID), nil))
- tbl, err := cat.CreateTable(ctx, tblID, schema,
catalog.WithProperties(iceberg.Properties{
- table.CommitNumRetriesKey: numRetries,
+ all := iceberg.Properties{
table.CommitMinRetryWaitMsKey: "1",
table.CommitMaxRetryWaitMsKey: "2",
- }))
+ }
+ maps.Copy(all, props)
+ tbl, err := cat.CreateTable(ctx, tblID, schema,
catalog.WithProperties(all))
s.Require().NoError(err)
for _, v := range values {
tbl = s.appendStringRow(ctx, tbl, table.MainBranch, v)
@@ -2033,49 +2027,37 @@ func (s *SqliteCatalogTestSuite)
TestStaleRollbackFailsAfterPeerAppend() {
s.Len(current.Metadata().Snapshots(), 3)
}
-// Snapshot expiry staged before a peer rolled the branch back must not
-// be replayed against the new head: its retention window was computed
-// from the old head, so it would expire snapshots the rolled-back
-// branch still retains.
-func (s *SqliteCatalogTestSuite)
TestStaleExpireSnapshotsFailsAfterPeerRollback() {
- ctx := context.Background()
- cat := s.getCatalogSqlite()
- defer cat.Close()
- tblID := s.createRetryingStringTable(ctx, cat, "2", "s1", "s2", "s3",
"s4")
-
- stale, err := cat.LoadTable(ctx, tblID)
- s.Require().NoError(err)
- snaps := slices.Clone(stale.Metadata().Snapshots())
- s.Require().Len(snaps, 4)
- slices.SortFunc(snaps, func(a, b table.Snapshot) int { return
cmp.Compare(a.SequenceNumber, b.SequenceNumber) })
- ids := make([]int64, len(snaps))
- for i, snap := range snaps {
- ids[i] = snap.SnapshotID
- }
+func (s *SqliteCatalogTestSuite)
TestStaleExpireSnapshotsReplansAfterPeerRollback() {
+ s.inBubble(func() {
+ ctx := context.Background()
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createStringTable(ctx, cat, bubbleTableProps, "s1",
"s2", "s3", "s4")
- peer, err := cat.LoadTable(ctx, tblID)
- s.Require().NoError(err)
- peerTx := peer.NewTransaction()
- s.Require().NoError(peerTx.RollbackToSnapshot(ids[1]))
- _, err = peerTx.Commit(ctx)
- s.Require().NoError(err)
+ stale, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ ids := snapshotIDsBySequence(stale.Metadata())
+ s.Require().Len(ids, 4)
- s.waitPastSnapshots(stale.Metadata())
+ peer, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ peerTx := peer.NewTransaction()
+ s.Require().NoError(peerTx.RollbackToSnapshot(ids[1]))
+ _, err = peerTx.Commit(ctx)
+ s.Require().NoError(err)
+ time.Sleep(time.Millisecond)
- tx := stale.NewTransaction()
- s.Require().NoError(tx.ExpireSnapshots(table.WithRetainLast(3),
table.WithOlderThan(0), table.WithPostCommit(false)))
- _, err = tx.Commit(ctx)
- s.Require().Error(err)
- s.ErrorIs(err, table.ErrCommitFailed)
+ tx := stale.NewTransaction()
+ s.Require().NoError(tx.ExpireSnapshots(table.WithRetainLast(3),
table.WithOlderThan(0), table.WithPostCommit(false)))
+ _, err = tx.Commit(ctx)
+ s.Require().NoError(err)
- current, err := cat.LoadTable(ctx, tblID)
- s.Require().NoError(err)
- s.Equal(ids[1], current.Metadata().CurrentSnapshot().SnapshotID)
- var remaining []int64
- for _, snap := range current.Metadata().Snapshots() {
- remaining = append(remaining, snap.SnapshotID)
- }
- s.ElementsMatch(ids, remaining)
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ s.Equal(ids[1], current.Metadata().CurrentSnapshot().SnapshotID)
+ s.Equal(ids[:2], snapshotIDsBySequence(current.Metadata()))
+ s.ElementsMatch([]string{"s1", "s2"}, s.storedStringValues(ctx,
current))
+ })
}
// A rollback of a branch the transaction staged must not be replayed
@@ -2112,44 +2094,149 @@ func (s *SqliteCatalogTestSuite)
TestStaleRollbackOfStagedBranchFailsAfterPeerCr
s.Len(current.Metadata().Snapshots(), 3)
}
-// Snapshot expiry in a transaction that staged a new branch must not be
-// replayed against a branch a peer created meanwhile: its retention was
-// computed without the peer's branch.
-func (s *SqliteCatalogTestSuite)
TestStaleExpireWithStagedBranchFailsAfterPeerCreatesIt() {
- ctx := context.Background()
- cat := s.getCatalogSqlite()
- defer cat.Close()
- tblID := s.createRetryingStringTable(ctx, cat, "2", "seed", "second")
+func (s *SqliteCatalogTestSuite)
TestStaleExpireWithStagedBranchReplansAfterPeerCreatesIt() {
+ s.inBubble(func() {
+ ctx := context.Background()
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createStringTable(ctx, cat, bubbleTableProps,
"seed", "second")
- stale, err := cat.LoadTable(ctx, tblID)
- s.Require().NoError(err)
+ stale, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ mainHead := stale.Metadata().CurrentSnapshot().SnapshotID
- peer, err := cat.LoadTable(ctx, tblID)
- s.Require().NoError(err)
- peer = s.appendStringRow(ctx, peer, "feature", "peer")
- peerHead := peer.Metadata().SnapshotByName("feature").SnapshotID
- s.waitPastSnapshots(peer.Metadata())
+ peer, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ peer = s.appendStringRow(ctx, peer, "feature", "peer")
+ peerHead := peer.Metadata().SnapshotByName("feature").SnapshotID
+ time.Sleep(time.Millisecond)
- tx, err := stale.NewTransactionOnBranch("feature")
- s.Require().NoError(err)
- s.stageStringRow(ctx, tx, "stale")
- s.Require().NoError(tx.ExpireSnapshots(table.WithRetainLast(1),
table.WithOlderThan(0), table.WithPostCommit(false)))
- _, err = tx.Commit(ctx)
- s.Require().Error(err)
- s.ErrorIs(err, table.ErrCommitFailed)
+ tx, err := stale.NewTransactionOnBranch("feature")
+ s.Require().NoError(err)
+ s.stageStringRow(ctx, tx, "stale")
+ s.Require().NoError(tx.ExpireSnapshots(table.WithRetainLast(1),
table.WithOlderThan(0), table.WithPostCommit(false)))
+ _, err = tx.Commit(ctx)
+ s.Require().NoError(err)
- current, err := cat.LoadTable(ctx, tblID)
- s.Require().NoError(err)
- s.Equal(peerHead,
current.Metadata().SnapshotByName("feature").SnapshotID)
- var remaining []int64
- for _, snap := range current.Metadata().Snapshots() {
- remaining = append(remaining, snap.SnapshotID)
- }
- var expected []int64
- for _, snap := range peer.Metadata().Snapshots() {
- expected = append(expected, snap.SnapshotID)
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ feature := current.Metadata().SnapshotByName("feature")
+ s.Require().NotNil(feature)
+ s.Require().NotNil(feature.ParentSnapshotID)
+ s.Equal(peerHead, *feature.ParentSnapshotID)
+ s.Equal([]int64{mainHead, feature.SnapshotID},
snapshotIDsBySequence(current.Metadata()))
+ s.ElementsMatch([]string{"seed", "second", "peer", "stale"},
+ s.storedStringValues(ctx, current,
table.WithSnapshotID(feature.SnapshotID)))
+ s.ElementsMatch([]string{"seed", "second"},
s.storedStringValues(ctx, current))
+ })
+}
+
+// A millisecond sleep makes earlier snapshots older than the expiry cutoff.
+// Use bubbleTableProps for tables created inside the bubble.
+func (s *SqliteCatalogTestSuite) inBubble(f func()) {
+ outer := s.T()
+ synctest.Test(outer, func(t *testing.T) {
+ s.SetT(t)
+ defer s.SetT(outer)
+ f()
+ })
+}
+
+// bubbleTableProps avoid zstd, whose shared decoder in arrow-go belongs to
+// the first bubble that uses it.
+var bubbleTableProps = iceberg.Properties{
+ table.CommitNumRetriesKey: "2",
+ table.ParquetCompressionKey: "snappy",
+}
+
+func snapshotIDsBySequence(meta table.Metadata) []int64 {
+ snaps := slices.SortedFunc(slices.Values(meta.Snapshots()), func(a, b
table.Snapshot) int {
+ return cmp.Compare(a.SequenceNumber, b.SequenceNumber)
+ })
+ ids := make([]int64, len(snaps))
+ for i, snap := range snaps {
+ ids[i] = snap.SnapshotID
}
- s.ElementsMatch(expected, remaining)
+
+ return ids
+}
+
+func (s *SqliteCatalogTestSuite)
TestStaleAppendAndExpireReplansAfterPeerRollback() {
+ s.inBubble(func() {
+ ctx := context.Background()
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createStringTable(ctx, cat, bubbleTableProps, "s1",
"s2", "s3", "s4")
+ stale, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ ids := snapshotIDsBySequence(stale.Metadata())
+ time.Sleep(time.Millisecond)
+
+ tx := stale.NewTransaction()
+ s.stageStringRow(ctx, tx, "a")
+ s.Require().NoError(tx.ExpireSnapshots(table.WithRetainLast(3),
table.WithOlderThan(0)))
+
+ peer, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ peerTx := peer.NewTransaction()
+ s.Require().NoError(peerTx.RollbackToSnapshot(ids[1]))
+ _, err = peerTx.Commit(ctx)
+ s.Require().NoError(err)
+
+ _, err = tx.Commit(ctx)
+ s.Require().NoError(err)
+
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ got := snapshotIDsBySequence(current.Metadata())
+ s.Require().Len(got, 3)
+ s.Equal(ids[:2], got[:2])
+ head := current.Metadata().CurrentSnapshot()
+ s.Equal(got[2], head.SnapshotID)
+ s.Require().NotNil(head.ParentSnapshotID)
+ s.Equal(ids[1], *head.ParentSnapshotID)
+ s.ElementsMatch([]string{"s1", "s2", "a"},
s.storedStringValues(ctx, current))
+ s.ElementsMatch([]string{"s1", "s2"}, s.storedStringValues(ctx,
current, table.WithSnapshotID(ids[1])))
+ s.ElementsMatch([]string{"s1"}, s.storedStringValues(ctx,
current, table.WithSnapshotID(ids[0])))
+ })
+}
+
+func (s *SqliteCatalogTestSuite)
TestStaleExpireSnapshotsKeepsBranchPeerCreated() {
+ s.inBubble(func() {
+ ctx := context.Background()
+ cat := s.getCatalogSqlite()
+ defer cat.Close()
+ tblID := s.createStringTable(ctx, cat, bubbleTableProps, "s1",
"s2", "s3")
+ stale, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ ids := snapshotIDsBySequence(stale.Metadata())
+ time.Sleep(time.Millisecond)
+
+ tx := stale.NewTransaction()
+ s.Require().NoError(tx.ExpireSnapshots(table.WithRetainLast(1),
table.WithOlderThan(0)))
+
+ // Creating an unread ref causes no conflict, so append to
force a retry.
+ _, _, err = cat.CommitTable(ctx, tblID, nil, []table.Update{
+ table.NewSetSnapshotRefUpdate("keep", ids[0],
table.BranchRef, 0, 0, 0),
+ })
+ s.Require().NoError(err)
+ peer, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ peer = s.appendStringRow(ctx, peer, table.MainBranch, "p")
+ p := peer.Metadata().CurrentSnapshot().SnapshotID
+
+ _, err = tx.Commit(ctx)
+ s.Require().NoError(err)
+
+ current, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err)
+ s.Equal([]int64{ids[0], p},
snapshotIDsBySequence(current.Metadata()))
+ keep := current.Metadata().SnapshotByName("keep")
+ s.Require().NotNil(keep)
+ s.Equal(ids[0], keep.SnapshotID)
+ s.ElementsMatch([]string{"s1", "s2", "s3", "p"},
s.storedStringValues(ctx, current))
+ s.ElementsMatch([]string{"s1"}, s.storedStringValues(ctx,
current, table.WithSnapshotID(ids[0])))
+ })
}
func (s *SqliteCatalogTestSuite) TestPurgeTable() {
diff --git a/table/commit.go b/table/commit.go
index 516c6e218..cedaa059d 100644
--- a/table/commit.go
+++ b/table/commit.go
@@ -18,6 +18,7 @@
package table
import (
+ "cmp"
"errors"
"slices"
)
@@ -87,6 +88,7 @@ func (t *Transaction) TableCommit() (TableCommit, error) {
// their updates stage no ref requirement. The helper returns a fresh
// slice, so appending here cannot touch t.reqs.
reqs := append(transactionRequirements(t.reqs, t.branch,
t.tbl.metadata), AssertTableUUID(meta.uuid))
+ reqs = append(reqs, expirationAssertions(t.tbl.metadata,
cmp.Or(t.branch, MainBranch), stagedWalkedRefs(t.expirations), reqs)...)
updates := make([]Update, len(meta.updates))
copy(updates, meta.updates)
diff --git a/table/commit_retry_test.go b/table/commit_retry_test.go
index 3501e0d3f..c195576a0 100644
--- a/table/commit_retry_test.go
+++ b/table/commit_retry_test.go
@@ -704,9 +704,7 @@ func TestDoCommit_OrphanCleanedOnCommitDiverged(t
*testing.T) {
assertOnlyManifestList(t, wfs, originalManifestList)
}
-// assertOnlyManifestList checks that want is the only manifest list left in
-// wfs: a failed commit keeps the staged list and removes every rebuilt one.
-func assertOnlyManifestList(t *testing.T, wfs *memIO, want string) {
+func assertOnlyManifestList(t *testing.T, wfs *memIO, want ...string) {
t.Helper()
var lists []string
for path := range wfs.files {
@@ -714,7 +712,8 @@ func assertOnlyManifestList(t *testing.T, wfs *memIO, want
string) {
lists = append(lists, path)
}
}
- assert.Equal(t, []string{want}, lists)
+ slices.Sort(lists)
+ assert.Equal(t, slices.Sorted(slices.Values(want)), lists)
}
// TestDoCommit_OrphanCleanedOnRetriesExhausted verifies that when every retry
diff --git a/table/expire_replay.go b/table/expire_replay.go
new file mode 100644
index 000000000..b844e3188
--- /dev/null
+++ b/table/expire_replay.go
@@ -0,0 +1,191 @@
+// 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"
+ "fmt"
+ "slices"
+
+ "github.com/apache/iceberg-go/io"
+)
+
+type stagedReplay struct {
+ branch string
+ records []expirationRecord
+
+ // Record positions stay fixed even when replanning changes the update
count.
+ // Snapshots use their most recent rebuild.
+ updates []Update
+
+ // Refs walked by expiries that produced updates.
+ walkedRefs []string
+
+ // Updates from the last successful replay.
+ // ran distinguishes an empty replay from no replay.
+ landed []Update
+ ran bool
+}
+
+func newStagedReplay(branch string, updates []Update, records
[]expirationRecord) *stagedReplay {
+ return &stagedReplay{
+ branch: branch,
+ records: records,
+ updates: slices.Clone(updates),
+ walkedRefs: stagedWalkedRefs(records),
+ }
+}
+
+// On error, replay returns partial updates and replaced lists for cleanup.
+// The partial updates include a rebuilt snapshot that failed to apply.
+func (r *stagedReplay) replay(ctx context.Context, fresh Metadata, fs
io.WriteFileIO, attempt int, force bool) ([]Update, []string, error) {
+ b, err := MetadataBuilderFromBase(fresh, "")
+ if err != nil {
+ return nil, nil, err
+ }
+
+ var (
+ out []Update
+ orphaned []string
+ walkedRefs []string
+ records = r.records
+ chainRebuilt = force
+ )
+ apply := func(updates ...Update) error {
+ for _, u := range updates {
+ if err := u.Apply(b); err != nil {
+ return fmt.Errorf("apply %s: %w", u.Action(),
err)
+ }
+ out = append(out, u)
+ }
+
+ return nil
+ }
+
+ for i := 0; i <= len(r.updates); {
+ if len(records) > 0 && records[0].pos == i {
+ rec := records[0]
+ records = records[1:]
+ updates, walked, err := r.replan(b, rec)
+ if err != nil {
+ return out, orphaned, err
+ }
+ if err := apply(updates...); err != nil {
+ return out, orphaned, err
+ }
+ if len(updates) > 0 {
+ walkedRefs = append(walkedRefs, walked...)
+ }
+ i += rec.count
+
+ continue
+ }
+ if i == len(r.updates) {
+ break
+ }
+
+ if su, ok := r.updates[i].(*addSnapshotUpdate); ok &&
su.rebuildManifestList != nil {
+ // Sequence numbers and row IDs must include earlier
replayed updates.
+ state, err := b.clone().Build()
+ if err != nil {
+ return out, orphaned, err
+ }
+ next, err := rebuildStagedSnapshot(ctx, su, state,
b.currentSnapshotForRef(r.branch), fs, attempt, chainRebuilt)
+ if err != nil {
+ return out, orphaned, err
+ }
+ if next != su {
+ r.updates[i] = next
+ orphaned = append(orphaned,
su.Snapshot.ManifestList)
+ chainRebuilt = true
+ }
+ }
+ if err := apply(r.updates[i]); err != nil {
+ return append(out, r.updates[i]), orphaned,
fmt.Errorf("%w: staged update does not apply after refresh: %w",
ErrTransactionUnusable, err)
+ }
+ i++
+ }
+ r.walkedRefs = walkedRefs
+ r.landed, r.ran = out, true
+
+ return out, orphaned, nil
+}
+
+func (r *stagedReplay) replan(b *MetadataBuilder, rec expirationRecord)
([]Update, []string, error) {
+ updates, walked, err := planExpiration(b, rec.cfg, rec.nowMs)
+ if err != nil {
+ return nil, nil, err
+ }
+ // A no-op expiry must not block other updates solely because GC is
disabled.
+ if len(updates) > 0 {
+ if err := validateGCEnabledForSnapshotExpiration(b.props); err
!= nil {
+ return nil, nil, fmt.Errorf("%w: %w",
ErrTransactionUnusable, err)
+ }
+ }
+
+ // Later updates depend on whether the staged plan removed the branch.
+ if removes := removesRef(updates, r.branch); removes !=
rec.removedBranch {
+ replanned, staged := "removes", "kept"
+ if !removes {
+ replanned, staged = "keeps", "removed"
+ }
+
+ return nil, nil, fmt.Errorf("%w: %w: replanned snapshot
expiration %s commit branch %q, which the staged plan %s",
+ ErrTransactionUnusable, ErrCommitFailed, replanned,
r.branch, staged)
+ }
+
+ return updates, walked, nil
+}
+
+func stagedWalkedRefs(records []expirationRecord) []string {
+ var refs []string
+ for _, rec := range records {
+ if rec.count > 0 {
+ refs = append(refs, rec.walked...)
+ }
+ }
+
+ return refs
+}
+
+// Use base ref IDs, not staged IDs. The commit branch already has an assertion
+// that the transaction rebases.
+func expirationAssertions(base Metadata, branch string, walkedRefs []string,
reqs []Requirement) []Requirement {
+ asserted := map[string]bool{branch: true}
+ for _, r := range reqs {
+ if a, ok := r.(*assertRefSnapshotID); ok {
+ asserted[a.Ref] = true
+ }
+ }
+
+ var names []string
+ for _, name := range walkedRefs {
+ if !asserted[name] {
+ asserted[name] = true
+ names = append(names, name)
+ }
+ }
+ slices.Sort(names)
+
+ out := make([]Requirement, len(names))
+ for i, name := range names {
+ out[i] = AssertRefSnapshotID(name, refSnapshotID(base, name))
+ }
+
+ return out
+}
diff --git a/table/expire_replay_test.go b/table/expire_replay_test.go
new file mode 100644
index 000000000..fd4fd7c03
--- /dev/null
+++ b/table/expire_replay_test.go
@@ -0,0 +1,824 @@
+// 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"
+ "maps"
+ "path"
+ "slices"
+ "strings"
+ "testing"
+ "testing/synctest"
+ "time"
+
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func newExpireReplayTable(t *testing.T, props iceberg.Properties) (*memIO,
*reqValidatingCatalog, *Table) {
+ t.Helper()
+ spec := iceberg.NewPartitionSpec()
+ all := iceberg.Properties{
+ PropertyFormatVersion: "2",
+ CommitNumRetriesKey: "3",
+ CommitMinRetryWaitMsKey: "1",
+ CommitMaxRetryWaitMsKey: "2",
+ CommitTotalRetryTimeoutMsKey: "60000",
+ }
+ maps.Copy(all, props)
+ meta, err := NewMetadata(simpleSchema(), &spec, UnsortedSortOrder,
retryCleanupLocation, all)
+ require.NoError(t, err)
+ fs := newMemIO(1<<20, nil)
+ cat := &reqValidatingCatalog{metadata: meta}
+
+ return fs, cat, newOCCTable(t, meta, fs, cat)
+}
+
+func stageFiles(t *testing.T, tx *Transaction, names ...string) {
+ t.Helper()
+ files := make([]iceberg.DataFile, 0, len(names))
+ for _, name := range names {
+ files = append(files, retryCleanupFile(t, name))
+ }
+ require.NoError(t, tx.AddDataFiles(t.Context(), files, nil))
+}
+
+func appendOnBranch(t *testing.T, tbl *Table, branch string, names ...string)
(*Table, int64) {
+ t.Helper()
+ tx, err := tbl.NewTransactionOnBranch(branch)
+ require.NoError(t, err)
+ stageFiles(t, tx, names...)
+ out, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ return out, out.Metadata().SnapshotByName(branch).SnapshotID
+}
+
+func peerAppend(t *testing.T, fs *memIO, cat *reqValidatingCatalog, branch
string, names ...string) int64 {
+ t.Helper()
+ _, id := appendOnBranch(t, newOCCTable(t, cat.metadata, fs, cat),
branch, names...)
+
+ return id
+}
+
+// peerUpdate mutates cat.metadata without counting a commit attempt.
+func peerUpdate(t *testing.T, cat *reqValidatingCatalog, updates ...Update) {
+ t.Helper()
+ meta, err := UpdateTableMetadata(cat.metadata, updates, "")
+ require.NoError(t, err)
+ cat.metadata = meta
+}
+
+func stagedHead(t *testing.T, tx *Transaction, branch string) int64 {
+ t.Helper()
+ staged, err := tx.StagedTable()
+ require.NoError(t, err)
+
+ return staged.Metadata().SnapshotByName(branch).SnapshotID
+}
+
+func survivors(meta Metadata) []int64 {
+ return slices.Sorted(slices.Values(snapshotIDs(meta.Snapshots())))
+}
+
+func sortedIDs(ids ...int64) []int64 {
+ return slices.Sorted(slices.Values(ids))
+}
+
+func refHead(t *testing.T, meta Metadata, ref string) int64 {
+ t.Helper()
+ snap := meta.SnapshotByName(ref)
+ require.NotNil(t, snap, "ref %s", ref)
+
+ return snap.SnapshotID
+}
+
+func parentOf(t *testing.T, meta Metadata, id int64) int64 {
+ t.Helper()
+ snap := meta.SnapshotByID(id)
+ require.NotNil(t, snap)
+ require.NotNil(t, snap.ParentSnapshotID)
+
+ return *snap.ParentSnapshotID
+}
+
+func branchFiles(t *testing.T, fs *memIO, meta Metadata, branch string)
[]string {
+ t.Helper()
+ head := meta.SnapshotByName(branch)
+ require.NotNil(t, head)
+ manifests, err := head.Manifests(t.Context(), fs)
+ require.NoError(t, err)
+
+ var names []string
+ for _, m := range manifests {
+ for entry, err := range m.Entries(t.Context(), fs, true) {
+ require.NoError(t, err)
+ names = append(names,
strings.TrimSuffix(path.Base(entry.DataFile().FilePath()), ".parquet"))
+ }
+ }
+ slices.Sort(names)
+
+ return names
+}
+
+func TestExpireReplayPlansWhereExpiryWasStaged(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t, nil)
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "b1")
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "b2")
+ tbl, b3 := appendOnBranch(t, tbl, MainBranch, "b3")
+ time.Sleep(time.Millisecond)
+
+ tx := tbl.NewTransaction()
+ stageFiles(t, tx, "a1")
+ a1 := stagedHead(t, tx, MainBranch)
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(3)))
+ stageFiles(t, tx, "a2")
+ a2 := stagedHead(t, tx, MainBranch)
+
+ p := peerAppend(t, fs, cat, MainBranch, "p")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ assert.Equal(t, sortedIDs(b3, p, a1, a2), survivors(meta))
+ assert.Equal(t, a2, refHead(t, meta, MainBranch))
+ assert.Equal(t, a1, parentOf(t, meta, a2))
+ assert.Equal(t, p, parentOf(t, meta, a1))
+ assert.Equal(t, []string{"a1", "a2", "b1", "b2", "b3", "p"},
branchFiles(t, fs, meta, MainBranch))
+ })
+}
+
+func TestExpireReplayAfterRollback(t *testing.T) {
+ stage := func(t *testing.T) (*memIO, *reqValidatingCatalog,
*Transaction, [3]int64, int64) {
+ t.Helper()
+ fs, cat, tbl := newExpireReplayTable(t, nil)
+ var base [3]int64
+ for i, name := range []string{"b1", "b2", "b3"} {
+ tbl, base[i] = appendOnBranch(t, tbl, MainBranch, name)
+ }
+ time.Sleep(time.Millisecond)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.RollbackToSnapshot(base[1]))
+ stageFiles(t, tx, "a")
+ a := stagedHead(t, tx, MainBranch)
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(2)))
+
+ return fs, cat, tx, base, a
+ }
+
+ t.Run("peer commits to another branch", func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tx, base, a := stage(t)
+ // The peer's sequence number forces a rebuild of a.
+ p := peerAppend(t, fs, cat, "other", "p")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ assert.Equal(t, sortedIDs(base[1], base[2], p, a),
survivors(meta))
+ assert.Equal(t, a, refHead(t, meta, MainBranch))
+ assert.Equal(t, base[1], parentOf(t, meta, a))
+ assert.Equal(t, []string{"a", "b1", "b2"},
branchFiles(t, fs, meta, MainBranch))
+ assert.Equal(t, p, refHead(t, meta, "other"))
+ })
+ })
+
+ t.Run("peer moves the rolled back branch", func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tx, _, _ := stage(t)
+ p := peerAppend(t, fs, cat, MainBranch, "p")
+ before := cat.metadata
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrTransactionUnusable)
+ require.ErrorIs(t, err, ErrCommitFailed)
+ assert.Contains(t, err.Error(), "requirement no longer
holds after refresh")
+ assert.Same(t, before, cat.metadata)
+ assert.Equal(t, p, refHead(t, cat.metadata, MainBranch))
+ })
+ })
+}
+
+func TestExpireReplayPlansEachExpiration(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t, nil)
+ for _, name := range []string{"b1", "b2", "b3", "b4"} {
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, name)
+ }
+ time.Sleep(time.Millisecond)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(3)))
+ stageFiles(t, tx, "a")
+ a := stagedHead(t, tx, MainBranch)
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(2)))
+
+ p := peerAppend(t, fs, cat, MainBranch, "p")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ assert.Equal(t, sortedIDs(p, a), survivors(meta))
+ assert.Equal(t, p, parentOf(t, meta, a))
+ assert.Equal(t, []string{"a", "b1", "b2", "b3", "b4", "p"},
branchFiles(t, fs, meta, MainBranch))
+ })
+}
+
+func TestExpireReplayOnNonMainBranch(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t, nil)
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "b1")
+ tbl, b2 := appendOnBranch(t, tbl, MainBranch, "b2")
+ tbl, d1 := appendOnBranch(t, tbl, "dev", "d1")
+ time.Sleep(time.Millisecond)
+
+ tx, err := tbl.NewTransactionOnBranch("dev")
+ require.NoError(t, err)
+ stageFiles(t, tx, "x")
+ x := stagedHead(t, tx, "dev")
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(1)))
+
+ p := peerAppend(t, fs, cat, "dev", "p")
+ peerUpdate(t, cat, NewSetSnapshotRefUpdate("hold", d1, TagRef,
0, 0, 0))
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ assert.Equal(t, sortedIDs(b2, d1, p, x), survivors(meta))
+ assert.Equal(t, x, refHead(t, meta, "dev"))
+ assert.Equal(t, p, parentOf(t, meta, x))
+ assert.Equal(t, d1, refHead(t, meta, "hold"))
+ assert.Equal(t, []string{"b1", "b2", "d1", "p", "x"},
branchFiles(t, fs, meta, "dev"))
+ assert.Equal(t, b2, refHead(t, meta, MainBranch))
+
+ assert.Contains(t, cat.lastReqs, AssertRefSnapshotID("hold",
&d1))
+ assert.Contains(t, cat.lastReqs,
AssertRefSnapshotID(MainBranch, &b2))
+ assert.Contains(t, cat.lastReqs,
assertBranchRefSnapshotID("dev", &p))
+ })
+}
+
+func TestExpireReplayKeepsOriginalCutoff(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t, nil)
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "b1")
+ time.Sleep(2 * time.Hour)
+ tbl, b2 := appendOnBranch(t, tbl, MainBranch, "b2")
+ tbl, b3 := appendOnBranch(t, tbl, MainBranch, "b3")
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(time.Hour),
WithRetainLast(1)))
+
+ time.Sleep(2 * time.Hour)
+ p := peerAppend(t, fs, cat, MainBranch, "p")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ // From the retry's clock, b2 and b3 would be two hours old.
+ assert.Equal(t, sortedIDs(b2, b3, p),
survivors(committed.Metadata()))
+ })
+}
+
+func TestExpireReplayFailsWhenCommitBranchRemovalChanges(t *testing.T) {
+ props := iceberg.Properties{MaxRefAgeMsKey: "3600000"}
+
+ t.Run("staged plan removed the branch", func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t, props)
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "b1")
+ tbl, _ = appendOnBranch(t, tbl, "dev", "d1")
+ time.Sleep(2 * time.Hour)
+
+ tx, err := tbl.NewTransactionOnBranch("dev")
+ require.NoError(t, err)
+ require.NoError(t, tx.ExpireSnapshots())
+
+ p := peerAppend(t, fs, cat, "dev", "p")
+ before := cat.metadata
+
+ _, err = tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrTransactionUnusable)
+ require.ErrorIs(t, err, ErrCommitFailed)
+ assert.EqualError(t, err, `transaction cannot be
committed again: commit failed, refresh and try again: `+
+ `replanned snapshot expiration keeps commit
branch "dev", which the staged plan removed`)
+ assert.Same(t, before, cat.metadata)
+ assert.Equal(t, p, refHead(t, cat.metadata, "dev"))
+ })
+ })
+
+ t.Run("staged plan kept the branch", func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ _, cat, tbl := newExpireReplayTable(t, props)
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "b1")
+ tbl, d1 := appendOnBranch(t, tbl, "dev", "d1")
+ time.Sleep(2 * time.Hour)
+ tbl, _ = appendOnBranch(t, tbl, "dev", "d2")
+
+ tx, err := tbl.NewTransactionOnBranch("dev")
+ require.NoError(t, err)
+ require.NoError(t, tx.ExpireSnapshots())
+ require.NoError(t,
tx.SetProperties(iceberg.Properties{"k": "v"}))
+
+ peerUpdate(t, cat, NewSetSnapshotRefUpdate("dev", d1,
BranchRef, 0, 0, 0))
+ before := cat.metadata
+
+ _, err = tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrTransactionUnusable)
+ require.ErrorIs(t, err, ErrCommitFailed)
+ assert.EqualError(t, err, `transaction cannot be
committed again: commit failed, refresh and try again: `+
+ `replanned snapshot expiration removes commit
branch "dev", which the staged plan kept`)
+ assert.Same(t, before, cat.metadata)
+ })
+ })
+}
+
+func TestExpireReplayPlansNoOpExpiryAgain(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t, nil)
+ tbl, s1 := appendOnBranch(t, tbl, MainBranch, "s1")
+ appendOnBranch(t, tbl, MainBranch, "s2")
+ peerUpdate(t, cat, NewSetSnapshotRefUpdate("keep", s1,
BranchRef, 0, 0, 0))
+ tbl = newOCCTable(t, cat.metadata, fs, cat)
+ time.Sleep(time.Millisecond)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(1)))
+ staged, err := tx.StagedTable()
+ require.NoError(t, err)
+ require.Len(t, staged.Metadata().Snapshots(), 2, "keep protects
s1, so the expiry stages nothing")
+ stageFiles(t, tx, "a")
+ a := stagedHead(t, tx, MainBranch)
+
+ peerUpdate(t, cat, NewRemoveSnapshotRefUpdate("keep"))
+ p := peerAppend(t, fs, cat, MainBranch, "p")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ assert.Equal(t, sortedIDs(p, a), survivors(meta))
+ assert.Equal(t, []string{"a", "p", "s1", "s2"}, branchFiles(t,
fs, meta, MainBranch))
+ })
+}
+
+type recordingCatalog struct {
+ *reqValidatingCatalog
+ updates [][]Update
+ loads int
+
+ before func(attempt int)
+}
+
+func (c *recordingCatalog) CommitTable(ctx context.Context, ident Identifier,
reqs []Requirement, updates []Update) (Metadata, string, error) {
+ if c.before != nil {
+ c.before(len(c.updates))
+ }
+ c.updates = append(c.updates, updates)
+
+ return c.reqValidatingCatalog.CommitTable(ctx, ident, reqs, updates)
+}
+
+func (c *recordingCatalog) LoadTable(ctx context.Context, ident Identifier)
(*Table, error) {
+ c.loads++
+
+ return c.reqValidatingCatalog.LoadTable(ctx, ident)
+}
+
+func removedSnapshots(updates []Update) []int64 {
+ var ids []int64
+ for _, u := range updates {
+ if r, ok := u.(*removeSnapshotsUpdate); ok {
+ ids = append(ids, r.SnapshotIDs...)
+ }
+ }
+
+ return ids
+}
+
+func TestExpireReplayKeepsExpiryThatReplannedToNothing(t *testing.T) {
+ stage := func(t *testing.T, numRetries string) (*memIO,
*recordingCatalog, *Transaction, int64, int64) {
+ t.Helper()
+ fs, inner, tbl := newExpireReplayTable(t,
iceberg.Properties{CommitNumRetriesKey: numRetries})
+ tbl, s1 := appendOnBranch(t, tbl, MainBranch, "s1")
+ tbl, s2 := appendOnBranch(t, tbl, MainBranch, "s2")
+ time.Sleep(time.Millisecond)
+
+ cat := &recordingCatalog{reqValidatingCatalog: inner}
+ tx := newOCCTable(t, tbl.Metadata(), fs, cat).NewTransaction()
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(1)))
+ require.NoError(t, tx.SetProperties(iceberg.Properties{"k":
"v"}))
+ inner.failFirst = inner.attempts.Load() + 2
+
+ return fs, cat, tx, s1, s2
+ }
+
+ t.Run("retry", func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ _, cat, tx, s1, s2 := stage(t, "2")
+ cat.before = func(attempt int) {
+ switch attempt {
+ case 0:
+ peerUpdate(t, cat.reqValidatingCatalog,
NewSetSnapshotRefUpdate("hold", s1, TagRef, 0, 0, 0))
+ case 1:
+ peerUpdate(t, cat.reqValidatingCatalog,
NewRemoveSnapshotRefUpdate("hold"))
+ }
+ }
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ require.Len(t, cat.updates, 3)
+ assert.Equal(t, []int64{s1},
removedSnapshots(cat.updates[0]))
+ assert.Empty(t, removedSnapshots(cat.updates[1]))
+ assert.Equal(t, []int64{s1},
removedSnapshots(cat.updates[2]))
+ assert.Equal(t, []int64{s2},
survivors(committed.Metadata()))
+ assert.Equal(t, "v", committed.Properties()["k"])
+ })
+ })
+
+ t.Run("resume", func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tx, s1, s2 := stage(t, "0")
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
+
+ peerUpdate(t, cat.reqValidatingCatalog,
NewSetSnapshotRefUpdate("hold", s1, TagRef, 0, 0, 0))
+ _, err = tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrCommitFailed)
+ require.NotErrorIs(t, err, ErrTransactionUnusable)
+
+ peerUpdate(t, cat.reqValidatingCatalog,
NewRemoveSnapshotRefUpdate("hold"))
+ p := peerAppend(t, fs, cat.reqValidatingCatalog,
MainBranch, "p")
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ require.Len(t, cat.updates, 3)
+ assert.Equal(t, []int64{s1},
removedSnapshots(cat.updates[0]))
+ assert.Empty(t, removedSnapshots(cat.updates[1]))
+ assert.ElementsMatch(t, []int64{s1, s2},
removedSnapshots(cat.updates[2]))
+ assert.Equal(t, []int64{p},
survivors(committed.Metadata()))
+ assert.Equal(t, "v", committed.Properties()["k"])
+ })
+ })
+}
+
+func TestExpireReplayToNothingCompletesWithoutCommit(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t, nil)
+ tbl, s1 := appendOnBranch(t, tbl, MainBranch, "s1")
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "s2")
+ time.Sleep(time.Millisecond)
+ list := tbl.Metadata().SnapshotByID(s1).ManifestList
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(1)))
+
+ // Inject a catalog conflict because no ref moved.
+ peerUpdate(t, cat, NewRemoveSnapshotsUpdate([]int64{s1}, false))
+ attempts := cat.attempts.Load()
+ cat.failFirst = attempts + 1
+ peerMeta := cat.metadata
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ assert.Equal(t, attempts+1, cat.attempts.Load(), "the replay
must not reach the catalog")
+ assert.Same(t, peerMeta, cat.metadata)
+ assert.Same(t, peerMeta, committed.Metadata())
+ assert.Contains(t, fs.files, list, "the peer's expiry owns the
cleanup")
+
+ _, err = tx.Commit(t.Context())
+ assert.EqualError(t, err, "transaction has already been
committed")
+ })
+}
+
+func TestExpirationAssertionsCoverPlansWithUpdates(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ _, cat, tbl := newExpireReplayTable(t, nil)
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "s0")
+ tbl, s1 := appendOnBranch(t, tbl, MainBranch, "s1")
+ tbl, s2 := appendOnBranch(t, tbl, MainBranch, "s2")
+ peerUpdate(t, cat,
+ NewSetSnapshotRefUpdate("t", s2, TagRef, 0, 0, 0),
+ NewSetSnapshotRefUpdate("b", s1, BranchRef, 0, 0, 0))
+ base := cat.metadata
+ time.Sleep(time.Millisecond)
+
+ noOp := New(tbl.Identifier(), base, "", nil,
cat).NewTransaction()
+ require.NoError(t,
noOp.ExpireSnapshots(WithOlderThan(time.Hour)))
+ require.NoError(t, noOp.SetProperties(iceberg.Properties{"k":
"v"}))
+ commit, err := noOp.TableCommit()
+ require.NoError(t, err)
+ assert.Equal(t, []Requirement{
+ assertBranchRefSnapshotID(MainBranch, &s2),
+ AssertTableUUID(base.TableUUID()),
+ }, commit.Requirements)
+
+ tx := New(tbl.Identifier(), base, "", nil, cat).NewTransaction()
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(time.Hour)))
+ require.NoError(t, tx.AssertRefSnapshotID("b"))
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(1)))
+ commit, err = tx.TableCommit()
+ require.NoError(t, err)
+ assert.Equal(t, []Requirement{
+ AssertRefSnapshotID("b", &s1),
+ assertBranchRefSnapshotID(MainBranch, &s2),
+ AssertTableUUID(base.TableUUID()),
+ AssertRefSnapshotID("t", &s2),
+ }, commit.Requirements)
+ })
+}
+
+func TestExpirationAssertionsSkipCreatedCommitBranch(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t, nil)
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "s0")
+ _, s1 := appendOnBranch(t, tbl, MainBranch, "s1")
+ base := cat.metadata
+ time.Sleep(time.Millisecond)
+
+ tx, err := newOCCTable(t, base, fs,
cat).NewTransactionOnBranch("audit")
+ require.NoError(t, err)
+ stageFiles(t, tx, "a")
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(1)))
+ require.Contains(t, stagedWalkedRefs(tx.expirations), "audit",
"the expiry must stage updates and walk the new branch")
+
+ commit, err := tx.TableCommit()
+ require.NoError(t, err)
+ assert.Equal(t, []Requirement{
+ assertBranchRefSnapshotID("audit", nil),
+ AssertTableUUID(base.TableUUID()),
+ AssertRefSnapshotID(MainBranch, &s1),
+ }, commit.Requirements)
+ })
+}
+
+func TestExpireReplayAssertsWalkedRefFromAttemptBase(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, inner, tbl := newExpireReplayTable(t, nil)
+ tbl, s1 := appendOnBranch(t, tbl, MainBranch, "s1")
+ tbl, s2 := appendOnBranch(t, tbl, MainBranch, "s2")
+ tbl, s3 := appendOnBranch(t, tbl, MainBranch, "s3")
+ _, s4 := appendOnBranch(t, tbl, MainBranch, "s4")
+ peerUpdate(t, inner, NewSetSnapshotRefUpdate("t", s3, TagRef,
0, 0, 0))
+ time.Sleep(time.Millisecond)
+
+ cat := &recordingCatalog{reqValidatingCatalog: inner}
+ tx := newOCCTable(t, inner.metadata, fs, cat).NewTransaction()
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(1)))
+ staged, err := tx.StagedTable()
+ require.NoError(t, err)
+ require.Equal(t, sortedIDs(s3, s4),
survivors(staged.Metadata()))
+
+ // The staged plan keeps s3 for tag t. A peer moves t to s2,
which that
+ // plan expires, and moves it again after the first refresh.
+ peerUpdate(t, inner, NewSetSnapshotRefUpdate("t", s2, TagRef,
0, 0, 0))
+ cat.before = func(attempt int) {
+ if attempt == 1 {
+ peerUpdate(t, inner,
NewSetSnapshotRefUpdate("t", s1, TagRef, 0, 0, 0))
+ }
+ }
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ assert.Equal(t, s1, refHead(t, meta, "t"))
+ assert.Equal(t, sortedIDs(s1, s4), survivors(meta))
+ assert.Len(t, cat.updates, 3)
+ assert.Contains(t, cat.lastReqs, AssertRefSnapshotID("t", &s1))
+ })
+}
+
+func TestExpireReplayKeepsAgedRefAfterPeerAdvancesIt(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t,
iceberg.Properties{MaxRefAgeMsKey: "3600000"})
+ tbl, b1 := appendOnBranch(t, tbl, MainBranch, "b1")
+ tbl, d1 := appendOnBranch(t, tbl, "dev", "d1")
+ time.Sleep(2 * time.Hour)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.ExpireSnapshots())
+ staged, err := tx.StagedTable()
+ require.NoError(t, err)
+ require.Nil(t, staged.Metadata().SnapshotByName("dev"), "the
staged plan removes the aged ref")
+ require.NoError(t, tx.SetProperties(iceberg.Properties{"k":
"v"}))
+
+ p := peerAppend(t, fs, cat, "dev", "p")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ assert.Equal(t, p, refHead(t, meta, "dev"))
+ assert.Equal(t, d1, parentOf(t, meta, p))
+ assert.Equal(t, b1, refHead(t, meta, MainBranch))
+ assert.Equal(t, "v", committed.Properties()["k"])
+ })
+}
+
+func TestExpireReplayFailsWhenPeerDisablesGC(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ _, cat, tbl := newExpireReplayTable(t, nil)
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "s1")
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "s2")
+ time.Sleep(time.Millisecond)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(1)))
+ // Inject a conflict because no requirement checks this
property.
+ peerUpdate(t, cat,
NewSetPropertiesUpdate(iceberg.Properties{GCEnabledKey: "false"}))
+ cat.failFirst = cat.attempts.Load() + 1
+ before := cat.metadata
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrTransactionUnusable)
+ require.NotErrorIs(t, err, ErrCommitFailed)
+ assert.ErrorContains(t, err, "GC is disabled")
+ assert.Same(t, before, cat.metadata)
+ })
+}
+
+func TestExpireReplayNoOpExpiryIgnoresDisabledGC(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t, nil)
+ tbl, s1 := appendOnBranch(t, tbl, MainBranch, "s1")
+ time.Sleep(time.Millisecond)
+
+ tx := tbl.NewTransaction()
+ stageFiles(t, tx, "a")
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(10)))
+ a := stagedHead(t, tx, MainBranch)
+
+ peerUpdate(t, cat,
NewSetPropertiesUpdate(iceberg.Properties{GCEnabledKey: "false"}))
+ p := peerAppend(t, fs, cat, MainBranch, "p")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ assert.Equal(t, sortedIDs(s1, p, a), survivors(meta))
+ assert.Equal(t, []string{"a", "p", "s1"}, branchFiles(t, fs,
meta, MainBranch))
+ })
+}
+
+// The expiry enables replay of the staged rollback on retry.
+func TestExpireReplayFailsWhenStagedUpdateDoesNotApply(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, inner, tbl := newExpireReplayTable(t, nil)
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "s1")
+ tbl, s2 := appendOnBranch(t, tbl, MainBranch, "s2")
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "s3")
+
+ cat := &recordingCatalog{reqValidatingCatalog: inner}
+ tx := newOCCTable(t, tbl.Metadata(), fs, cat).NewTransaction()
+ require.NoError(t, tx.RollbackToSnapshot(s2))
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(time.Hour)))
+
+ // Keep main at s3 so the rollback assertion still holds after
refresh.
+ peerUpdate(t, inner, NewRemoveSnapshotsUpdate([]int64{s2},
false))
+ inner.failFirst = inner.attempts.Load() + 1
+ before := inner.metadata
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrTransactionUnusable)
+ require.NotErrorIs(t, err, ErrCommitFailed)
+ require.ErrorIs(t, err, ErrSnapshotNotFound)
+ assert.ErrorContains(t, err, "staged update does not apply
after refresh: apply set-snapshot-ref")
+ assert.Len(t, cat.updates, 1)
+ assert.Equal(t, 1, cat.loads)
+ assert.Same(t, before, inner.metadata)
+
+ _, err = tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrTransactionUnusable)
+ assert.Len(t, cat.updates, 1, "an unusable transaction must not
commit")
+ assert.Equal(t, 1, cat.loads, "an unusable transaction must not
refresh")
+ })
+}
+
+func TestExpireReplayKeepsExplicitPin(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t, nil)
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "s1")
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "s2")
+ time.Sleep(time.Millisecond)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AssertRefSnapshotID(MainBranch))
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(1)))
+ peerAppend(t, fs, cat, MainBranch, "p")
+ before := cat.metadata
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrTransactionUnusable)
+ assert.Contains(t, err.Error(), "requirement no longer holds
after refresh")
+ assert.Same(t, before, cat.metadata)
+ })
+}
+
+func manifestListsOf(meta Metadata, ids ...int64) []string {
+ lists := make([]string, len(ids))
+ for i, id := range ids {
+ lists[i] = meta.SnapshotByID(id).ManifestList
+ }
+
+ return lists
+}
+
+func TestExpireReplayCleansUpReplannedRemoval(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t, nil)
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "s1")
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "s2")
+ tbl, s3 := appendOnBranch(t, tbl, MainBranch, "s3")
+ time.Sleep(time.Millisecond)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(2)))
+ p := peerAppend(t, fs, cat, MainBranch, "p")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ assert.Equal(t, sortedIDs(s3, p), survivors(meta))
+ assertOnlyManifestList(t, fs, manifestListsOf(meta, s3, p)...)
+ })
+}
+
+func TestExpireReplayCleansUpRebuiltSnapshotItExpires(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t, nil)
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "b1")
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "b2")
+
+ tx := tbl.NewTransaction()
+ stageFiles(t, tx, "a1")
+ a1 := stagedHead(t, tx, MainBranch)
+ stageFiles(t, tx, "a2")
+ a2 := stagedHead(t, tx, MainBranch)
+ time.Sleep(time.Millisecond)
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(1)))
+ staged, err := tx.StagedTable()
+ require.NoError(t, err)
+ require.Nil(t, staged.Metadata().SnapshotByID(a1), "the staged
plan expires a1")
+
+ peerAppend(t, fs, cat, MainBranch, "p")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ assert.Equal(t, []int64{a2}, survivors(meta))
+ assert.Equal(t, a1, parentOf(t, meta, a2))
+ assert.Equal(t, []string{"a1", "a2", "b1", "b2", "p"},
branchFiles(t, fs, meta, MainBranch))
+ assertOnlyManifestList(t, fs, manifestListsOf(meta, a2)...)
+ })
+}
+
+// The retry rebuilds a on the peer's head, but a keeps its staging timestamp,
+// which is more than a minute before the peer's snapshot.
+func TestExpireReplayRemovesRebuiltListWhenApplyFails(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t, nil)
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "b1")
+
+ tx := tbl.NewTransaction()
+ stageFiles(t, tx, "a")
+ require.NoError(t, tx.ExpireSnapshots(WithOlderThan(time.Hour)))
+
+ time.Sleep(2 * time.Minute)
+ peerAppend(t, fs, cat, MainBranch, "p")
+ files := slices.Sorted(maps.Keys(fs.files))
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrTransactionUnusable)
+ require.NotErrorIs(t, err, ErrCommitFailed)
+ assert.ErrorContains(t, err, "apply add-snapshot: invalid
snapshot timestamp")
+ assert.Equal(t, files, slices.Sorted(maps.Keys(fs.files)), "the
staged files are kept and rebuilt files are removed")
+ })
+}
diff --git a/table/table.go b/table/table.go
index f0ebc4dd3..2fc31a6c3 100644
--- a/table/table.go
+++ b/table/table.go
@@ -55,11 +55,16 @@ import (
// again retries the conflict unless the error also matches
// ErrTransactionUnusable.
//
-// After a refresh, a stale requirement, a validator conflict and a deleted
-// commit branch match ErrCommitFailed. A commit branch that became a tag, a
-// replaced table, a diverged base snapshot and files to rewrite that a
-// concurrent commit removed do not, so a retry loop keyed on ErrCommitFailed
-// stops.
+// After refresh, these failures match ErrCommitFailed: a stale requirement,
+// a validator conflict, a deleted commit branch, or a replanned expiry that
+// changes whether a non-main commit branch is removed.
+//
+// These failures do not: a commit branch that became a tag, a replaced table,
+// a diverged base snapshot, or files to rewrite that a concurrent commit
+// removed.
+// For transactions with snapshot expiry, disabled GC or a staged update that
+// cannot apply after refresh also do not match. An ErrCommitFailed retry loop
+// stops on these failures.
var ErrCommitFailed = errors.New("commit failed, refresh and try again")
// ErrCommitRejected means the catalog rejected a commit without writing
@@ -83,8 +88,10 @@ var ErrCommitRejected = errors.New("commit rejected")
// non-replayable commit whose branch has moved), a validator conflict with a
// concurrent commit, a diverged base snapshot, files to rewrite that a
// concurrent commit removed, a commit branch that became a tag or was
-// deleted, and a replaced table. Only the commit branch's head assertion is
-// rebased.
+// deleted, or a replaced table. With snapshot expiry, they also include
+// disabled GC, replanning that changes whether a non-main commit branch is
+// removed, and a staged update that cannot apply after refresh. Only the
+// commit branch's head assertion is rebased.
var ErrTransactionUnusable = errors.New("transaction cannot be committed
again")
// ErrWriteIORequired is returned by write paths when the table's file system
@@ -606,6 +613,8 @@ type commitOpts struct {
// failed without committing. A resumed commit refreshes and replays
// before its first submission, as the next retry attempt would.
resumes int
+
+ replay *stagedReplay
}
type commitOption func(*commitOpts)
@@ -638,6 +647,10 @@ func withCommitResume(resumes int) commitOption {
return func(o *commitOpts) { o.resumes = resumes }
}
+func withCommitReplay(replay *stagedReplay) commitOption {
+ return func(o *commitOpts) { o.replay = replay }
+}
+
func (t Table) doCommit(ctx context.Context, updates []Update, reqs
[]Requirement, opts ...commitOption) (*Table, error) {
var co commitOpts
for _, apply := range opts {
@@ -735,7 +748,7 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
// equals t.metadata (so the conflict context's concurrent-snapshot
// walk is empty and validators short-circuit). On subsequent
// attempts it is the freshly-loaded post-conflict state.
- current := t.metadata
+ current, currentLoc := t.metadata, t.metadataLocation
// numRetries counts retries; total attempts = 1 initial + numRetries.
totalAttempts := cfg.numRetries + 1
@@ -798,7 +811,7 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
return nil, fmt.Errorf("%w: table was
replaced: %w", ErrTransactionUnusable, err)
}
}
- current = fresh.metadata
+ current, currentLoc = fresh.metadata,
fresh.metadataLocation
reqs = rewriteRefSnapshotRequirements(reqs, co.branch,
current, pinned)
if err := validateNonRebasedRequirements(reqs,
co.branch, pinned, current); err != nil {
@@ -831,7 +844,18 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
// submit an older snapshot whose manifests it now
counts as
// superseded and deletes on success.
forceRebuild := co.resumes > 0 && attempt == 0 &&
!co.noReplay
- rebuiltUpdates, orphaned, rebuildErr :=
rebuildSnapshotUpdates(retryCtx, updates, current, co.branch, wfs,
attemptBase+int(attempt), forceRebuild)
+ var (
+ rebuiltUpdates []Update
+ orphaned []string
+ rebuildErr error
+ step = "rebuild manifest list"
+ )
+ if co.replay != nil {
+ step = "replay staged updates"
+ rebuiltUpdates, orphaned, rebuildErr =
co.replay.replay(retryCtx, current, wfs, attemptBase+int(attempt), forceRebuild)
+ } else {
+ rebuiltUpdates, orphaned, rebuildErr =
rebuildSnapshotUpdates(retryCtx, updates, current, co.branch, wfs,
attemptBase+int(attempt), forceRebuild)
+ }
// A failed rebuild still owns the lists it wrote
before failing.
for list := range stagedManifestLists(rebuiltUpdates) {
if _, ok := staged[list]; !ok {
@@ -839,16 +863,26 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
}
}
if rebuildErr != nil {
+ if errors.Is(rebuildErr,
ErrTransactionUnusable) {
+ return nil, rebuildErr
+ }
// Replaced files missing from the branch head
stay missing, so
// every later attempt fails the same way.
if errors.Is(rebuildErr, ErrCommitDiverged) {
return nil, fmt.Errorf("%w: files to
rewrite were removed: %w", ErrTransactionUnusable, rebuildErr)
}
- return nil, fmt.Errorf("rebuild manifest list
for retry attempt %d: %w", attempt, rebuildErr)
+ return nil, fmt.Errorf("%s for retry attempt
%d: %w", step, attempt, rebuildErr)
}
orphanedManifests = append(orphanedManifests,
orphaned...)
updates = rebuiltUpdates
+
+ // A peer may have already expired the snapshots in the
staged plan.
+ if co.replay != nil && len(updates) == 0 {
+ committed = true
+
+ return t.withMetadata(current, currentLoc), nil
+ }
}
// Pre-flight client-side conflict validation. Producers can
@@ -898,7 +932,14 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
return nil, context.Cause(retryCtx)
}
- newMeta, newLoc, err = t.cat.CommitTable(retryCtx,
slices.Clone(t.identifier), reqs, updates)
+ // Keep per-attempt expiry assertions out of reqs so refresh
does not
+ // treat them as fixed staged requirements.
+ submitReqs := reqs
+ if co.replay != nil {
+ submitReqs = append(slices.Clone(reqs),
expirationAssertions(current, co.replay.branch, co.replay.walkedRefs, reqs)...)
+ }
+
+ newMeta, newLoc, err = t.cat.CommitTable(retryCtx,
slices.Clone(t.identifier), submitReqs, updates)
if err == nil {
committed = true
attemptsUsed = int64(attempt) + 1
@@ -962,10 +1003,14 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
}
}
+ return t.withMetadata(newMeta, newLoc), nil
+}
+
+func (t Table) withMetadata(meta Metadata, loc string) *Table {
next := New(
t.identifier,
- newMeta,
- newLoc,
+ meta,
+ loc,
t.fsF,
t.cat,
withReporterState(t.reporter, t.reporterSet),
@@ -976,7 +1021,7 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
)
next.manifestContentCache =
reuseManifestContentCache(t.manifestContentCache, next.manifestContentCache)
- return next, nil
+ return next
}
// rewriteRefSnapshotRequirements returns a copy of reqs with every
@@ -996,9 +1041,9 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
// branch check that follows rejects the commit.
//
// Assertions on refs in pinned guard the commit (an explicit
-// Transaction.AssertRefSnapshotID, a rollback, or snapshot expiry) and
-// are never rewritten: a ref that has changed must fail the commit, not
-// be replayed against the new head.
+// Transaction.AssertRefSnapshotID or a rollback) and are never
+// rewritten: a ref that has changed must fail the commit, not be
+// replayed against the new head.
func rewriteRefSnapshotRequirements(reqs []Requirement, branch string, fresh
Metadata, pinned map[string]struct{}) []Requirement {
if branch == "" || fresh == nil {
return reqs
@@ -1184,46 +1229,48 @@ func rebuildSnapshotUpdates(ctx context.Context,
updates []Update, freshMeta Met
continue
}
- // Skip only when nothing upstream changed: the running head
still
- // matches this snapshot's recorded parent, no earlier snapshot
in this
- // chain was rebuilt, and the table-wide values the rebuild
recomputes
- // are still acceptable. Saves an unnecessary S3 write.
- if !chainRebuilt && freshHead != nil &&
su.Snapshot.ParentSnapshotID != nil &&
- *su.Snapshot.ParentSnapshotID == freshHead.SnapshotID &&
- stagedSnapshotStillValid(su.Snapshot, freshMeta) {
- // Already parented on the running head; it becomes the
parent of
- // the next staged snapshot in the chain.
- freshHead = su.Snapshot
-
- continue
- }
-
- oldManifestList := su.Snapshot.ManifestList
-
- newSnap, rebuildErr := su.rebuildManifestList(ctx, freshMeta,
freshHead, fs, attempt)
+ next, rebuildErr := rebuildStagedSnapshot(ctx, su, freshMeta,
freshHead, fs, attempt, chainRebuilt)
if rebuildErr != nil {
return result, orphanedPaths, rebuildErr
}
- result[i] = &addSnapshotUpdate{
- baseUpdate: su.baseUpdate,
- Snapshot: newSnap,
- ownManifests: su.ownManifests,
- rebuildManifestList: su.rebuildManifestList,
- written: su.written,
+ freshHead = next.Snapshot
+ if next == su {
+ continue
}
- // The old manifest list is now an orphaned object in object
storage.
- orphanedPaths = append(orphanedPaths, oldManifestList)
+ result[i] = next
- // Chain the next staged snapshot onto the one just rebuilt.
- freshHead = newSnap
+ orphanedPaths = append(orphanedPaths, su.Snapshot.ManifestList)
chainRebuilt = true
}
return result, orphanedPaths, nil
}
+// Reuse requires an unchanged parent, no earlier rebuild in the chain, and
+// valid table-wide values. This avoids an unnecessary S3 write.
+func rebuildStagedSnapshot(ctx context.Context, su *addSnapshotUpdate, meta
Metadata, head *Snapshot, fs icebergio.WriteFileIO, attempt int, chainRebuilt
bool) (*addSnapshotUpdate, error) {
+ if !chainRebuilt && head != nil && su.Snapshot.ParentSnapshotID != nil
&&
+ *su.Snapshot.ParentSnapshotID == head.SnapshotID &&
+ stagedSnapshotStillValid(su.Snapshot, meta) {
+ return su, nil
+ }
+
+ newSnap, err := su.rebuildManifestList(ctx, meta, head, fs, attempt)
+ if err != nil {
+ return nil, err
+ }
+
+ return &addSnapshotUpdate{
+ baseUpdate: su.baseUpdate,
+ Snapshot: newSnap,
+ ownManifests: su.ownManifests,
+ rebuildManifestList: su.rebuildManifestList,
+ written: su.written,
+ }, nil
+}
+
const (
maxRetryDurationMs = uint64(math.MaxInt64 / int64(time.Millisecond))
maxRetryCount = uint64(math.MaxUint32)
diff --git a/table/table_test.go b/table/table_test.go
index 25d9a105a..c76183041 100644
--- a/table/table_test.go
+++ b/table/table_test.go
@@ -36,6 +36,7 @@ import (
"strconv"
"strings"
"testing"
+ "testing/synctest"
"time"
"github.com/apache/arrow-go/v18/arrow"
@@ -2241,7 +2242,8 @@ type validatingCatalog struct {
}
func (m *validatingCatalog) LoadTable(ctx context.Context, ident
table.Identifier) (*table.Table, error) {
- return nil, nil
+ return table.New(ident, m.metadata, "",
+ func(context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }, m), nil
}
func (m *validatingCatalog) CommitTable(ctx context.Context, ident
table.Identifier, reqs []table.Requirement, updates []table.Update)
(table.Metadata, string, error) {
@@ -2262,166 +2264,120 @@ func (m *validatingCatalog) CommitTable(ctx
context.Context, ident table.Identif
return meta, "", nil
}
-// TestExpireSnapshotsRejectsOnRefRollback verifies that ExpireSnapshots fails
-// when a ref is rolled back to an ancestor snapshot concurrently.
-//
-// Scenario:
-// - main -> snapshot 5 (newest), chain: 5 <- 4 <- 3 <- 2 <- 1
-// - ExpireSnapshots calculates: keep {5, 4, 3}, delete {2, 1}
-// - Concurrently, client rolls main -> snapshot 2
-// - Without assertion: would delete snapshots 1, leaving main with only 1
accessible snapshot
-// - With assertion: commit fails because main's snapshot ID changed
-func (t *TableWritingTestSuite) TestExpireSnapshotsRejectsOnRefRollback() {
- fs := iceio.LocalFS{}
-
- files := make([]string, 0)
- for i := range 5 {
- filePath :=
fmt.Sprintf("%s/expire_reject_rollback_v%d/data-%d.parquet", t.location,
t.formatVersion, i)
- t.writeParquet(fs, filePath, t.arrTablePromotedTypes)
- files = append(files, filePath)
- }
+// A millisecond sleep makes earlier snapshots older than the expiry cutoff.
+func (t *TableWritingTestSuite) inBubble(f func()) {
+ outer := t.T()
+ synctest.Test(outer, func(st *testing.T) {
+ t.SetT(st)
+ defer t.SetT(outer)
+ f()
+ })
+}
- ident := table.Identifier{"default", "expire_reject_rollback_v" +
strconv.Itoa(t.formatVersion)}
+func (t *TableWritingTestSuite) newExpireRetryTable(name string, count int,
sleepAfterFirst time.Duration) (*table.Table, *validatingCatalog) {
+ fs := iceio.LocalFS{}
+ ident := table.Identifier{"default", name + "_v" +
strconv.Itoa(t.formatVersion)}
meta, err := table.NewMetadata(t.tableSchemaPromotedTypes,
iceberg.UnpartitionedSpec,
- table.UnsortedSortOrder, t.location,
iceberg.Properties{table.PropertyFormatVersion: strconv.Itoa(t.formatVersion)})
+ table.UnsortedSortOrder, t.location, iceberg.Properties{
+ table.PropertyFormatVersion:
strconv.Itoa(t.formatVersion),
+ table.CommitNumRetriesKey: "1",
+ table.CommitMinRetryWaitMsKey: "1",
+ table.CommitMaxRetryWaitMsKey: "2",
+ })
t.Require().NoError(err)
ctx := context.Background()
cat := &validatingCatalog{meta}
-
- tbl := table.New(
- ident,
- meta,
- t.getMetadataLoc(),
- func(ctx context.Context) (iceio.IO, error) {
- return fs, nil
- },
- cat,
- )
-
- // Create 5 snapshots
- for i := range 5 {
+ tbl := table.New(ident, meta, t.getMetadataLoc(),
+ func(ctx context.Context) (iceio.IO, error) { return fs, nil },
cat)
+ for i := range count {
+ file := fmt.Sprintf("%s/%s_v%d/data-%d.parquet", t.location,
name, t.formatVersion, i)
+ t.writeParquet(fs, file, t.arrTablePromotedTypes)
tx := tbl.NewTransaction()
- t.Require().NoError(tx.AddDataFilesByPath(ctx, files[i:i+1],
nil, false))
+ t.Require().NoError(tx.AddDataFilesByPath(ctx, []string{file},
nil, false))
tbl, err = tx.Commit(ctx)
t.Require().NoError(err)
+ if i == 0 {
+ time.Sleep(sleepAfterFirst)
+ }
}
+ t.Require().Len(tbl.Metadata().Snapshots(), count)
- t.Require().Equal(5, len(tbl.Metadata().Snapshots()))
-
- // Get snapshot IDs for later use
- snapshots := tbl.Metadata().Snapshots()
- snapshot2 := snapshots[1] // Second snapshot (index 1)
-
- // Start ExpireSnapshots transaction (will calculate based on current
main -> snapshot 5)
- tx := tbl.NewTransaction()
- t.Require().NoError(tx.ExpireSnapshots(table.WithOlderThan(0),
table.WithRetainLast(3)))
+ return tbl, cat
+}
- // Simulate concurrent rollback: update catalog's metadata to point
main -> snapshot 2
- // This simulates another client rolling back main before
ExpireSnapshots commits
- rollbackUpdates := []table.Update{
- table.NewSetSnapshotRefUpdate("main", snapshot2.SnapshotID,
table.BranchRef, -1, -1, -1),
+func snapshotIDsOf(meta table.Metadata) []int64 {
+ var ids []int64
+ for _, snap := range meta.Snapshots() {
+ ids = append(ids, snap.SnapshotID)
}
- cat.metadata, _, err = cat.CommitTable(ctx, ident, nil, rollbackUpdates)
- t.Require().NoError(err)
- // Attempt to commit ExpireSnapshots - should fail due to
AssertRefSnapshotID
- _, err = tx.Commit(ctx)
- t.Require().Error(err)
- t.Require().Contains(err.Error(), "requirement failed")
- t.Require().Contains(err.Error(), "main")
+ return ids
}
-// TestExpireSnapshotsRejectsOnRefUpdate verifies that ExpireSnapshots fails
-// when a ref eligible for deletion is concurrently updated to a newer
snapshot.
-//
// Scenario:
-// - tag1 -> old snapshot, eligible for deletion (maxRefAgeMs exceeded)
-// - ExpireSnapshots decides to remove tag1
-// - Concurrently, client updates tag1 -> newer snapshot (no longer eligible)
-// - Without assertion: tag1 would be deleted despite being updated
-// - With assertion: commit fails because tag1's snapshot ID changed
-func (t *TableWritingTestSuite) TestExpireSnapshotsRejectsOnRefUpdate() {
- fs := iceio.LocalFS{}
+// - main -> snapshot 5 (newest), chain: 5 <- 4 <- 3 <- 2 <- 1
+// - Concurrently, client rolls main -> snapshot 2
+func (t *TableWritingTestSuite) TestExpireSnapshotsReplansAfterRefRollback() {
+ t.inBubble(func() {
+ tbl, cat := t.newExpireRetryTable("expire_replan_rollback", 5,
0)
+ snapshots := tbl.Metadata().Snapshots()
+ time.Sleep(time.Millisecond)
- files := make([]string, 0)
- for i := range 3 {
- filePath :=
fmt.Sprintf("%s/expire_reject_update_v%d/data-%d.parquet", t.location,
t.formatVersion, i)
- t.writeParquet(fs, filePath, t.arrTablePromotedTypes)
- files = append(files, filePath)
- }
+ ctx := context.Background()
+ tx := tbl.NewTransaction()
+ t.Require().NoError(tx.ExpireSnapshots(table.WithOlderThan(0),
table.WithRetainLast(3)))
- ident := table.Identifier{"default", "expire_reject_update_v" +
strconv.Itoa(t.formatVersion)}
- meta, err := table.NewMetadata(t.tableSchemaPromotedTypes,
iceberg.UnpartitionedSpec,
- table.UnsortedSortOrder, t.location,
iceberg.Properties{table.PropertyFormatVersion: strconv.Itoa(t.formatVersion)})
- t.Require().NoError(err)
+ var err error
+ cat.metadata, _, err = cat.CommitTable(ctx, tbl.Identifier(),
nil, []table.Update{
+ table.NewSetSnapshotRefUpdate("main",
snapshots[1].SnapshotID, table.BranchRef, -1, -1, -1),
+ })
+ t.Require().NoError(err)
- ctx := context.Background()
- cat := &validatingCatalog{meta}
+ committed, err := tx.Commit(ctx)
+ t.Require().NoError(err)
+ t.Equal([]int64{snapshots[0].SnapshotID,
snapshots[1].SnapshotID}, snapshotIDsOf(committed.Metadata()))
+ t.Equal(snapshots[1].SnapshotID,
committed.Metadata().CurrentSnapshot().SnapshotID)
+ t.Equal(snapshotIDsOf(committed.Metadata()),
snapshotIDsOf(cat.metadata))
+ })
+}
- tbl := table.New(
- ident,
- meta,
- t.getMetadataLoc(),
- func(ctx context.Context) (iceio.IO, error) {
- return fs, nil
- },
- cat,
- )
+// Scenario:
+// - snapshot 1 is two hours older than snapshots 2 and 3
+// - expiring-tag -> snapshot 1, with a max ref age of one hour
+func (t *TableWritingTestSuite) TestExpireSnapshotsReplansAfterRefUpdate() {
+ t.inBubble(func() {
+ tbl, cat := t.newExpireRetryTable("expire_replan_update", 3,
2*time.Hour)
+ snapshots := tbl.Metadata().Snapshots()
+
+ ctx := context.Background()
+ maxRefAgeMs := time.Hour.Milliseconds()
+ var err error
+ cat.metadata, _, err = cat.CommitTable(ctx, tbl.Identifier(),
nil, []table.Update{
+ table.NewSetSnapshotRefUpdate("expiring-tag",
snapshots[0].SnapshotID, table.TagRef, maxRefAgeMs, -1, -1),
+ })
+ t.Require().NoError(err)
+ tbl = table.New(tbl.Identifier(), cat.metadata,
t.getMetadataLoc(),
+ func(ctx context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }, cat)
- // Create 3 snapshots
- for i := range 3 {
tx := tbl.NewTransaction()
- t.Require().NoError(tx.AddDataFilesByPath(ctx, files[i:i+1],
nil, false))
- tbl, err = tx.Commit(ctx)
+
t.Require().NoError(tx.ExpireSnapshots(table.WithOlderThan(time.Hour),
table.WithRetainLast(1)))
+ staged, err := tx.StagedTable()
t.Require().NoError(err)
- }
- t.Require().Equal(3, len(tbl.Metadata().Snapshots()))
-
- snapshots := tbl.Metadata().Snapshots()
- oldSnapshot := snapshots[0] // Oldest snapshot
- newerSnapshot := snapshots[2] // Newest snapshot
-
- // Create a tag pointing to the old snapshot with a short maxRefAgeMs
- // This tag will be eligible for deletion
- maxRefAgeMs := int64(1) // 1ms - will definitely be exceeded
- tagUpdates := []table.Update{
- table.NewSetSnapshotRefUpdate("expiring-tag",
oldSnapshot.SnapshotID, table.TagRef, maxRefAgeMs, -1, -1),
- }
- cat.metadata, _, err = cat.CommitTable(ctx, ident, nil, tagUpdates)
- t.Require().NoError(err)
+
t.Require().Nil(staged.Metadata().SnapshotByName("expiring-tag"), "the staged
plan removes the tag")
- // Reload table with updated metadata
- tbl = table.New(
- ident,
- cat.metadata,
- t.getMetadataLoc(),
- func(ctx context.Context) (iceio.IO, error) {
- return fs, nil
- },
- cat,
- )
-
- // Wait a bit to ensure the tag's ref age exceeds maxRefAgeMs
- time.Sleep(10 * time.Millisecond)
-
- // Start ExpireSnapshots transaction (will identify expiring-tag as
eligible for deletion)
- tx := tbl.NewTransaction()
- t.Require().NoError(tx.ExpireSnapshots(table.WithOlderThan(time.Hour),
table.WithRetainLast(1)))
-
- // Simulate concurrent update: another client updates the tag to point
to a newer snapshot
- // This makes the tag no longer eligible for deletion
- updateTagUpdates := []table.Update{
- table.NewSetSnapshotRefUpdate("expiring-tag",
newerSnapshot.SnapshotID, table.TagRef, maxRefAgeMs, -1, -1),
- }
- cat.metadata, _, err = cat.CommitTable(ctx, ident, nil,
updateTagUpdates)
- t.Require().NoError(err)
+ cat.metadata, _, err = cat.CommitTable(ctx, tbl.Identifier(),
nil, []table.Update{
+ table.NewSetSnapshotRefUpdate("expiring-tag",
snapshots[2].SnapshotID, table.TagRef, maxRefAgeMs, -1, -1),
+ })
+ t.Require().NoError(err)
- // Attempt to commit ExpireSnapshots - should fail due to
AssertRefSnapshotID
- _, err = tx.Commit(ctx)
- t.Require().Error(err)
- t.Require().Contains(err.Error(), "requirement failed")
- t.Require().Contains(err.Error(), "expiring-tag")
+ committed, err := tx.Commit(ctx)
+ t.Require().NoError(err)
+ t.Equal([]int64{snapshots[1].SnapshotID,
snapshots[2].SnapshotID}, snapshotIDsOf(committed.Metadata()))
+ tag := committed.Metadata().SnapshotByName("expiring-tag")
+ t.Require().NotNil(tag)
+ t.Equal(snapshots[2].SnapshotID, tag.SnapshotID)
+ })
}
func (t *TableWritingTestSuite) TestWriteSpecialCharacterColumn() {
diff --git a/table/transaction.go b/table/transaction.go
index ee88ac319..37a74fc3b 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -95,12 +95,14 @@ type Transaction struct {
noReplay bool
// pinnedRefs names refs whose AssertRefSnapshotID requirement guards
- // the commit: those from AssertRefSnapshotID, RollbackToSnapshot, and
- // ExpireSnapshots. doCommit's refresh-and-replay must not rewrite
- // their assertions to the fresh head between retries, so a ref that
- // has changed fails the commit.
+ // the commit: those from AssertRefSnapshotID and RollbackToSnapshot.
+ // doCommit's refresh-and-replay must not rewrite their assertions to
+ // the fresh head between retries, so a ref that has changed fails the
+ // commit.
pinnedRefs map[string]struct{}
+ expirations []expirationRecord
+
mx sync.Mutex
committed bool
@@ -683,6 +685,19 @@ type expireSnapshotsCfg struct {
validationErr error
}
+// Keep calls that staged no updates so retries can replan them.
+type expirationRecord struct {
+ cfg expireSnapshotsCfg
+ nowMs int64
+
+ // pos and count refer to the original staged updates, not a retry's
plan.
+ pos, count int
+
+ walked []string
+
+ removedBranch bool
+}
+
type ExpireSnapshotsOpt func(*expireSnapshotsCfg)
// WithRetainLast sets the minimum number of snapshots to keep per branch,
@@ -765,20 +780,27 @@ func getRetentionInt64(props iceberg.Properties,
standardKey, legacyKey string,
// }
// newTbl, err := txn.Commit(ctx)
//
-// An expiry that removes something pins every ref it read, including the
-// commit branch. A transaction that also appends then fails rather than
-// rebasing onto a peer's append, since the expiry was computed from the old
-// head.
+// After refresh, a retry or resumed commit replans the expiry at its original
+// position, with this call's options and cutoff time. Replanning retains
+// snapshots made reachable by concurrent commits, including rollback targets
+// and new branch heads. It can expire snapshots that a concurrent append moved
+// outside retention. On each attempt, an expiry that removes snapshots or
+// refs requires the refs it read to be unchanged since loading or refreshing
+// the table.
+//
+// Retries replan even if staging or an earlier attempt produced no expiry
+// updates. If no updates remain, Commit returns the refreshed table without
+// committing. A transaction containing only expiries with no updates makes no
+// commit attempt and never refreshes or replans.
+//
+// The replay failures that make the transaction unusable are listed on
+// ErrTransactionUnusable.
//
// The "iceberg expire-snapshots" CLI command wraps the same operation.
func (t *Transaction) ExpireSnapshots(opts ...ExpireSnapshotsOpt) error {
var (
- cfg = expireSnapshotsCfg{postCommit: true}
- updates []Update
- reqs []Requirement
- pinned []string
- snapsToKeep = make(map[int64]struct{})
- nowMs = time.Now().UnixMilli()
+ cfg = expireSnapshotsCfg{postCommit: true}
+ nowMs = time.Now().UnixMilli()
)
for _, opt := range opts {
@@ -795,6 +817,56 @@ func (t *Transaction) ExpireSnapshots(opts
...ExpireSnapshotsOpt) error {
return err
}
+ updates, walked, err := planExpiration(meta, cfg, nowMs)
+ if err != nil {
+ return err
+ }
+
+ return t.stageExpiration(expirationRecord{
+ cfg: cfg,
+ nowMs: nowMs,
+ walked: walked,
+ removedBranch: removesRef(updates, cmp.Or(t.branch,
MainBranch)),
+ }, updates)
+}
+
+// Record positions under the same lock as the updates they locate.
+func (t *Transaction) stageExpiration(rec expirationRecord, updates []Update)
error {
+ if err := t.checkNotNil(); err != nil {
+ return err
+ }
+ t.mx.Lock()
+ defer t.mx.Unlock()
+
+ meta, err := t.txnMeta()
+ if err != nil {
+ return err
+ }
+ rec.pos = len(meta.updates)
+ if err := t.applyLocked(updates, nil); err != nil {
+ return err
+ }
+ rec.count = len(t.meta.updates) - rec.pos
+ t.expirations = append(t.expirations, rec)
+
+ return nil
+}
+
+func removesRef(updates []Update, name string) bool {
+ return slices.ContainsFunc(updates, func(u Update) bool {
+ r, ok := u.(*removeSnapshotRefUpdate)
+
+ return ok && r.RefName == name
+ })
+}
+
+func planExpiration(meta *MetadataBuilder, cfg expireSnapshotsCfg, nowMs
int64) ([]Update, []string, error) {
+ var (
+ updates []Update
+ walked = make([]string, 0, len(meta.refs))
+ snapsToKeep = make(map[int64]struct{})
+ )
+
// Read table-level retention properties as the last-resort defaults,
// mirroring the Java implementation. The unprefixed property names are
// retained as compatibility fallbacks for tables created by older Go
@@ -814,7 +886,7 @@ func (t *Transaction) ExpireSnapshots(opts
...ExpireSnapshotsOpt) error {
}
snapshots, err := meta.snapshotsForFullScan()
if err != nil {
- return err
+ return nil, nil, err
}
lookup := func(id int64) *Snapshot {
@@ -825,19 +897,11 @@ func (t *Transaction) ExpireSnapshots(opts
...ExpireSnapshotsOpt) error {
retainedRefs := make(map[string]SnapshotRef, len(meta.refs))
for refName, ref := range meta.refs {
- // Assert and pin the ref's base snapshot id so we don't expire
- // snapshots that concurrently updated refs now reference. Refs
- // the transaction itself staged (absent on the base) already
- // carry an absence assertion, which the pin keeps from being
- // rebased.
- if id := t.baseRefSnapshotID(refName); id != nil {
- reqs = append(reqs, AssertRefSnapshotID(refName, id))
- }
- pinned = append(pinned, refName)
+ walked = append(walked, refName)
snap, err := meta.SnapshotByID(ref.SnapshotID)
if err != nil {
- return err
+ return nil, nil, err
}
maxRefAgeMs := cmp.Or(ref.MaxRefAgeMs, &propMaxRefAgeMs)
@@ -914,14 +978,7 @@ func (t *Transaction) ExpireSnapshots(opts
...ExpireSnapshotsOpt) error {
updates = append(updates, removal)
}
- // A no-op expiry must not pin the commit branch, which would stop the
- // transaction's other updates from rebasing on retry. apply still
- // rejects a committed or unusable transaction.
- if len(updates) == 0 {
- return t.apply(nil, nil)
- }
-
- return t.applyPinned(updates, reqs, pinned)
+ return updates, walked, nil
}
func (t *Transaction) AppendTable(ctx context.Context, tbl arrow.Table,
batchSize int64, snapshotProps iceberg.Properties) error {
@@ -2638,12 +2695,17 @@ func (t *Transaction) Commit(ctx context.Context)
(*Table, error) {
if len(meta.updates) > 0 {
reqs := append(transactionRequirements(t.reqs, t.branch,
t.tbl.metadata), AssertTableUUID(meta.uuid))
+ var replay *stagedReplay
+ if len(t.expirations) > 0 {
+ replay = newStagedReplay(cmp.Or(t.branch, MainBranch),
meta.updates, t.expirations)
+ }
tbl, err := t.tbl.doCommit(ctx, meta.updates, reqs,
withCommitBranch(t.branch),
withCommitValidators(t.validators...),
withCommitNoReplay(t.noReplay),
withCommitPinnedRefs(t.pinnedRefs),
withCommitResume(t.failedCommits),
+ withCommitReplay(replay),
)
if err != nil {
// A failure that committed nothing stays committable,
and the next
@@ -2668,8 +2730,12 @@ func (t *Transaction) Commit(ctx context.Context)
(*Table, error) {
// otherwise a retry would re-run the catalog commit.
t.committed = true
+ landed := meta.updates
+ if replay != nil && replay.ran {
+ landed = replay.landed
+ }
var removals []*removeSnapshotsUpdate
- for _, u := range meta.updates {
+ for _, u := range landed {
if removal, ok := u.(*removeSnapshotsUpdate); ok {
removals = append(removals, removal)
diff --git a/website/src/concurrent-writes.md b/website/src/concurrent-writes.md
index 2dbca534a..fadd8fad4 100644
--- a/website/src/concurrent-writes.md
+++ b/website/src/concurrent-writes.md
@@ -84,11 +84,11 @@ rejects when concurrent deletes touch the same files.
iceberg-go can refresh the table and replay the operation against the latest
snapshot between attempts. On each retry the table metadata is reloaded, the
producer's validators re-run against the fresh snapshot, and the commit is
-re-submitted only if it is still valid. `RollbackToSnapshot`, `ExpireSnapshots`
-and `AssertRefSnapshotID` pin the refs they guard, so a retry does not rebase
-those assertions. If a guarded ref has moved, the commit fails with
-`ErrCommitFailed` instead of retrying. An `ExpireSnapshots` that removes
nothing
-adds no guards.
+re-submitted only if it is still valid. `RollbackToSnapshot` and
+`AssertRefSnapshotID` pin their refs, so retries do not rebase those
assertions.
+If a pinned ref moved, the commit fails with `ErrCommitFailed` instead of
+retrying. `ExpireSnapshots` does not pin refs. Retries replan it as described
in
+[Snapshot expiration](#snapshot-expiration).
Retry is **off by default** (`commit.retry.num-retries` is `0`). Opt in by
setting the retry properties — see [Manifest and
commit](./configuration.md#manifest-and-commit)
@@ -129,6 +129,34 @@ manifests that later attempts merged. `TableCommit`
refuses a transaction whose
commit failed, since the multi-table path does not refresh. Call `Commit` to
resume it, or build a new transaction.
-Because an `ExpireSnapshots` that removes something pins the commit branch, a
-transaction that also appends fails instead of rebasing onto a peer's append.
-This is deliberate, since the expiry was computed from the old head.
+### Snapshot expiration
+
+A retry replays staged changes in order against the refreshed table. It replans
+each `ExpireSnapshots` call at its original position, with the original options
+and cutoff time. The plan retains snapshots that a concurrent commit made
+reachable, such as rollback targets and new branch heads. It can expire
+snapshots that a concurrent append moved outside the retention window. A
+transaction that appends and expires snapshots rebases onto a concurrent append
+like any other append.
+
+Each attempt requires the refs read by expiries that remove snapshots or refs
+to be unchanged since loading or refreshing the table. An expiry that removes
+nothing adds no such requirement. If a concurrent commit moves one of those
+refs before the attempt succeeds, it causes another conflict. The next attempt
+replans the expiry. If a concurrent commit creates a ref that the plan did not
+read, that ref does not cause a conflict. Expiring its snapshot also removes
+the ref.
+
+- Retries replan an expiry even if staging or an earlier attempt produced no
+ expiry updates.
+- If no updates remain, for example because a concurrent commit already expired
+ the same snapshots, `Commit` returns the refreshed table without committing.
+- A transaction containing only expiries with no updates makes no commit
attempt
+ and never refreshes or replans.
+- Post-commit cleanup deletes files for snapshots expired by the successful
+ attempt, if cleanup is enabled.
+
+A retry fails with `ErrTransactionUnusable` if GC is disabled after refresh, a
+staged change no longer applies, or replanning changes whether a non-main
+commit branch is removed. `ErrTransactionUnusable` and `ErrCommitFailed` list
+these cases and which of them a new transaction can retry.
diff --git a/website/src/releases.md b/website/src/releases.md
index bfc29a9fd..f4885948d 100644
--- a/website/src/releases.md
+++ b/website/src/releases.md
@@ -62,6 +62,7 @@ Per-release notes (highlights, breaking changes,
contributors) are published on
- **Hadoop catalog construction takes a context.** `hadoop.NewCatalog(name,
warehouse, props)` becomes `hadoop.NewCatalog(ctx, name, warehouse, props)`.
The context bounds filesystem setup; the catalog's filesystem keeps its values
but not its cancellation.
- **Config loading reports errors.** `config.LoadConfig` now returns `([]byte,
error)`, and `config.LoadConfigFile` is removed; call `LoadConfig` instead. A
missing default `~/.iceberg-go.yaml` still returns no data and no error.
- **Branch transactions return initialization errors.**
`Table.NewTransactionOnBranch(branch)` now returns `(*Transaction, error)`, and
`Table.NewTransactionOnBranchWithError` is removed. Use
`NewTransactionOnBranch` and handle its error.
+- **Snapshot expiration now tolerates concurrent ref changes.** If a
concurrent commit moves a ref that the expiry read, a retry or later `Commit`
replans instead of failing with `table.ErrCommitFailed`. The committed expiry
can differ from `Transaction.StagedTable`. See [Snapshot
expiration](./concurrent-writes.md#snapshot-expiration) for cases that now fail
with `table.ErrTransactionUnusable`.
- **Schema JSON constructor uses the Go initialism.** Replace
`iceberg.NewSchemaFromJsonFields` with `iceberg.NewSchemaFromJSONFields`.
- **AWS identifiers use the Go initialism.** `glue.AwsProperties`,
`glue.WithAwsConfig` and `glue.WithAwsProperties` become `glue.AWSProperties`,
`glue.WithAWSConfig` and `glue.WithAWSProperties`; `sigv4.WithAwsConfig`
becomes `sigv4.WithAWSConfig`; `utils.WithAwsConfig` and `utils.GetAwsConfig`
become `utils.WithAWSConfig` and `utils.GetAWSConfig`; and
`config.CatalogConfig.AwsProfile` becomes `AWSProfile`. The `aws-profile` YAML
key is unchanged.
- **KMS factory registration now takes a context.** Change
`encryption.KMSFactory` implementations to accept `(context.Context,
map[string]string)` instead of only properties, and call
`encryption.LoadKeyManagementClient(ctx, props)` instead of
`LoadKeyManagementClient(props)`. The existing global `RegisterKMS`,
`UnregisterKMS`, and `GetRegisteredKMSNames` functions still use the default
registry; `RegisterKMS` still panics for invalid registrations. Use
`encryption.MustRegisterKMS` for [...]
