zeroshade commented on code in PR #1682:
URL: https://github.com/apache/iceberg-go/pull/1682#discussion_r3808087946
##########
table/rewrite_data_files.go:
##########
@@ -433,64 +484,237 @@ 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
+ if err := t.checkNotNil(); err != nil {
+ return nil, err
+ }
+ t.mx.Lock()
+ defer t.mx.Unlock()
+ meta, err := t.txnMeta()
+ if err != nil {
+ return nil, err
+ }
+ if t.committed {
+ return nil, errors.New("transaction has already been committed")
+ }
+ result := &RewriteResult{Table: t.tbl}
+ if len(groups) == 0 {
+ return result, nil
+ }
+ 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
Review Comment:
These pre-commit exits can orphan files already written by earlier groups in
the same batch. For example, with two groups and `MaxCommits: 1`, group 1 can
write its replacement successfully and group 2 can fail here while reading its
source; `batchResults` then owns group 1's `NewDataFiles`, but this return
leaves them on storage. I reproduced that case and observed the extra generated
Parquet file remain. Please clean generated batch outputs on every known
pre-commit failure path (context cancellation, group execution, delete
collection, transaction creation, and staging), while continuing not to clean
on unknown catalog commit outcomes. `ExecuteCompactionGroup` also needs to
clean files already yielded before a later `WriteRecords` error, since its
current error return discards those paths.
--
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]
