This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 84b84fb5f fix(table): check partial-progress batches against the
planning snapshot (#2302)
84b84fb5f is described below
commit 84b84fb5f7a5923ea909a931b2ce8ea005867be4
Author: Matt Topol <[email protected]>
AuthorDate: Wed Nov 4 10:40:13 2026 -0500
fix(table): check partial-progress batches against the planning snapshot
(#2302)
* fix(table): check partial-progress batches against the planning snapshot
Partial-progress compaction plans every group once, then commits batches
one by one, each starting from the previous batch's committed table. When a
batch's refresh-and-replay retry rebases onto a peer snapshot, that snapshot
becomes the next batch's base, so the next batch's commit-time validator never
sees it. If the peer added a position delete or deletion vector for a data file
in a later batch, that batch rewrote the file from tasks planned without the
delete, then CollectDeadPos [...]
Before rewriting a batch, run the rewrite conflict rules over every
snapshot committed since planning. A conflicting batch is recorded in
FailedGroups and counts toward MaxFailedCommits, like a failed commit, and
nothing is written for it.
* test(table): pin rejected batches against MaxFailedCommits
Address review: count planning-snapshot rejections toward
MaxFailedCommits in a test, and update the FailedGroups doc to cover
batches rejected before reaching the catalog. Rebase onto main
(AddFiles is now AddDataFilesByPath).
* refactor(table): read each snapshot once when checking partial-progress
batches
Collect the delete files committed since planning incrementally, so each
batch only reads snapshots committed since the previous batch instead of
rewalking the whole window. Document that partial-progress groups must be
planned from the transaction's table, and report batches rejected by the
planning check as rejected rather than committed.
* test(table): cover deletion vectors absorbed by an earlier
partial-progress batch
---
table/conflict_validation.go | 131 +++++++++++++++++++--------
table/rewrite_data_files.go | 114 ++++++++++++++++++++++--
table/rewrite_data_files_test.go | 186 +++++++++++++++++++++++++++++++++++++++
table/rewrite_files_test.go | 16 +++-
4 files changed, 400 insertions(+), 47 deletions(-)
diff --git a/table/conflict_validation.go b/table/conflict_validation.go
index 6ca0af9ce..9d0a4ca33 100644
--- a/table/conflict_validation.go
+++ b/table/conflict_validation.go
@@ -1044,54 +1044,115 @@ func validateNoNewDeletesForRewrittenFiles(ctx
*conflictContext, rewrittenFiles
return nil
}
- rewrittenPaths := make(map[string]struct{}, len(rewrittenFiles))
- rewrittenPartitions := make(map[string]struct{}, len(rewrittenFiles))
- for _, df := range rewrittenFiles {
- rewrittenPaths[df.FilePath()] = struct{}{}
+ rewritten, err := newRewrittenFileSet(rewrittenFiles)
+ if err != nil {
+ return err
+ }
+
+ return ctx.forEachAddedEntry(iceberg.ManifestContentDeletes, func(snap
Snapshot, e iceberg.ManifestEntry) error {
+ added, err := newAddedDelete(snap, e.DataFile())
+ if err != nil {
+ return err
+ }
+
+ return rewritten.conflict(added)
+ })
+}
+
+// addedDelete is a delete file added by a concurrent snapshot, reduced to
+// what rewrittenFileSet.conflict needs, so callers can keep it without
+// holding on to manifest entries.
+type addedDelete struct {
+ snapshotID int64
+ path string
+ specID int32
+ content iceberg.ManifestEntryContent
+ // refPath is the data file a position delete targets, or "" when the
+ // delete is partition-scoped.
+ refPath string
+ // partitionKey is set only for partition-scoped position deletes.
+ partitionKey string
+}
+
+func newAddedDelete(snap Snapshot, df iceberg.DataFile) (addedDelete, error) {
+ added := addedDelete{
+ snapshotID: snap.SnapshotID,
+ path: df.FilePath(),
+ specID: df.SpecID(),
+ content: df.ContentType(),
+ }
+ if added.content != iceberg.EntryContentPosDeletes {
+ return added, nil
+ }
+
+ if added.refPath = referencedDataFilePath(df); added.refPath != "" {
+ return added, nil
+ }
+
+ key, err := canonicalPartitionKey(df.SpecID(), dataFilePartition(df))
+ if err != nil {
+ return added, fmt.Errorf("building partition conflict key for
concurrent pos-delete %s (spec %d): %w",
+ df.FilePath(), df.SpecID(), err)
+ }
+ added.partitionKey = key
+
+ return added, nil
+}
+
+// rewrittenFileSet indexes the data files a rewrite replaces by path and by
+// (spec, partition).
+type rewrittenFileSet struct {
+ paths map[string]struct{}
+ partitions map[string]struct{}
+}
+
+func newRewrittenFileSet(files []iceberg.DataFile) (rewrittenFileSet, error) {
+ set := rewrittenFileSet{
+ paths: make(map[string]struct{}, len(files)),
+ partitions: make(map[string]struct{}, len(files)),
+ }
+ for _, df := range files {
+ set.paths[df.FilePath()] = struct{}{}
key, err := canonicalPartitionKey(df.SpecID(),
dataFilePartition(df))
if err != nil {
- return fmt.Errorf("building partition conflict key for
rewritten file %s (spec %d): %w",
+ return set, fmt.Errorf("building partition conflict key
for rewritten file %s (spec %d): %w",
df.FilePath(), df.SpecID(), err)
}
- rewrittenPartitions[key] = struct{}{}
+ set.partitions[key] = struct{}{}
}
- return ctx.forEachAddedEntry(iceberg.ManifestContentDeletes, func(snap
Snapshot, e iceberg.ManifestEntry) error {
- df := e.DataFile()
- switch df.ContentType() {
- case iceberg.EntryContentPosDeletes:
- if path := referencedDataFilePath(df); path != "" {
- if _, overlap := rewrittenPaths[path]; overlap {
- return fmt.Errorf("%w: snapshot %d
added pos-delete %s referencing rewritten file %s",
- ErrConflictingDeleteFiles,
snap.SnapshotID, df.FilePath(), path)
- }
-
- return nil
- }
+ return set, nil
+}
- key, err := canonicalPartitionKey(df.SpecID(),
dataFilePartition(df))
- if err != nil {
- return fmt.Errorf("building partition conflict
key for concurrent pos-delete %s (spec %d): %w",
- df.FilePath(), df.SpecID(), err)
- }
- if _, overlap := rewrittenPartitions[key]; overlap {
- return fmt.Errorf("%w: snapshot %d added
partition-scoped pos-delete %s overlapping rewritten partition (spec %d)",
- ErrConflictingDeleteFiles,
snap.SnapshotID, df.FilePath(), df.SpecID())
+// conflict returns ErrConflictingDeleteFiles when d could apply to a file in
+// the set.
+func (s rewrittenFileSet) conflict(d addedDelete) error {
+ switch d.content {
+ case iceberg.EntryContentPosDeletes:
+ if d.refPath != "" {
+ if _, overlap := s.paths[d.refPath]; overlap {
+ return fmt.Errorf("%w: snapshot %d added
pos-delete %s referencing rewritten file %s",
+ ErrConflictingDeleteFiles,
d.snapshotID, d.path, d.refPath)
}
return nil
+ }
- case iceberg.EntryContentEqDeletes:
- // Conservative: any concurrent eq-delete is a conflict
- // for a rewrite, because the predicate could cover a
- // rewritten file. Follow-up: accept partition-overlap
- // hints to narrow this.
- return fmt.Errorf("%w: snapshot %d added equality
delete %s during rewrite",
- ErrConflictingDeleteFiles, snap.SnapshotID,
df.FilePath())
+ if _, overlap := s.partitions[d.partitionKey]; overlap {
+ return fmt.Errorf("%w: snapshot %d added
partition-scoped pos-delete %s overlapping rewritten partition (spec %d)",
+ ErrConflictingDeleteFiles, d.snapshotID,
d.path, d.specID)
}
- return nil
- })
+ case iceberg.EntryContentEqDeletes:
+ // Conservative: any concurrent eq-delete is a conflict
+ // for a rewrite, because the predicate could cover a
+ // rewritten file. Follow-up: accept partition-overlap
+ // hints to narrow this.
+ return fmt.Errorf("%w: snapshot %d added equality delete %s
during rewrite",
+ ErrConflictingDeleteFiles, d.snapshotID, d.path)
+ }
+
+ return nil
}
// referencedDataFilePath resolves the single data file a pos-delete
diff --git a/table/rewrite_data_files.go b/table/rewrite_data_files.go
index e27933545..7d12e574f 100644
--- a/table/rewrite_data_files.go
+++ b/table/rewrite_data_files.go
@@ -74,8 +74,9 @@ type RewriteResult struct {
// partial-progress mode.
CompletedGroups []CompactionGroupResult
- // FailedGroups contains groups whose catalog commits failed. A partial
- // rewrite can return both completed and failed groups.
+ // FailedGroups contains groups whose batch failed to commit, or was
+ // rejected because a delete for one of its files was committed after
+ // planning. A partial rewrite can return both completed and failed
groups.
FailedGroups []CompactionGroupFailure
}
@@ -184,6 +185,11 @@ type RewriteDataFilesOptions struct {
// PartialProgress, when true, commits non-empty groups in durable
catalog
// batches. When false (the default), every group is staged in a single
// atomic rewrite snapshot.
+ //
+ // Partial progress checks each batch against the delete files committed
+ // since the transaction's starting snapshot, so the groups must be
planned
+ // from the transaction's table. Deletes committed between an older
+ // planning snapshot and that starting snapshot are not detected.
PartialProgress bool
// MaxCommits bounds the number of catalog snapshots produced in
@@ -192,10 +198,12 @@ type RewriteDataFilesOptions struct {
// effect for atomic rewrites.
MaxCommits int
- // MaxFailedCommits bounds catalog commit failures in partial-progress
mode.
- // Zero (the default) and negative values allow unlimited failures.
Failures
- // within the bound are returned in RewriteResult.FailedGroups without
making
- // the operation return an error.
+ // MaxFailedCommits bounds failed batches in partial-progress mode:
catalog
+ // commit failures, and batches rejected because a delete file for one
of
+ // their data files was committed after the groups were planned. Zero
(the
+ // default) and negative values allow unlimited failures. Failures
within
+ // the bound are returned in RewriteResult.FailedGroups without making
the
+ // operation return an error.
MaxFailedCommits int
// SnapshotProps are added to the rewrite snapshot's summary.
@@ -670,7 +678,9 @@ func allTasksHaveRowLineage(tasks []FileScanTask) bool {
// 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.
+// rolling back snapshots already committed for earlier batches. Each batch is
+// also checked against every delete committed since the groups were planned,
+// since an earlier batch's retry can make those snapshots part of its base.
func (t *Transaction) rewriteDataFilesPartial(ctx context.Context, groups
[]CompactionTaskGroup, opts RewriteDataFilesOptions) (*RewriteResult, error) {
if err := t.checkNotNil(); err != nil {
return nil, err
@@ -718,6 +728,7 @@ func (t *Transaction) rewriteDataFilesPartial(ctx
context.Context, groups []Comp
groupsPerCommit := (len(pendingGroups)-1)/maxCommits + 1
props := maps.Clone(opts.SnapshotProps)
current := t.tbl
+ sincePlanning := deletesSincePlanning{checked: t.tbl.Metadata(),
branch: t.branch}
failedCommits := 0
for batchStart := 0; batchStart < len(pendingGroups); batchStart +=
groupsPerCommit {
@@ -725,6 +736,7 @@ func (t *Transaction) rewriteDataFilesPartial(ctx
context.Context, groups []Comp
return result, err
}
+ batchNum := batchStart/groupsPerCommit + 1
batchEnd := min(batchStart+groupsPerCommit, len(pendingGroups))
batchGroups := pendingGroups[batchStart:batchEnd]
batchResults := make([]CompactionGroupResult, 0,
len(batchGroups))
@@ -734,6 +746,33 @@ func (t *Transaction) rewriteDataFilesPartial(ctx
context.Context, groups []Comp
if err != nil {
return result, fmt.Errorf("open table IO for partial
rewrite batch: %w", err)
}
+
+ // The groups were planned against the starting table, but an
earlier
+ // batch's refresh-and-replay retry can commit on top of peer
+ // snapshots. Those snapshots then form this batch's base, so
its
+ // commit-time validator never sees them, while its planned
tasks lack
+ // any delete files they added. Rewriting anyway would write
the deleted
+ // rows back and then drop those deletes as dead.
+ if err := sincePlanning.validate(current.Metadata(), fs,
batchGroups); err != nil {
+ if !errors.Is(err, ErrConflictingDeleteFiles) {
+ return result, fmt.Errorf("validate partial
rewrite batch %d against its planning snapshot: %w", batchNum, err)
+ }
+
+ failedCommits++
+ for _, group := range batchGroups {
+ result.FailedGroups =
append(result.FailedGroups, CompactionGroupFailure{
+ PartitionKey: group.PartitionKey,
+ Err: err,
+ })
+ }
+ if maxFailedCommits > 0 && failedCommits >
maxFailedCommits {
+ return result, fmt.Errorf("reject partial
rewrite batch %d: %w (maximum failed commits reached)",
+ batchNum, err)
+ }
+
+ continue
+ }
+
cleanupBatch := func(cause error, extras
...CompactionGroupResult) error {
results := batchResults
if len(extras) > 0 {
@@ -850,7 +889,7 @@ func (t *Transaction) rewriteDataFilesPartial(ctx
context.Context, groups []Comp
}
if maxFailedCommits > 0 && failedCommits >
maxFailedCommits {
return result, fmt.Errorf("commit partial
rewrite batch %d: %w (maximum failed commits reached)",
- batchStart/groupsPerCommit+1, err)
+ batchNum, err)
}
continue
@@ -947,6 +986,65 @@ func rewriteValidatorWithReferencedDataFiles(
}
}
+// deletesSincePlanning collects the delete files committed on a branch since
+// the groups were planned. Each validate call reads only the snapshots
+// committed since the previous call, so a partial rewrite reads every
+// snapshot's manifests once rather than once per batch.
+type deletesSincePlanning struct {
+ // checked is the metadata the collected deletes run up to. It starts as
+ // the metadata the groups were planned from.
+ checked Metadata
+ branch string
+ deletes []addedDelete
+}
+
+// validate collects the deletes committed between d.checked and current,
+// then returns ErrConflictingDeleteFiles when any delete committed since
+// planning could apply to a data file in groups.
+func (d *deletesSincePlanning) validate(current Metadata, fs iceio.IO, groups
[]CompactionTaskGroup) error {
+ // The rewrite validator binds no filters, so case sensitivity is moot.
+ cc, err := newConflictContext(d.checked, current, d.branch, fs, true)
+ if err != nil {
+ return err
+ }
+ deletes := d.deletes
+ err = cc.forEachAddedEntry(iceberg.ManifestContentDeletes, func(snap
Snapshot, e iceberg.ManifestEntry) error {
+ added, err := newAddedDelete(snap, e.DataFile())
+ if err != nil {
+ return err
+ }
+ deletes = append(deletes, added)
+
+ return nil
+ })
+ if err != nil {
+ return err
+ }
+ d.checked, d.deletes = current, deletes
+
+ if len(d.deletes) == 0 {
+ return nil
+ }
+
+ var files []iceberg.DataFile
+ for _, group := range groups {
+ for _, task := range group.Tasks {
+ files = append(files, task.File)
+ }
+ }
+ rewritten, err := newRewrittenFileSet(files)
+ if err != nil {
+ return err
+ }
+ for _, added := range d.deletes {
+ if err := rewritten.conflict(added); err != nil {
+ return err
+ }
+ }
+
+ return nil
+}
+
// CollectSafePositionDeletes returns position delete files from the
// given tasks that are safe to remove during compaction.
//
diff --git a/table/rewrite_data_files_test.go b/table/rewrite_data_files_test.go
index ef2e168ae..4d91bd22b 100644
--- a/table/rewrite_data_files_test.go
+++ b/table/rewrite_data_files_test.go
@@ -23,6 +23,7 @@ import (
"fmt"
"os"
"path/filepath"
+ "slices"
"strings"
"testing"
@@ -34,6 +35,7 @@ import (
iceio "github.com/apache/iceberg-go/io"
"github.com/apache/iceberg-go/table"
"github.com/apache/iceberg-go/table/compaction"
+ "github.com/apache/iceberg-go/table/dv"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -791,6 +793,190 @@ func
TestRewriteDataFiles_PartialProgressRetainsSharedPositionDeleteAfterFailure
"generated files from tolerated failed batches must be removed")
}
+// TestRewriteDataFiles_PartialProgressRejectsDeleteAbsorbedByEarlierBatch
+// lands a peer delete for a batch-2 data file just before batch 1 commits.
+// Batch 1's retry rebases onto the peer snapshot without conflict, since the
+// delete targets a file batch 1 does not rewrite. Batch 2's tasks were planned
+// before the delete existed and its transaction starts from batch 1's table,
+// so only a check back to the planning snapshot can stop it from rewriting the
+// file without the delete and then dropping the delete as dead, which would
+// bring the deleted row back. File-scoped position deletes and deletion
+// vectors take different paths through newAddedDelete, so both are covered.
+func TestRewriteDataFiles_PartialProgressRejectsDeleteAbsorbedByEarlierBatch(t
*testing.T) {
+ for _, tc := range []struct {
+ name string
+ kind absorbedDeleteKind
+ }{
+ {name: "position delete", kind: absorbedPositionDelete},
+ {name: "deletion vector", kind: absorbedDeletionVector},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ tbl, cat, groups, deletePaths :=
newAbsorbedDeleteFixture(t, tc.kind, 2, 1)
+
+ beforeFiles := parquetFiles(t, tbl.Location())
+ tx := tbl.NewTransaction()
+ result, err := tx.RewriteDataFiles(t.Context(), groups,
table.RewriteDataFilesOptions{
+ PartialProgress: true,
+ MaxCommits: 2,
+ })
+ require.NoError(t, err)
+ require.Nil(t, cat.beforeCommit, "the peer commit must
have landed during the rewrite")
+
+ require.Len(t, result.CompletedGroups, 1)
+ assert.Equal(t, "absorbed-group-0",
result.CompletedGroups[0].PartitionKey)
+ require.Len(t, result.FailedGroups, 1)
+ assert.Equal(t, "absorbed-group-1",
result.FailedGroups[0].PartitionKey)
+ assert.ErrorIs(t, result.FailedGroups[0].Err,
table.ErrConflictingDeleteFiles)
+ assert.Zero(t, result.RemovedPositionDeleteFiles)
+
+ final, err := cat.LoadTable(t.Context(),
tbl.Identifier())
+ require.NoError(t, err)
+ assertRowCount(t, final, 3)
+ finalTasks, err := final.Scan().PlanFiles(t.Context())
+ require.NoError(t, err)
+ var liveDeletes []string
+ for _, task := range finalTasks {
+ for _, df := range task.DeleteFiles {
+ liveDeletes = append(liveDeletes,
df.FilePath())
+ }
+ for _, df := range task.DeletionVectorFiles {
+ liveDeletes = append(liveDeletes,
df.FilePath())
+ }
+ }
+ assert.Equal(t, deletePaths, liveDeletes, "the peer's
delete must stay live")
+
+ // The rejected batch is detected before it writes
anything.
+ expectedFiles := beforeFiles
+ if tc.kind == absorbedPositionDelete {
+ expectedFiles =
append(slices.Clone(deletePaths), beforeFiles...)
+ }
+ for _, file := range
result.CompletedGroups[0].NewDataFiles {
+ expectedFiles = append(expectedFiles,
file.FilePath())
+ }
+ assert.ElementsMatch(t, expectedFiles, parquetFiles(t,
tbl.Location()),
+ "the rejected batch must not leave compacted
outputs behind")
+ })
+ }
+}
+
+//
TestRewriteDataFiles_PartialProgressRejectedBatchesCountTowardMaxFailedCommits
+// rejects two batches through the planning-snapshot check and asserts that
+// they count toward MaxFailedCommits like catalog commit failures do.
+func
TestRewriteDataFiles_PartialProgressRejectedBatchesCountTowardMaxFailedCommits(t
*testing.T) {
+ tbl, cat, groups, _ := newAbsorbedDeleteFixture(t,
absorbedPositionDelete, 3, 1, 2)
+
+ tx := tbl.NewTransaction()
+ result, err := tx.RewriteDataFiles(t.Context(), groups,
table.RewriteDataFilesOptions{
+ PartialProgress: true,
+ MaxCommits: 3,
+ MaxFailedCommits: 1,
+ })
+ require.Nil(t, cat.beforeCommit, "the peer commit must have landed
during the rewrite")
+ require.ErrorIs(t, err, table.ErrConflictingDeleteFiles)
+ assert.ErrorContains(t, err, "maximum failed commits reached")
+ require.Len(t, result.CompletedGroups, 1)
+ assert.Equal(t, "absorbed-group-0",
result.CompletedGroups[0].PartitionKey)
+ require.Len(t, result.FailedGroups, 2)
+}
+
+type absorbedDeleteKind int
+
+const (
+ absorbedPositionDelete absorbedDeleteKind = iota
+ absorbedDeletionVector
+)
+
+// newAbsorbedDeleteFixture writes n two-row data files, each committed on its
+// own, and returns one compaction group per file in path order. Before the
+// rewrite's first catalog commit, a peer commits one delete of row 0 for each
+// data file index in deleteTargets: a file-scoped position delete, or, on a v3
+// table, a deletion vector. Index 0's batch commits first, so targets must be
+// greater than 0 to be absorbed by an earlier batch. The returned paths are
+// the peer's delete files.
+func newAbsorbedDeleteFixture(t *testing.T, kind absorbedDeleteKind, n int,
deleteTargets ...int) (*table.Table, *concurrentTestCatalog,
[]table.CompactionTaskGroup, []string) {
+ t.Helper()
+ version := 2
+ if kind == absorbedDeletionVector {
+ version = 3
+ }
+ tbl, cat := newConcurrentRewriteTestTableVersion(t, version)
+ arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
+ require.NoError(t, err)
+
+ paths := make([]string, n)
+ for i := range paths {
+ paths[i] = tbl.Location() +
fmt.Sprintf("/data/absorbed-%d.parquet", i)
+ writeParquetFile(t, paths[i], arrowSc, fmt.Sprintf(
+ `[{"id": %d, "data": "a"}, {"id": %d, "data": "b"}]`,
i*2+1, i*2+2))
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFilesByPath(t.Context(),
[]string{paths[i]}, nil, false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ }
+
+ tasks, err := tbl.Scan().PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, n)
+ taskByPath := make(map[string]table.FileScanTask, len(tasks))
+ for _, task := range tasks {
+ taskByPath[task.File.FilePath()] = task
+ }
+ groups := make([]table.CompactionTaskGroup, 0, n)
+ for i, path := range paths {
+ task := taskByPath[path]
+ groups = append(groups, table.CompactionTaskGroup{
+ PartitionKey: fmt.Sprintf("absorbed-group-%d", i),
+ Tasks: []table.FileScanTask{task},
+ TotalSizeBytes: task.File.FileSizeBytes(),
+ })
+ }
+
+ deletePaths := make([]string, len(deleteTargets))
+ for i, target := range deleteTargets {
+ ext := "parquet"
+ if kind == absorbedDeletionVector {
+ ext = "puffin"
+ }
+ deletePaths[i] = tbl.Location() +
fmt.Sprintf("/data/absorbed-peer-delete-%d.%s", target, ext)
+ }
+ cat.beforeCommit = func() {
+ peer, err := cat.LoadTable(t.Context(), tbl.Identifier())
+ require.NoError(t, err)
+ peerTx := peer.NewTransaction()
+ delta := peerTx.NewRowDelta(nil)
+ for i, target := range deleteTargets {
+ if kind == absorbedDeletionVector {
+ // A DV names its data file in
referenced_data_file, so the
+ // earlier batch's validator ignores it.
+ w := dv.NewDVWriter(iceio.LocalFS{},
unpartitionedSpecByID)
+ require.NoError(t, w.Add(paths[target],
[]int64{0}, 0, nil))
+ dvFiles, err := w.Flush(t.Context(),
deletePaths[i])
+ require.NoError(t, err)
+ delta = delta.AddDeletes(dvFiles...)
+
+ continue
+ }
+
+ // Equal file_path bounds make the delete file-scoped,
so the
+ // earlier batch's validator ignores it.
+ writeParquetFile(t, deletePaths[i],
table.PositionalDeleteArrowSchema,
+ fmt.Sprintf(`[{"file_path": "%s", "pos": 0}]`,
paths[target]))
+ builder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec,
iceberg.EntryContentPosDeletes,
+ deletePaths[i], iceberg.ParquetFile, nil, nil,
nil, 1, 128)
+ require.NoError(t, err)
+ bound := map[int][]byte{iceberg.DeleteFilePathFieldID:
[]byte(paths[target])}
+ delta =
delta.AddDeletes(builder.LowerBoundValues(bound).UpperBoundValues(bound).Build())
+ }
+ require.NoError(t, delta.Commit(t.Context()))
+ _, err = peerTx.Commit(t.Context())
+ require.NoError(t, err)
+ }
+
+ return tbl, cat, groups, deletePaths
+}
+
func TestRewriteDataFiles_PartialProgressCleansOutputsBeforeCommit(t
*testing.T) {
tbl, _ := newPartialProgressTestTable(t)
arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
diff --git a/table/rewrite_files_test.go b/table/rewrite_files_test.go
index df2c512e8..7f0acc1e0 100644
--- a/table/rewrite_files_test.go
+++ b/table/rewrite_files_test.go
@@ -48,11 +48,15 @@ import (
// whose AssertRefSnapshotID points at a stale snapshot fails its
// first attempt and triggers refresh-and-replay. attempts counts
// CommitTable invocations so tests can prove the retry boundary.
+// beforeCommit, when set, runs once at the start of the next
+// CommitTable, so a test can land a peer commit inside an operation
+// that commits more than once.
type concurrentTestCatalog struct {
- metadata table.Metadata
- location string
- fsF table.FSysF
- attempts atomic.Int32
+ metadata table.Metadata
+ location string
+ fsF table.FSysF
+ attempts atomic.Int32
+ beforeCommit func()
}
func (c *concurrentTestCatalog) LoadTable(_ context.Context, ident
table.Identifier) (*table.Table, error) {
@@ -61,6 +65,10 @@ func (c *concurrentTestCatalog) LoadTable(_ context.Context,
ident table.Identif
func (c *concurrentTestCatalog) CommitTable(_ context.Context, _
table.Identifier, reqs []table.Requirement, updates []table.Update)
(table.Metadata, string, error) {
c.attempts.Add(1)
+ if hook := c.beforeCommit; hook != nil {
+ c.beforeCommit = nil
+ hook()
+ }
for _, req := range reqs {
if err := req.Validate(c.metadata); err != nil {
return nil, "", fmt.Errorf("%w: %w",
table.ErrCommitFailed, err)
