Copilot commented on code in PR #2468:
URL: https://github.com/apache/iceberg-go/pull/2468#discussion_r4516633081
##########
table/table.go:
##########
@@ -656,6 +659,9 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
for _, apply := range opts {
apply(&co)
}
+ if co.replay == nil {
+ co.replay = newStagedReplay(cmp.Or(co.branch, MainBranch),
updates, nil)
Review Comment:
`withCommitNoReplay` looks like it is intended to disable replay, but
`doCommit` now always initializes `co.replay` when nil and unconditionally
calls `co.replay.replay(...)` / injects `co.replay.requirements(...)`. As a
result, `noReplay` appears to only affect `forceRebuild` and no longer disables
replay behavior. If callers rely on `withCommitNoReplay(true)` to bypass
replay, this is a behavioral break. Consider making replay creation/calls
conditional on `!co.noReplay` (and skipping replay-derived requirements as
well), or reworking the option so it still achieves “no replay” semantics.
##########
table/table.go:
##########
@@ -932,12 +927,9 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
return nil, context.Cause(retryCtx)
}
- // 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)...)
- }
+ // Keep replay assertions out of reqs so refresh does not treat
them as
+ // fixed staged requirements.
+ submitReqs := append(slices.Clone(reqs),
co.replay.requirements(current, reqs)...)
Review Comment:
`withCommitNoReplay` looks like it is intended to disable replay, but
`doCommit` now always initializes `co.replay` when nil and unconditionally
calls `co.replay.replay(...)` / injects `co.replay.requirements(...)`. As a
result, `noReplay` appears to only affect `forceRebuild` and no longer disables
replay behavior. If callers rely on `withCommitNoReplay(true)` to bypass
replay, this is a behavioral break. Consider making replay creation/calls
conditional on `!co.noReplay` (and skipping replay-derived requirements as
well), or reworking the option so it still achieves “no replay” semantics.
##########
table/table.go:
##########
@@ -844,18 +850,7 @@ 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
- 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)
- }
+ rebuiltUpdates, orphaned, rebuildErr :=
co.replay.replay(retryCtx, current, wfs, attemptBase+int(attempt), forceRebuild)
Review Comment:
`withCommitNoReplay` looks like it is intended to disable replay, but
`doCommit` now always initializes `co.replay` when nil and unconditionally
calls `co.replay.replay(...)` / injects `co.replay.requirements(...)`. As a
result, `noReplay` appears to only affect `forceRebuild` and no longer disables
replay behavior. If callers rely on `withCommitNoReplay(true)` to bypass
replay, this is a behavioral break. Consider making replay creation/calls
conditional on `!co.noReplay` (and skipping replay-derived requirements as
well), or reworking the option so it still achieves “no replay” semantics.
##########
table/commit_refresh_replay_test.go:
##########
@@ -652,3 +654,149 @@ func TestRewriteRefSnapshotRequirements(t *testing.T) {
missingBranchOut := rewriteRefSnapshotRequirements(reqs,
"non-existent", fresh, nil)
assert.Equal(t, reqs, missingBranchOut)
}
+
+// A conflict on another branch must not override a staged rollback of main.
+func TestRefreshReplayKeepsRollbackBeforeAppend(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t, nil)
+ tbl, s1 := appendOnBranch(t, tbl, MainBranch, "s1")
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "s2")
+ peerUpdate(t, cat, NewSetSnapshotRefUpdate("other", s1, BranchRef, 0,
0, 0))
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.RollbackToSnapshot(s1))
+ stageFiles(t, tx, "a")
+ peerAppend(t, fs, cat, "other", "p")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ head := meta.SnapshotByName(MainBranch)
+ require.NotNil(t, head)
+ assert.Equal(t, s1, parentOf(t, meta, head.SnapshotID))
+ assert.Equal(t, []string{"a", "s1"}, branchFiles(t, fs, meta,
MainBranch))
+}
+
+// Sequence numbers and row IDs must advance across a staged rollback.
+func TestRefreshReplayKeepsCountersAcrossStagedRollback(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t,
iceberg.Properties{PropertyFormatVersion: "3"})
+ stage := func(tx *Transaction, name string) {
+ builder, err :=
iceberg.NewDataFileBuilder(iceberg.NewPartitionSpec(), iceberg.EntryContentData,
+ retryCleanupLocation+"/data/"+name+".parquet",
iceberg.ParquetFile, nil, nil, nil, 1, 1)
+ require.NoError(t, err)
+ require.NoError(t, tx.AddDataFiles(t.Context(),
[]iceberg.DataFile{builder.FirstRowID(0).Build()}, nil))
+ }
+ commit := func(tbl *Table, branch, name string) *Table {
+ tx, err := tbl.NewTransactionOnBranch(branch)
+ require.NoError(t, err)
+ stage(tx, name)
+ out, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ return out
+ }
+ tbl = commit(tbl, MainBranch, "s1")
+ s1 := tbl.CurrentSnapshot().SnapshotID
+ tbl = commit(tbl, MainBranch, "s2")
+ peerUpdate(t, cat, NewSetSnapshotRefUpdate("other", s1, BranchRef, 0,
0, 0))
+
+ tx := tbl.NewTransaction()
+ stage(tx, "a")
+ a := stagedHead(t, tx, MainBranch)
+ require.NoError(t, tx.RollbackToSnapshot(s1))
+ stage(tx, "b")
+ b := stagedHead(t, tx, MainBranch)
+ commit(newOCCTable(t, cat.metadata, fs, cat), "other", "p")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ snapA, snapB := meta.SnapshotByID(a), meta.SnapshotByID(b)
+ peer := meta.SnapshotByName("other")
+ require.NotNil(t, snapA)
+ require.NotNil(t, snapB)
+ assert.Equal(t, peer.SequenceNumber+1, snapA.SequenceNumber)
+ assert.Equal(t, snapA.SequenceNumber+1, snapB.SequenceNumber)
+ require.NotNil(t, peer.FirstRowID)
+ require.NotNil(t, peer.AddedRows)
+ require.NotNil(t, snapA.FirstRowID)
+ require.NotNil(t, snapA.AddedRows)
+ require.NotNil(t, snapB.FirstRowID)
+ assert.Equal(t, *peer.FirstRowID+*peer.AddedRows, *snapA.FirstRowID)
+ assert.Equal(t, *snapA.FirstRowID+*snapA.AddedRows, *snapB.FirstRowID)
+ assert.Equal(t, *snapB.FirstRowID+1, meta.NextRowID())
+ assert.Equal(t, s1, parentOf(t, meta, b))
+ assert.Equal(t, []string{"b", "s1"}, branchFiles(t, fs, meta,
MainBranch))
+}
+
+// A v1 snapshot has no sequence number, so replay onto a v2 table reports
+// a conflict.
+func TestRefreshReplayApplyFailureKeepsConflictCause(t *testing.T) {
+ _, cat, tbl := newExpireReplayTable(t,
iceberg.Properties{PropertyFormatVersion: "1"})
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "base")
+ tx := tbl.NewTransaction()
+ stageFiles(t, tx, "a")
+ peerUpdate(t, cat, NewUpgradeFormatVersionUpdate(2))
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrTransactionUnusable)
+ require.ErrorIs(t, err, ErrCommitFailed)
+ assert.ErrorContains(t, err, "apply add-snapshot")
+}
+
+// A peer append to main more than a minute after staging moves the table's
+// last update past the staged timestamps without moving the feature branch.
+func TestRefreshReplayRenewsTimestampsOnRebuild(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, "feature", "f1")
+ tx, err := tbl.NewTransactionOnBranch("feature")
+ require.NoError(t, err)
+ stageFiles(t, tx, "a")
+ a := stagedHead(t, tx, "feature")
+ stageFiles(t, tx, "b")
+ b := stagedHead(t, tx, "feature")
+
+ time.Sleep(61 * time.Second)
+ peerAppend(t, fs, cat, MainBranch, "p")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ snapA, snapB := meta.SnapshotByID(a), meta.SnapshotByID(b)
+ require.NotNil(t, snapA)
+ require.NotNil(t, snapB)
+ assert.Equal(t, time.Now().UnixMilli(), snapA.TimestampMs)
+ assert.Equal(t, snapA.TimestampMs+1, snapB.TimestampMs, "a
child is stamped after its parent")
Review Comment:
These assertions compare snapshot timestamps to `time.Now().UnixMilli()` for
exact equality. With the new behavior (`timestamp = max(now, parent+1ms)`), the
stamped value may legitimately be `parent.TimestampMs+1` (or otherwise differ
from the `time.Now()` call made in the test), making this brittle/flaky even
under `synctest`. Recommendation: assert relative/ordering properties instead
(e.g., `>= beforeCommit`, `>= parent.TimestampMs+1`, and optionally `<=
afterCommit` when applicable), or compute an expected value using the same
`max(now, parent+1)` rule from captured timestamps.
##########
table/expire_replay_test.go:
##########
@@ -787,38 +789,62 @@ func TestExpireReplayCleansUpRebuiltSnapshotItExpires(t
*testing.T) {
require.NoError(t, err)
require.Nil(t, staged.Metadata().SnapshotByID(a1), "the staged
plan expires a1")
- peerAppend(t, fs, cat, MainBranch, "p")
+ p := 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, sortedIDs(a1, a2, p), survivors(meta))
assert.Equal(t, a1, parentOf(t, meta, a2))
+ assert.Equal(t, p, parentOf(t, meta, a1))
assert.Equal(t, []string{"a1", "a2", "b1", "b2", "p"},
branchFiles(t, fs, meta, MainBranch))
- assertOnlyManifestList(t, fs, manifestListsOf(meta, a2)...)
+ assertOnlyManifestList(t, fs, manifestListsOf(meta, a1, a2,
p)...)
})
}
-// 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.
+// The peer's v3 table requires first-row-id, which the staged v2 snapshot
+// lacks.
func TestExpireReplayRemovesRebuiltListWhenApplyFails(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)))
+
+ peerAppend(t, fs, cat, MainBranch, "p")
+ peerUpdate(t, cat, NewUpgradeFormatVersionUpdate(3))
+ 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 row lineage:
first-row-id is required")
+ assert.Equal(t, files, slices.Sorted(maps.Keys(fs.files)), "the staged
files are kept and rebuilt files are removed")
+}
+
+// A peer append more than a minute after staging makes the staged timestamp
+// too old for the refreshed table.
+func TestExpireReplayRenewsStagedSnapshotTimestamp(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")
+ a := stagedHead(t, tx, MainBranch)
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))
+ p := peerAppend(t, fs, cat, MainBranch, "p")
- _, 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")
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ assert.Equal(t, p, parentOf(t, meta, a))
+ assert.Equal(t, time.Now().UnixMilli(),
meta.SnapshotByID(a).TimestampMs)
Review Comment:
The test asserts the renewed snapshot timestamp equals
`time.Now().UnixMilli()` exactly. Since renewal uses `max(now, parent+1ms)`,
the correct value may be `parent.TimestampMs+1` (and may not match the test's
`time.Now()` call). To avoid flaky behavior, assert the timestamp is at least
`meta.SnapshotByID(p).TimestampMs+1` (and at least the commit-attempt time),
rather than exact equality with `time.Now()`.
##########
table/commit_refresh_replay_test.go:
##########
@@ -652,3 +654,149 @@ func TestRewriteRefSnapshotRequirements(t *testing.T) {
missingBranchOut := rewriteRefSnapshotRequirements(reqs,
"non-existent", fresh, nil)
assert.Equal(t, reqs, missingBranchOut)
}
+
+// A conflict on another branch must not override a staged rollback of main.
+func TestRefreshReplayKeepsRollbackBeforeAppend(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t, nil)
+ tbl, s1 := appendOnBranch(t, tbl, MainBranch, "s1")
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "s2")
+ peerUpdate(t, cat, NewSetSnapshotRefUpdate("other", s1, BranchRef, 0,
0, 0))
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.RollbackToSnapshot(s1))
+ stageFiles(t, tx, "a")
+ peerAppend(t, fs, cat, "other", "p")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ head := meta.SnapshotByName(MainBranch)
+ require.NotNil(t, head)
+ assert.Equal(t, s1, parentOf(t, meta, head.SnapshotID))
+ assert.Equal(t, []string{"a", "s1"}, branchFiles(t, fs, meta,
MainBranch))
+}
+
+// Sequence numbers and row IDs must advance across a staged rollback.
+func TestRefreshReplayKeepsCountersAcrossStagedRollback(t *testing.T) {
+ fs, cat, tbl := newExpireReplayTable(t,
iceberg.Properties{PropertyFormatVersion: "3"})
+ stage := func(tx *Transaction, name string) {
+ builder, err :=
iceberg.NewDataFileBuilder(iceberg.NewPartitionSpec(), iceberg.EntryContentData,
+ retryCleanupLocation+"/data/"+name+".parquet",
iceberg.ParquetFile, nil, nil, nil, 1, 1)
+ require.NoError(t, err)
+ require.NoError(t, tx.AddDataFiles(t.Context(),
[]iceberg.DataFile{builder.FirstRowID(0).Build()}, nil))
+ }
+ commit := func(tbl *Table, branch, name string) *Table {
+ tx, err := tbl.NewTransactionOnBranch(branch)
+ require.NoError(t, err)
+ stage(tx, name)
+ out, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ return out
+ }
+ tbl = commit(tbl, MainBranch, "s1")
+ s1 := tbl.CurrentSnapshot().SnapshotID
+ tbl = commit(tbl, MainBranch, "s2")
+ peerUpdate(t, cat, NewSetSnapshotRefUpdate("other", s1, BranchRef, 0,
0, 0))
+
+ tx := tbl.NewTransaction()
+ stage(tx, "a")
+ a := stagedHead(t, tx, MainBranch)
+ require.NoError(t, tx.RollbackToSnapshot(s1))
+ stage(tx, "b")
+ b := stagedHead(t, tx, MainBranch)
+ commit(newOCCTable(t, cat.metadata, fs, cat), "other", "p")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ snapA, snapB := meta.SnapshotByID(a), meta.SnapshotByID(b)
+ peer := meta.SnapshotByName("other")
+ require.NotNil(t, snapA)
+ require.NotNil(t, snapB)
+ assert.Equal(t, peer.SequenceNumber+1, snapA.SequenceNumber)
+ assert.Equal(t, snapA.SequenceNumber+1, snapB.SequenceNumber)
+ require.NotNil(t, peer.FirstRowID)
+ require.NotNil(t, peer.AddedRows)
+ require.NotNil(t, snapA.FirstRowID)
+ require.NotNil(t, snapA.AddedRows)
+ require.NotNil(t, snapB.FirstRowID)
+ assert.Equal(t, *peer.FirstRowID+*peer.AddedRows, *snapA.FirstRowID)
+ assert.Equal(t, *snapA.FirstRowID+*snapA.AddedRows, *snapB.FirstRowID)
+ assert.Equal(t, *snapB.FirstRowID+1, meta.NextRowID())
+ assert.Equal(t, s1, parentOf(t, meta, b))
+ assert.Equal(t, []string{"b", "s1"}, branchFiles(t, fs, meta,
MainBranch))
+}
+
+// A v1 snapshot has no sequence number, so replay onto a v2 table reports
+// a conflict.
+func TestRefreshReplayApplyFailureKeepsConflictCause(t *testing.T) {
+ _, cat, tbl := newExpireReplayTable(t,
iceberg.Properties{PropertyFormatVersion: "1"})
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "base")
+ tx := tbl.NewTransaction()
+ stageFiles(t, tx, "a")
+ peerUpdate(t, cat, NewUpgradeFormatVersionUpdate(2))
+
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, ErrTransactionUnusable)
+ require.ErrorIs(t, err, ErrCommitFailed)
+ assert.ErrorContains(t, err, "apply add-snapshot")
+}
+
+// A peer append to main more than a minute after staging moves the table's
+// last update past the staged timestamps without moving the feature branch.
+func TestRefreshReplayRenewsTimestampsOnRebuild(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, "feature", "f1")
+ tx, err := tbl.NewTransactionOnBranch("feature")
+ require.NoError(t, err)
+ stageFiles(t, tx, "a")
+ a := stagedHead(t, tx, "feature")
+ stageFiles(t, tx, "b")
+ b := stagedHead(t, tx, "feature")
+
+ time.Sleep(61 * time.Second)
+ peerAppend(t, fs, cat, MainBranch, "p")
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ meta := committed.Metadata()
+ snapA, snapB := meta.SnapshotByID(a), meta.SnapshotByID(b)
+ require.NotNil(t, snapA)
+ require.NotNil(t, snapB)
+ assert.Equal(t, time.Now().UnixMilli(), snapA.TimestampMs)
+ assert.Equal(t, snapA.TimestampMs+1, snapB.TimestampMs, "a
child is stamped after its parent")
+ assert.Equal(t, []string{"a", "b", "f1", "s1"}, branchFiles(t,
fs, meta, "feature"))
+ })
+}
+
+// A peer update to properties can make the staged timestamp too old without
+// invalidating the manifest list.
+func TestRefreshReplayRenewsTimestampOnReuse(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ _, cat, tbl := newExpireReplayTable(t, nil)
+ tbl, _ = appendOnBranch(t, tbl, MainBranch, "s1")
+ tx := tbl.NewTransaction()
+ stageFiles(t, tx, "a")
+ staged, err := tx.StagedTable()
+ require.NoError(t, err)
+ stagedSnap := staged.Metadata().SnapshotByName(MainBranch)
+
+ time.Sleep(61 * time.Second)
+ peerUpdate(t, cat,
NewSetPropertiesUpdate(iceberg.Properties{"peer": "property"}))
+ cat.failFirst = cat.attempts.Load() + 1
+
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ snap := committed.Metadata().SnapshotByID(stagedSnap.SnapshotID)
+ require.NotNil(t, snap)
+ assert.Equal(t, stagedSnap.ManifestList, snap.ManifestList,
"the staged list is reused")
+ assert.Equal(t, time.Now().UnixMilli(), snap.TimestampMs)
Review Comment:
These assertions compare snapshot timestamps to `time.Now().UnixMilli()` for
exact equality. With the new behavior (`timestamp = max(now, parent+1ms)`), the
stamped value may legitimately be `parent.TimestampMs+1` (or otherwise differ
from the `time.Now()` call made in the test), making this brittle/flaky even
under `synctest`. Recommendation: assert relative/ordering properties instead
(e.g., `>= beforeCommit`, `>= parent.TimestampMs+1`, and optionally `<=
afterCommit` when applicable), or compute an expected value using the same
`max(now, parent+1)` rule from captured timestamps.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
