hcrosse commented on code in PR #2354: URL: https://github.com/apache/iceberg-go/pull/2354#discussion_r4431436133
########## 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: I think we should keep the assertions on every walked ref. A peer can move any existing ref onto a snapshot that the stale plan removes, for example by rolling a branch back. The [metadata builder then drops that ref](https://github.com/apache/iceberg-go/blob/3b0a0ae83c060bb865b1851bce14e48738301e3c/table/metadata.go#L1103-L1109) when it removes the snapshot. Also, `planExpiration` removes an aged non-main ref by name. If a peer advances that ref, the stale plan still removes it. I added `TestExpireReplayKeepsAgedRefAfterPeerAdvancesIt` for that case. The cost is one retry when a peer moves a ref between refresh and submit, with replanning on the retry. ########## 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: Thanks, I took your second suggestion. `doCommit` returns `(*Table, error)` again, and the test call sites match `main`. After a successful replay, `stagedReplay` records the landed updates and a `ran` flag. `Commit` uses those updates when replay ran, otherwise the staged updates. An empty replay runs no cleanup. `doCommit` still takes `withCommitReplay`. Replay and its ref assertions depend on each attempt's refreshed table, and `doCommit` owns cleanup of manifest lists rebuilt by failed attempts. Moving replay entirely into `Commit` would require a second copy of the retry loop there. -- 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]
