hcrosse commented on code in PR #2468:
URL: https://github.com/apache/iceberg-go/pull/2468#discussion_r4516689843
##########
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:
The test runs inside `synctest`, so `time.Now()` is the same virtual instant
the replay used, and the parent is more than a minute older, so `max(now,
parent+1)` is exactly `now`. The parent floor is asserted on the next line with
the exact value `snapA.TimestampMs+1`, and an ordering check would be weaker
than either.
##########
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:
The test runs inside `synctest`, so `time.Now()` is the same virtual instant
the replay used, and the parent is more than a minute older, so `max(now,
parent+1)` is exactly `now`. An ordering check would also pass if renewal used
a stale clock reading, which exact equality rules out.
##########
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 runs inside `synctest`, and the retry backoff of at least 1 ms puts
`now` at or past `p.TimestampMs+1`, so `max(now, parent+1)` always equals
`now`. The clock is virtual, so the result does not depend on wall time or
scheduling.
--
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]
