hcrosse commented on code in PR #2354:
URL: https://github.com/apache/iceberg-go/pull/2354#discussion_r4431356878
##########
catalog/sql/sql_test.go:
##########
@@ -2044,44 +2026,216 @@ 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)
TestStaleExpireSnapshotsReplansAfterPeerAppend() {
Review Comment:
You're right! I'll clean them up :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [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]
