laskoviymishka commented on code in PR #2354:
URL: https://github.com/apache/iceberg-go/pull/2354#discussion_r4431208345
##########
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:
Do we need this one and `TestStaleAppendAndExpireReplansAfterPeerAppend`
here as well? As I read them, they cover the same scenarios as
`TestExpireReplayCleansUpReplannedRemoval` and
`TestExpireReplayPlansWhereExpiryWasStaged`. The rollback and staged-branch
tests have no unit counterpart, so those make sense in SQLite. Unless going
through the real catalog's requirement checks is the point for these two as
well?
##########
table/expire_replay.go:
##########
@@ -0,0 +1,184 @@
+// 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
+
+ plans [][]string
+}
+
+func newStagedReplay(branch string, updates []Update, records
[]expirationRecord) *stagedReplay {
+ return &stagedReplay{
+ branch: branch,
+ records: records,
+ updates: slices.Clone(updates),
+ plans: stagedPlans(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
+ plans [][]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 {
+ plans = append(plans, 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.plans = plans
+
+ return out, orphaned, nil
+}
+
+func (r *stagedReplay) replan(b *MetadataBuilder, rec expirationRecord)
([]Update, []string, error) {
+ if err := validateGCEnabledForSnapshotExpiration(b.props); err != nil {
+ return nil, nil, fmt.Errorf("%w: %w", ErrTransactionUnusable,
err)
+ }
+
+ updates, walked, err := planExpiration(b, rec.cfg, rec.nowMs)
+ if err != nil {
+ return nil, nil, 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 stagedPlans(records []expirationRecord) [][]string {
+ var plans [][]string
+ for _, rec := range records {
+ if rec.count > 0 {
+ plans = append(plans, rec.walked)
+ }
+ }
+
+ return plans
+}
+
+// Use base ref IDs, not staged IDs. The commit branch already has an assertion
+// that the transaction rebases.
+func expirationAssertions(base Metadata, branch string, plans [][]string, reqs
[]Requirement) []Requirement {
Review Comment:
Do we need to assert every walked ref, or only the ones the removal's
outcome depends on? `planExpiration` puts every ref into `walked`
([transaction.go#L901-L903](https://github.com/apache/iceberg-go/blob/3b0a0ae83c060bb865b1851bce14e48738301e3c/table/transaction.go#L901-L903)),
so a peer moving any other ref between refresh and submit costs a retry. This
is already looser than `main`, which pinned these refs across all retries, and
Java adds no requirement for snapshot removal
([UpdateRequirements.java#L92-L115](https://github.com/apache/iceberg/blob/6275107e38cc87a7d416ef231019ceb51556f196/core/src/main/java/org/apache/iceberg/UpdateRequirements.java#L92-L115)).
##########
table/table.go:
##########
@@ -626,15 +635,22 @@ func withCommitResume(resumes int) commitOption {
return func(o *commitOpts) { o.resumes = resumes }
}
-func (t Table) doCommit(ctx context.Context, updates []Update, reqs
[]Requirement, opts ...commitOption) (*Table, error) {
+func withCommitReplay(replay *stagedReplay) commitOption {
+ return func(o *commitOpts) { o.replay = replay }
+}
+
+// Returned updates are from the successful attempt, not the staged plan.
+// If replay leaves no updates, return the refreshed table without a catalog
+// commit.
+func (t Table) doCommit(ctx context.Context, updates []Update, reqs
[]Requirement, opts ...commitOption) (*Table, []Update, error) {
Review Comment:
Could `doCommit` stay out of expiry replay altogether, with `Commit` owning
both ends? Today it takes `withCommitReplay` and returns the landed updates
beside the table.
If not, could the landed list ride on the `stagedReplay` `Commit` holds,
skipping the 24 test edits? As I read it, that needs an explicit "replay ran"
signal: with a nil check, an empty replay falls back to the staged removal and
deletes the manifest list
[`TestExpireReplayToNothingCompletesWithoutCommit`](https://github.com/apache/iceberg-go/blob/3b0a0ae83c060bb865b1851bce14e48738301e3c/table/expire_replay_test.go#L491-L519)
keeps.
--
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]
