zeroshade commented on code in PR #1682:
URL: https://github.com/apache/iceberg-go/pull/1682#discussion_r3807606875
##########
table/rewrite_data_files.go:
##########
@@ -433,64 +488,204 @@ func allTasksHaveRowLineage(tasks []FileScanTask) bool {
return true
}
-// rewriteDataFilesPartial stages each group as its own rewrite
-// snapshot via [Transaction.ReplaceFiles] directly. Per-group staging
-// lets a mid-loop write failure leave already-staged groups on the
-// transaction; the catalog still receives them at
-// [Transaction.Commit] time.
-//
-// Validator registration is coalesced: a single [rewriteValidator]
-// covering every rewritten path across all groups is registered once,
-// after the loop, instead of one per group. The transaction's
-// validator list otherwise grows linearly with the group count, and
-// each entry independently walks the concurrent-snapshot set on
-// refresh-replay — the union walk subsumes them.
+// rewriteDataFilesPartial executes groups and commits them in durable batches.
+// A batch is the atomic unit for both the MaxCommits bound and delete cleanup:
+// classic position deletes are rechecked against the union of every old data
+// file in the batch before they are removed. A later batch can fail without
+// rolling back snapshots already committed for earlier batches.
func (t *Transaction) rewriteDataFilesPartial(ctx context.Context, groups
[]CompactionTaskGroup, opts RewriteDataFilesOptions) (*RewriteResult, error) {
- result := &RewriteResult{}
- props := maps.Clone(opts.SnapshotProps)
- var allRewritten []iceberg.DataFile
+ result := &RewriteResult{Table: t.tbl}
+ meta, err := t.txnMeta()
+ if err != nil {
+ return nil, err
+ }
+ if len(meta.updates) > 0 || len(t.reqs) > 0 || len(t.validators) > 0 {
+ return nil, fmt.Errorf("%w: partial progress requires a fresh
transaction",
+ ErrInvalidOperation)
+ }
+ maxCommits := opts.MaxCommits
+ if maxCommits == 0 {
+ maxCommits = 10
+ }
+ if maxCommits < 0 {
+ return nil, fmt.Errorf("%w: MaxCommits must be non-negative",
ErrInvalidOperation)
+ }
+ maxFailedCommits := opts.MaxFailedCommits
+ pendingGroups := make([]CompactionTaskGroup, 0, len(groups))
for _, group := range groups {
+ if len(group.Tasks) > 0 {
+ pendingGroups = append(pendingGroups, group)
+ }
+ }
+ if len(pendingGroups) == 0 {
+ return result, nil
+ }
+
+ // Match Iceberg's action semantics: MaxCommits is a bound on snapshots,
+ // not on the number of groups processed. Distribute all groups across
at
+ // most MaxCommits batches.
+ groupsPerCommit := (len(pendingGroups)-1)/maxCommits + 1
+ props := maps.Clone(opts.SnapshotProps)
+ current := t.tbl
+ failedCommits := 0
+
+ for batchStart := 0; batchStart < len(pendingGroups); batchStart +=
groupsPerCommit {
if err := ctx.Err(); err != nil {
return result, err
}
- if len(group.Tasks) == 0 {
+ batchEnd := min(batchStart+groupsPerCommit, len(pendingGroups))
+ batchGroups := pendingGroups[batchStart:batchEnd]
+ batchResults := make([]CompactionGroupResult, 0,
len(batchGroups))
+ rewrittenPaths := make(map[string]struct{})
+ rewrittenFiles := make([]iceberg.DataFile, 0)
+
+ for _, group := range batchGroups {
+ if err := ctx.Err(); err != nil {
+ return result, err
+ }
+
+ gr, err := ExecuteCompactionGroup(ctx, current, group,
opts.GroupOptions...)
+ if err != nil {
+ return result, err
+ }
+
+ if len(gr.OldDataFiles) == 0 && len(gr.NewDataFiles) ==
0 {
+ continue
+ }
+ batchResults = append(batchResults, gr)
+ for _, df := range gr.OldDataFiles {
+ if _, ok := rewrittenPaths[df.FilePath()]; ok {
+ continue
+ }
+ rewrittenPaths[df.FilePath()] = struct{}{}
+ rewrittenFiles = append(rewrittenFiles, df)
+ }
+ }
+
+ if len(batchResults) == 0 {
continue
}
- gr, err := ExecuteCompactionGroup(ctx, t.tbl, group,
opts.GroupOptions...)
+ fs, err := current.fsF(ctx)
if err != nil {
- return result, err
+ return result, fmt.Errorf("open table IO for partial
rewrite batch: %w", err)
+ }
+ deadPositionDeletes, err := CollectDeadPositionDeletes(
Review Comment:
**Blocking:** This always resolves `main`, even when `t.branch` targets an
existing divergent branch. `deletesToRemove` can therefore be calculated from
main’s survivors and remove a shared partition-scoped position delete that is
still needed by a branch-only file, resurrecting rows on that branch. Please
resolve the snapshot with the same branch-aware semantics as the child
transaction (including the main fallback only when creating a new branch), and
add a divergent-main/branch regression test.
##########
cmd/iceberg/compact.go:
##########
@@ -165,9 +165,11 @@ func compactRun(ctx context.Context, output Output, tbl
*table.Table, plan compa
os.Exit(1)
}
- if _, err := tx.Commit(ctx); err != nil {
- output.Error(fmt.Errorf("commit failed: %w", err))
- os.Exit(1)
+ if !cfg.partialProgress {
Review Comment:
`RewriteDataFiles` can return `nil` with `result.FailedGroups` populated
while failures remain within the configured limit. This path then skips the
parent commit, ignores those failures, and later prints `Done`; with the
current default (`MaxFailedCommits == 0`, unlimited), even every batch can fail
while the command exits successfully. Please report/check failed groups and
return a nonzero status, or expose an explicit failure-tolerance contract in
the CLI instead of silently succeeding.
##########
table/rewrite_data_files.go:
##########
@@ -433,64 +488,204 @@ func allTasksHaveRowLineage(tasks []FileScanTask) bool {
return true
}
-// rewriteDataFilesPartial stages each group as its own rewrite
-// snapshot via [Transaction.ReplaceFiles] directly. Per-group staging
-// lets a mid-loop write failure leave already-staged groups on the
-// transaction; the catalog still receives them at
-// [Transaction.Commit] time.
-//
-// Validator registration is coalesced: a single [rewriteValidator]
-// covering every rewritten path across all groups is registered once,
-// after the loop, instead of one per group. The transaction's
-// validator list otherwise grows linearly with the group count, and
-// each entry independently walks the concurrent-snapshot set on
-// refresh-replay — the union walk subsumes them.
+// rewriteDataFilesPartial executes groups and commits them in durable batches.
+// A batch is the atomic unit for both the MaxCommits bound and delete cleanup:
+// classic position deletes are rechecked against the union of every old data
+// file in the batch before they are removed. A later batch can fail without
+// rolling back snapshots already committed for earlier batches.
func (t *Transaction) rewriteDataFilesPartial(ctx context.Context, groups
[]CompactionTaskGroup, opts RewriteDataFilesOptions) (*RewriteResult, error) {
- result := &RewriteResult{}
- props := maps.Clone(opts.SnapshotProps)
- var allRewritten []iceberg.DataFile
+ result := &RewriteResult{Table: t.tbl}
+ meta, err := t.txnMeta()
+ if err != nil {
+ return nil, err
+ }
+ if len(meta.updates) > 0 || len(t.reqs) > 0 || len(t.validators) > 0 {
+ return nil, fmt.Errorf("%w: partial progress requires a fresh
transaction",
+ ErrInvalidOperation)
+ }
+ maxCommits := opts.MaxCommits
+ if maxCommits == 0 {
+ maxCommits = 10
+ }
+ if maxCommits < 0 {
+ return nil, fmt.Errorf("%w: MaxCommits must be non-negative",
ErrInvalidOperation)
+ }
+ maxFailedCommits := opts.MaxFailedCommits
+ pendingGroups := make([]CompactionTaskGroup, 0, len(groups))
for _, group := range groups {
+ if len(group.Tasks) > 0 {
+ pendingGroups = append(pendingGroups, group)
+ }
+ }
+ if len(pendingGroups) == 0 {
+ return result, nil
+ }
+
+ // Match Iceberg's action semantics: MaxCommits is a bound on snapshots,
+ // not on the number of groups processed. Distribute all groups across
at
+ // most MaxCommits batches.
+ groupsPerCommit := (len(pendingGroups)-1)/maxCommits + 1
+ props := maps.Clone(opts.SnapshotProps)
+ current := t.tbl
+ failedCommits := 0
+
+ for batchStart := 0; batchStart < len(pendingGroups); batchStart +=
groupsPerCommit {
if err := ctx.Err(); err != nil {
return result, err
}
- if len(group.Tasks) == 0 {
+ batchEnd := min(batchStart+groupsPerCommit, len(pendingGroups))
+ batchGroups := pendingGroups[batchStart:batchEnd]
+ batchResults := make([]CompactionGroupResult, 0,
len(batchGroups))
+ rewrittenPaths := make(map[string]struct{})
+ rewrittenFiles := make([]iceberg.DataFile, 0)
+
+ for _, group := range batchGroups {
+ if err := ctx.Err(); err != nil {
+ return result, err
+ }
+
+ gr, err := ExecuteCompactionGroup(ctx, current, group,
opts.GroupOptions...)
+ if err != nil {
+ return result, err
+ }
+
+ if len(gr.OldDataFiles) == 0 && len(gr.NewDataFiles) ==
0 {
+ continue
+ }
+ batchResults = append(batchResults, gr)
+ for _, df := range gr.OldDataFiles {
+ if _, ok := rewrittenPaths[df.FilePath()]; ok {
+ continue
+ }
+ rewrittenPaths[df.FilePath()] = struct{}{}
+ rewrittenFiles = append(rewrittenFiles, df)
+ }
+ }
+
+ if len(batchResults) == 0 {
continue
}
- gr, err := ExecuteCompactionGroup(ctx, t.tbl, group,
opts.GroupOptions...)
+ fs, err := current.fsF(ctx)
if err != nil {
- return result, err
+ return result, fmt.Errorf("open table IO for partial
rewrite batch: %w", err)
+ }
+ deadPositionDeletes, err := CollectDeadPositionDeletes(
+ ctx, fs, current.CurrentSnapshot(), rewrittenPaths)
+ if err != nil {
+ return result, fmt.Errorf("collect dead position
deletes for partial rewrite batch: %w", err)
}
- if len(gr.OldDataFiles) == 0 && len(gr.NewDataFiles) == 0 {
- continue
+ // Deletion vectors are one-to-one with their referenced data
file, so
+ // task-level results are sufficient. Deduplicate by reference
because
+ // ReplaceFiles rejects multiple DVs for one data file.
+ safeDVs := make([]iceberg.DataFile, 0)
+ seenDVRefs := make(map[string]struct{})
+ for _, gr := range batchResults {
+ for _, dv := range gr.SafeDeletionVectors {
+ ref := dv.ReferencedDataFile()
+ if ref == nil {
+ continue
+ }
+ if _, ok := seenDVRefs[*ref]; ok {
+ continue
+ }
+ seenDVRefs[*ref] = struct{}{}
+ safeDVs = append(safeDVs, dv)
+ }
}
+ deletesToRemove := append(deadPositionDeletes, safeDVs...)
- deletesToRemove := append(slices.Clone(gr.SafePosDeletes),
gr.SafeDeletionVectors...)
- if err := t.ReplaceFiles(ctx, gr.OldDataFiles, gr.NewDataFiles,
deletesToRemove,
+ groupTxn, err :=
current.NewTransactionOnBranchWithError(t.branch)
+ if err != nil {
+ return result, fmt.Errorf("create transaction for
partial rewrite batch: %w", err)
+ }
+ newDataFiles := make([]iceberg.DataFile, 0)
+ oldDataFiles := make([]iceberg.DataFile, 0, len(rewrittenFiles))
+ for _, gr := range batchResults {
+ oldDataFiles = append(oldDataFiles, gr.OldDataFiles...)
+ newDataFiles = append(newDataFiles, gr.NewDataFiles...)
+ }
+ if err := groupTxn.ReplaceFiles(ctx, oldDataFiles,
newDataFiles, deletesToRemove,
props, withRewriteSemantics()); err != nil {
- return result, fmt.Errorf("commit compaction group %q:
%w", group.PartitionKey, err)
+ return result, fmt.Errorf("stage partial rewrite batch:
%w", err)
}
+ groupTxn.addValidator(rewriteValidator(rewrittenFiles))
- allRewritten = append(allRewritten, gr.OldDataFiles...)
- accumulateGroupMetrics(result, gr)
- }
+ next, err := groupTxn.Commit(ctx)
+ if err != nil {
+ if next != nil {
+ // A non-nil table means the catalog commit
succeeded even if a
+ // post-commit hook returned an error. Keep the
committed state and
+ // stop before planning another batch from an
error-bearing result.
+ recordCommittedRewriteBatch(result, next,
batchResults, deadPositionDeletes, safeDVs, ¤t, t)
+
+ return result, err
+ }
- if len(allRewritten) > 0 {
- t.addValidator(rewriteValidator(allRewritten))
+ // ErrCommitFailed is the only error that proves the
catalog did not
+ // commit. Every other error leaves commit state
unknown, so continuing
+ // with the old table could apply later batches on
stale state.
+ if !errors.Is(err, ErrCommitFailed) {
+ t.committed = true
+
+ return result, err
+ }
+
+ failedCommits++
Review Comment:
At this point every `NewDataFiles` output for the batch has already been
written. `ErrCommitFailed` is the known-not-committed case, but this path
records the failure and continues without deleting those outputs;
`FailedGroups` also does not expose their paths to the caller. Repeated
tolerated failures therefore leak full unreferenced compacted files. Please
remove the batch’s generated files before continuing and add a cleanup
regression test.
##########
table/rewrite_data_files.go:
##########
@@ -433,64 +488,204 @@ func allTasksHaveRowLineage(tasks []FileScanTask) bool {
return true
}
-// rewriteDataFilesPartial stages each group as its own rewrite
-// snapshot via [Transaction.ReplaceFiles] directly. Per-group staging
-// lets a mid-loop write failure leave already-staged groups on the
-// transaction; the catalog still receives them at
-// [Transaction.Commit] time.
-//
-// Validator registration is coalesced: a single [rewriteValidator]
-// covering every rewritten path across all groups is registered once,
-// after the loop, instead of one per group. The transaction's
-// validator list otherwise grows linearly with the group count, and
-// each entry independently walks the concurrent-snapshot set on
-// refresh-replay — the union walk subsumes them.
+// rewriteDataFilesPartial executes groups and commits them in durable batches.
+// A batch is the atomic unit for both the MaxCommits bound and delete cleanup:
+// classic position deletes are rechecked against the union of every old data
+// file in the batch before they are removed. A later batch can fail without
+// rolling back snapshots already committed for earlier batches.
func (t *Transaction) rewriteDataFilesPartial(ctx context.Context, groups
[]CompactionTaskGroup, opts RewriteDataFilesOptions) (*RewriteResult, error) {
- result := &RewriteResult{}
- props := maps.Clone(opts.SnapshotProps)
- var allRewritten []iceberg.DataFile
+ result := &RewriteResult{Table: t.tbl}
+ meta, err := t.txnMeta()
+ if err != nil {
+ return nil, err
+ }
+ if len(meta.updates) > 0 || len(t.reqs) > 0 || len(t.validators) > 0 {
Review Comment:
This freshness check does not include `t.committed`. After the first
successful partial batch, `recordCommittedRewriteBatch` marks the parent
terminal, but a second call to `RewriteDataFiles(...PartialProgress: true)`
proceeds and commits more child transactions. I reproduced this directly: the
second call returned `nil`. Please reject an already committed/unknown-state
parent before performing any rewrite I/O and cover the reuse case.
--
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]
