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 06cfb8cd4 fix(rowdelta): accept deletes for data files added in the
same commit (#2194)
06cfb8cd4 is described below
commit 06cfb8cd48e3820f17c60dc1b5bf549f0c5f6968
Author: Badal Prasad Singh <[email protected]>
AuthorDate: Tue Oct 20 22:43:26 2026 +0530
fix(rowdelta): accept deletes for data files added in the same commit
(#2194)
* fix: validation
Signed-off-by: badalprasadsingh <[email protected]>
* chore: minor
Signed-off-by: badalprasadsingh <[email protected]>
* chore: minor
Signed-off-by: badalprasadsingh <[email protected]>
* fix: check deletes against the files staged before them
Signed-off-by: badalprasadsingh <[email protected]>
---------
Signed-off-by: badalprasadsingh <[email protected]>
---
table/conflict_validation.go | 131 ++++++++++++-
table/mor_delete_conflict_test.go | 19 ++
table/producer_validate_test.go | 29 ++-
table/row_delta.go | 75 ++++---
table/row_delta_test.go | 400 ++++++++++++++++++++++++++++++++++++++
table/table.go | 6 +-
table/transaction.go | 14 +-
7 files changed, 629 insertions(+), 45 deletions(-)
diff --git a/table/conflict_validation.go b/table/conflict_validation.go
index 0ade50718..e8b56a159 100644
--- a/table/conflict_validation.go
+++ b/table/conflict_validation.go
@@ -154,9 +154,11 @@ var (
// writing or replacing.
ErrConflictingDeleteFiles = fmt.Errorf("%w: concurrent delete files
added", ErrCommitFailed)
- // ErrDataFilesMissing is returned when a concurrent commit deleted
- // data files the committer explicitly references (e.g. a position
- // delete pointing at a file that is no longer reachable).
+ // ErrDataFilesMissing is returned when data files the committer
+ // explicitly references (e.g. the target of a position delete) are
+ // not live: a concurrent commit removed them, or, when RowDelta.Commit
+ // stages the delete, an operation staged earlier in the same
+ // transaction did. A refresh does not help in the second case.
ErrDataFilesMissing = fmt.Errorf("%w: referenced data files missing",
ErrCommitFailed)
)
@@ -411,8 +413,9 @@ func (c *conflictContext) manifestsFor(snap Snapshot)
([]iceberg.ManifestFile, e
// partition-spec projection resolves the same column identifiers.
//
// When base and current point at the same snapshot on the branch, the
-// returned context has no concurrent snapshots and validators
-// short-circuit to nil. When the base snapshot is no longer on the
+// returned context has no concurrent snapshots and validators that
+// inspect them short-circuit to nil; validateDataFilesExist still reads
+// the branch head. When the base snapshot is no longer on the
// branch (divergent commit, expired base), newConflictContext returns
// ErrCommitDiverged; the commit cannot be safely revalidated without
// a refresh-and-rebuild.
@@ -511,6 +514,8 @@ func (c *conflictContext) forEachAddedEntry(content
iceberg.ManifestContent, vis
// validateDataFilesExist verifies that every file path in
// referencedPaths is still a live data file on the current branch head.
+// Files the transaction stages are not on the head yet, so callers pass
+// only the paths Transaction.headDataFiles returns when the delete is staged.
// This is the check a position-delete commit performs to prove the
// files it references have not been removed by a concurrent commit
// (at which point the pos-delete would apply to rewritten data and
@@ -580,13 +585,127 @@ func validateDataFilesExist(ctx *conflictContext,
referencedPaths []string) erro
for p := range needed {
missing = append(missing, p)
}
+
+ return dataFilesMissingError(missing, "missing")
+}
+
+// headDataFiles returns the paths in referenced that a delete staged next
must still
+// find live on the branch head when the transaction commits
(validateDataFilesExist).
+// It replays, oldest first, the data files added and removed by the staged
snapshots
+// the delete is layered on (its parent and the parent's staged ancestors, a
chain a
+// rollback can shorten), then sorts each referenced file into one case:
+// - added by those snapshots or in added (the delete's own data files), and
not
+// removed since: left out, as no concurrent commit can remove an
uncommitted file;
+// - removed by those snapshots: rejected with ErrDataFilesMissing, as it
would be
+// across separate commits;
+// - anything else: returned for the head check, which also rejects a file
that only
+// a later snapshot adds, since the delete would carry the lower sequence
number.
+//
+// A retry sees the same changes: it reuses the staged snapshots' added
manifests and
+// fails if a file one of them removes is gone from the fresh parent
(checkRemovedFiles).
+func (t *Transaction) headDataFiles(fs iceio.IO, referenced []string, added
[]iceberg.DataFile) ([]string, error) {
+ if len(referenced) == 0 {
+ return nil, nil
+ }
+ meta, err := t.txnMeta()
+ if err != nil {
+ return nil, err
+ }
+
+ staged := make(map[int64]*addSnapshotUpdate)
+ for _, u := range meta.updates {
+ if su, ok := u.(*addSnapshotUpdate); ok {
+ staged[su.Snapshot.SnapshotID] = su
+ }
+ }
+ // A parent is always added before its child, so the walk ends at the
first
+ // snapshot this transaction did not stage.
+ var chain []*addSnapshotUpdate
+ if parent := t.planningSnapshot(meta); parent != nil {
+ for su, ok := staged[parent.SnapshotID]; ok; {
+ chain = append(chain, su)
+ if su.Snapshot.ParentSnapshotID == nil {
+ break
+ }
+ su, ok = staged[*su.Snapshot.ParentSnapshotID]
+ }
+ }
+ slices.Reverse(chain)
+
+ live := make(map[string]struct{}, len(added))
+ removed := make(map[string]struct{})
+ for _, su := range chain {
+ id := su.Snapshot.SnapshotID
+ manifests, err := su.Snapshot.Manifests(fs)
+ if err != nil {
+ return nil, fmt.Errorf("iterating data files removed by
staged snapshot %d: %w", id, err)
+ }
+ for _, mf := range manifests {
+ // Only data manifests this snapshot wrote can hold its
tombstones; inherited
+ // ones restate the head (re-read at commit) and the
snapshots replayed before it.
+ if mf.ManifestContent() != iceberg.ManifestContentData
|| mf.SnapshotID() != id || mf.DeletedDataFiles() == 0 {
+ continue
+ }
+ for entry, err := range mf.Entries(fs, false) {
+ if err != nil {
+ return nil, fmt.Errorf("iterating data
files removed by staged snapshot %d: %w", id, err)
+ }
+ if entry.Status() == iceberg.EntryStatusDELETED
&& entry.SnapshotID() == id {
+ path := entry.DataFile().FilePath()
+ delete(live, path)
+ removed[path] = struct{}{}
+ }
+ }
+ }
+
+ for _, mf := range su.ownManifests {
+ if mf.ManifestContent() != iceberg.ManifestContentData {
+ continue
+ }
+ for entry, err := range mf.Entries(fs, true) {
+ if err != nil {
+ return nil, fmt.Errorf("iterating data
files added by staged snapshot %d: %w", id, err)
+ }
+ path := entry.DataFile().FilePath()
+ live[path] = struct{}{}
+ }
+ }
+ }
+ for _, df := range added {
+ live[df.FilePath()] = struct{}{}
+ }
+
+ head := make([]string, 0, len(referenced))
+ var missing []string
+ for _, path := range referenced {
+ // Not committed yet, so no concurrent commit can remove it.
+ if _, ok := live[path]; ok {
+ continue
+ }
+ if _, ok := removed[path]; ok {
+ missing = append(missing, path)
+
+ continue
+ }
+ head = append(head, path)
+ }
+ if len(missing) > 0 {
+ return nil, dataFilesMissingError(missing, "removed earlier in
this transaction")
+ }
+
+ return head, nil
+}
+
+// dataFilesMissingError reports the referenced data files a delete cannot
apply to,
+// naming the first few in path order; state says what became of them.
+func dataFilesMissingError(missing []string, state string) error {
sort.Strings(missing)
sample := missing
if len(sample) > 5 {
sample = sample[:5]
}
- return fmt.Errorf("%w: %d files missing, e.g. %v", ErrDataFilesMissing,
len(missing), sample)
+ return fmt.Errorf("%w: %d files %s, e.g. %v", ErrDataFilesMissing,
len(missing), state, sample)
}
// validateAddedDataFilesMatchingFilter returns an error if any
diff --git a/table/mor_delete_conflict_test.go
b/table/mor_delete_conflict_test.go
index 44e6fc737..84778c39b 100644
--- a/table/mor_delete_conflict_test.go
+++ b/table/mor_delete_conflict_test.go
@@ -128,3 +128,22 @@ func TestMergeOnReadDeleteConflict_NoConflictCommits(t
*testing.T) {
require.Equal(t, []int64{1, 2, 3, 4, 5, 6, 8, 9, 10}, idsInTable(t,
tbl))
}
+
+// TestMergeOnReadDeleteOfDataFileAddedInSameTransaction proves a merge-on-read
+// DELETE may target a data file added earlier in the same transaction:
+// the file is not on the branch head yet, but it lands atomically with the
deletion vector.
+func TestMergeOnReadDeleteOfDataFileAddedInSameTransaction(t *testing.T) {
+ ctx := context.Background()
+ tbl := appendTenRows(t, newV3MoRConflictTestTable(t))
+
+ path := tbl.Location() + "/data/added-in-txn.parquet"
+ writeRowDeltaDataFile(t, tbl, path, 11, 12)
+
+ txn := tbl.NewTransaction()
+ require.NoError(t, txn.AddFiles(ctx, []string{path}, nil, false))
+ require.NoError(t, txn.Delete(ctx,
iceberg.EqualTo(iceberg.Reference("id"), int64(11)), nil))
+ tbl, err := txn.Commit(ctx)
+ require.NoError(t, err)
+
+ require.Equal(t, []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 12},
idsInTable(t, tbl))
+}
diff --git a/table/producer_validate_test.go b/table/producer_validate_test.go
index 594e74af3..6bee44b36 100644
--- a/table/producer_validate_test.go
+++ b/table/producer_validate_test.go
@@ -224,19 +224,32 @@ func TestRowDelta_ValidateNoDeletes(t *testing.T) {
cc := newEmptyConflictContext(t)
txn := newValidateTestTxn(t, nil)
rd := &RowDelta{txn: txn}
- require.NoError(t, rd.validate(cc))
+ require.NoError(t, rd.validate(cc, nil))
}
// TestRowDelta_ValidatePosDeleteWithoutReference covers the realistic
// case where the pos-delete file does not record its referenced data
-// file (ReferencedDataFile returns nil). validate() must skip the
-// referenced-file check, matching Java's behavior.
+// file (ReferencedDataFile returns nil). referencedDataFiles() must
+// leave it out so validate() skips the referenced-file check.
func TestRowDelta_ValidatePosDeleteWithoutReference(t *testing.T) {
cc := newEmptyConflictContext(t)
txn := newValidateTestTxn(t, nil)
posDelete := newTestPosDeleteFile(t, "pos-1.parquet", nil)
rd := &RowDelta{txn: txn, delFiles: []iceberg.DataFile{posDelete}}
- require.NoError(t, rd.validate(cc))
+ require.NoError(t, rd.validate(cc, rd.referencedDataFiles()))
+}
+
+// TestRowDelta_ReferencedDataFilesAreDistinct pins what Commit checks:
+// each referenced data file once, skipping position deletes that record none
and equality deletes.
+func TestRowDelta_ReferencedDataFilesAreDistinct(t *testing.T) {
+ ref := "data.parquet"
+ rd := &RowDelta{delFiles: []iceberg.DataFile{
+ newTestPosDeleteFile(t, "pos-1.parquet", &ref),
+ newTestPosDeleteFile(t, "pos-2.parquet", nil),
+ newTestEqDeleteFile(t, "eq-1.parquet", []int{1}),
+ newTestPosDeleteFile(t, "pos-3.parquet", &ref),
+ }}
+ assert.Equal(t, []string{ref}, rd.referencedDataFiles())
}
// TestRowDelta_ValidateEqDeleteIsolationGates proves the isolation
@@ -253,7 +266,7 @@ func TestRowDelta_ValidateEqDeleteIsolationGates(t
*testing.T) {
WriteDeleteIsolationLevelKey: string(level),
})
rd := &RowDelta{txn: txn, delFiles:
[]iceberg.DataFile{eqDelete}}
- require.NoError(t, rd.validate(cc), "%s + no concurrent should
be nil", level)
+ require.NoError(t, rd.validate(cc, nil), "%s + no concurrent
should be nil", level)
}
}
@@ -267,7 +280,7 @@ func TestRowDelta_ValidateInvalidIsolationLevel(t
*testing.T) {
})
rd := &RowDelta{txn: txn, delFiles: []iceberg.DataFile{eqDelete}}
- assert.ErrorIs(t, rd.validate(cc), ErrInvalidIsolationLevel)
+ assert.ErrorIs(t, rd.validate(cc, nil), ErrInvalidIsolationLevel)
}
func TestRowDelta_ValidateInvalidIsolationLevelWithPosDeletesOnly(t
*testing.T) {
@@ -278,7 +291,7 @@ func
TestRowDelta_ValidateInvalidIsolationLevelWithPosDeletesOnly(t *testing.T)
})
rd := &RowDelta{txn: txn, delFiles: []iceberg.DataFile{posDelete}}
- assert.ErrorIs(t, rd.validate(cc), ErrInvalidIsolationLevel)
+ assert.ErrorIs(t, rd.validate(cc, nil), ErrInvalidIsolationLevel)
}
func TestRowDelta_ValidateInvalidIsolationLevelWithNilContext(t *testing.T) {
@@ -287,7 +300,7 @@ func
TestRowDelta_ValidateInvalidIsolationLevelWithNilContext(t *testing.T) {
})
rd := &RowDelta{txn: txn}
- assert.ErrorIs(t, rd.validate(nil), ErrInvalidIsolationLevel)
+ assert.ErrorIs(t, rd.validate(nil, nil), ErrInvalidIsolationLevel)
}
func TestRewriteValidator_Smokes(t *testing.T) {
diff --git a/table/row_delta.go b/table/row_delta.go
index c19a331c2..600204185 100644
--- a/table/row_delta.go
+++ b/table/row_delta.go
@@ -46,7 +46,8 @@ import (
// Client-side conflict validation runs before the commit is sent to
// the catalog:
// - Position deletes: referenced data files must still be reachable
-// from the current branch head (validateDataFilesExist).
+// from the current branch head (validateDataFilesExist), unless
+// this delta or an operation staged before it in the transaction adds
them.
// - Equality deletes under write.delete.isolation-level=serializable
// (the default): concurrent data files in the same partition(s) as
// the equality deletes are rejected. For partitioned tables an
@@ -60,8 +61,9 @@ import (
// The pre-flight runs before cat.CommitTable on every commit attempt.
// On the first attempt the writer's view and the catalog state are the
// same, so there are no concurrent snapshots to inspect and the checks
-// short-circuit; on retries doCommit refreshes the catalog state and
-// the checks run against the freshly loaded branch head.
+// against them short-circuit, while the position-delete check still
+// reads the branch head; on retries doCommit refreshes the catalog state
+// and the checks run against the freshly loaded branch head.
//
// Usage:
//
@@ -149,8 +151,10 @@ func (rd *RowDelta) RemoveDeletes(files
...iceberg.DataFile) *RowDelta {
// deletion vector lacks a replacement or is not referenced by the
// current snapshot, if the delta would leave two live deletion vectors
// on one data file (a replacement added while the superseded live DV
-// is not removed, or two added replacements for one data file), or if
-// the table format version does not support delete files.
+// is not removed, or two added replacements for one data file), if a
+// position delete references a data file that an operation staged
+// earlier in the transaction removed (ErrDataFilesMissing), or if the
+// table format version does not support delete files.
// Deletes require v2 or later, DVs require v3 or later.
//
// With removals present the staged commit is non-replayable: an
@@ -242,6 +246,11 @@ func (rd *RowDelta) Commit(ctx context.Context) error {
return err
}
+ referenced, err := rd.txn.headDataFiles(fs, rd.referencedDataFiles(),
rd.dataFiles)
+ if err != nil {
+ return err
+ }
+
op := rd.Operation()
// Find live DVs that would be superseded by added DVs, regardless of
@@ -321,11 +330,38 @@ func (rd *RowDelta) Commit(ctx context.Context) error {
// suppressed above; RowDelta semantics (pos-delete references,
// eq-delete predicate) require a dedicated check that
// snapshot_producers does not know about.
- rd.txn.addValidator(rd.validate)
+ rd.txn.addValidator(func(cc *conflictContext) error {
+ return rd.validate(cc, referenced)
+ })
return rd.txn.apply(updates, reqs)
}
+// referencedDataFiles returns the distinct data files that this delta's
+// position deletes name in referenced_data_file. A position delete that
+// leaves the field unset, such as a v2 file covering several data files,
+// names its targets only in its file_path column, which is not read here,
+// so those targets go unchecked.
+func (rd *RowDelta) referencedDataFiles() []string {
+ var referenced []string
+ seen := make(map[string]struct{})
+ for _, f := range rd.delFiles {
+ if f.ContentType() != iceberg.EntryContentPosDeletes {
+ continue
+ }
+ ref := iceberginternal.BorrowedDataFileReferencedDataFile(f)
+ if ref == nil || *ref == "" {
+ continue
+ }
+ if _, ok := seen[*ref]; !ok {
+ seen[*ref] = struct{}{}
+ referenced = append(referenced, *ref)
+ }
+ }
+
+ return referenced
+}
+
// pathRefKey is the removal identity of a delete-manifest entry: its
// file path plus the data file its deletion vector references (""
// for delete files that record none).
@@ -589,10 +625,12 @@ func (rd *RowDelta) resolveRemovedDeletes(fs iceio.IO,
meta *MetadataBuilder) (r
// reflects the freshly refreshed branch state. Two invariants are
// enforced:
//
-// - Every data file referenced by a position-delete in this RowDelta
-// must still be reachable from the branch head. A concurrent
-// compaction or overwrite that rewrote a referenced file would
-// orphan this pos-delete and produce incorrect results — reject.
+// - Every data file in referenced, the position-delete targets that
+// Commit did not find among the transaction's staged files
+// (Transaction.headDataFiles), must still be reachable from the
+// branch head. A concurrent compaction or overwrite that rewrote a
+// referenced file would orphan this pos-delete and produce incorrect
+// results — reject.
// Always runs, no isolation gating.
//
// - When any equality-delete is included and isolation is
@@ -608,7 +646,7 @@ func (rd *RowDelta) resolveRemovedDeletes(fs iceio.IO, meta
*MetadataBuilder) (r
//
// Fast appends alongside a RowDelta see no validators from RowDelta:
// data-only commits are as safe as a fastAppend.
-func (rd *RowDelta) validate(cc *conflictContext) error {
+func (rd *RowDelta) validate(cc *conflictContext, referenced []string) error {
meta, err := rd.txn.txnMeta()
if err != nil {
return err
@@ -620,22 +658,9 @@ func (rd *RowDelta) validate(cc *conflictContext) error {
return err
}
- // Collect every data-file path the pos-deletes in this delta
- // reference. A nil ReferencedDataFile means the pos-delete does
- // not record its target — we cannot check it here; the file is
- // still present in the per-row position_delete_file column and
- // would apply correctly regardless of concurrent removals,
- // matching Java's behavior when the referenced-file column is
- // unset.
- var referenced []string
var eqDeleteFiles []iceberg.DataFile
for _, f := range rd.delFiles {
- switch f.ContentType() {
- case iceberg.EntryContentPosDeletes:
- if ref :=
iceberginternal.BorrowedDataFileReferencedDataFile(f); ref != nil && *ref != ""
{
- referenced = append(referenced, *ref)
- }
- case iceberg.EntryContentEqDeletes:
+ if f.ContentType() == iceberg.EntryContentEqDeletes {
eqDeleteFiles = append(eqDeleteFiles, f)
}
}
diff --git a/table/row_delta_test.go b/table/row_delta_test.go
index 86e4ddd73..86d025758 100644
--- a/table/row_delta_test.go
+++ b/table/row_delta_test.go
@@ -20,6 +20,7 @@ package table_test
import (
"context"
"iter"
+ "os"
"path/filepath"
"strconv"
"strings"
@@ -1870,3 +1871,402 @@ func TestRowDeltaRejectsConcurrentFirstDVOnRetry(t
*testing.T) {
assert.Equal(t, int32(1), catA.commitTableCalls.Load(),
"the validator must reject on the refreshed retry, before a
second commit attempt")
}
+
+func writeRowDeltaDataFile(t *testing.T, tbl *table.Table, path string, ids
...int64) iceberg.DataFile {
+ t.Helper()
+
+ arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
+ require.NoError(t, err)
+
+ rows := make([]string, 0, len(ids))
+ for _, id := range ids {
+ row := `{"id": ` + strconv.FormatInt(id, 10) + `}`
+ rows = append(rows, row)
+ }
+
+ jsonData := "[" + strings.Join(rows, ",") + "]"
+ writeParquetFile(t, path, arrowSc, jsonData)
+
+ info, err := os.Stat(path)
+ require.NoError(t, err)
+
+ b, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentData, path,
iceberg.ParquetFile,
+ nil, nil, nil, int64(len(ids)), info.Size())
+ require.NoError(t, err)
+
+ return b.Build()
+}
+
+func seedRowDeltaDataFile(t *testing.T, tbl *table.Table, path string, ids
...int64) *table.Table {
+ t.Helper()
+
+ writeRowDeltaDataFile(t, tbl, path, ids...)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(t.Context(), []string{path}, nil, false))
+
+ tbl, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ require.NotNil(t, tbl.CurrentSnapshot())
+
+ return tbl
+}
+
+func writeRowDeltaPositionDelete(t *testing.T, location, name, dataPath
string, positions []int64) iceberg.DataFile {
+ t.Helper()
+
+ rows := make([]string, 0, len(positions))
+ for _, pos := range positions {
+ row := `{"file_path": "` + dataPath + `", "pos": ` +
strconv.FormatInt(pos, 10) + `}`
+ rows = append(rows, row)
+ }
+ path := location + "/data/" + name
+ jsonData := "[" + strings.Join(rows, ",") + "]"
+ writeParquetFile(t, path, table.PositionalDeleteArrowSchema, jsonData)
+
+ info, err := os.Stat(path)
+ require.NoError(t, err)
+
+ b, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
path, iceberg.ParquetFile,
+ nil, nil, nil, int64(len(positions)), info.Size())
+ require.NoError(t, err)
+
+ return b.ReferencedDataFile(dataPath).Build()
+}
+
+// addedDataFileStages stage a RowDelta that adds dataB with deletes,
+// either in the same RowDelta or in a later one of the same transaction.
+var addedDataFileStages = []struct {
+ name string
+ stage func(t *testing.T, tx *table.Transaction, dataB iceberg.DataFile,
deletes ...iceberg.DataFile)
+}{
+ {
+ name: "same row delta",
+ stage: func(t *testing.T, tx *table.Transaction, dataB
iceberg.DataFile, deletes ...iceberg.DataFile) {
+ require.NoError(t,
tx.NewRowDelta(nil).AddRows(dataB).AddDeletes(deletes...).Commit(t.Context()))
+ },
+ },
+ {
+ name: "earlier row delta in the transaction",
+ stage: func(t *testing.T, tx *table.Transaction, dataB
iceberg.DataFile, deletes ...iceberg.DataFile) {
+ require.NoError(t,
tx.NewRowDelta(nil).AddRows(dataB).Commit(t.Context()))
+ require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(deletes...).Commit(t.Context()))
+ },
+ },
+}
+
+// A data file added by the commit is not on the branch head yet.
+//
+// Condition: one commit adds file B and deletes for head file A and for B,
+// via one RowDelta or two in one transaction, as v2 position deletes or DVs.
+// Assertion: the commit succeeds and each delete hides only its own row.
+func TestRowDeltaDeletesDataFileAddedInSameCommit(t *testing.T) {
+ formats := []struct {
+ name string
+ formatVersion int
+ ext string
+ writeDelete func(t *testing.T, location, name, dataPath
string, positions []int64) iceberg.DataFile
+ }{
+ {name: "v2 position delete", formatVersion: 2, ext: ".parquet",
writeDelete: writeRowDeltaPositionDelete},
+ {name: "v3 deletion vector", formatVersion: 3, ext: ".puffin",
writeDelete: writeDV},
+ }
+ for _, format := range formats {
+ for _, tt := range addedDataFileStages {
+ t.Run(format.name+"/"+tt.name, func(t *testing.T) {
+ tbl := newRowDeltaCommitTestTableVersion(t,
format.formatVersion)
+ location := tbl.Location()
+ pathA := location + "/data/data-a.parquet"
+ pathB := location + "/data/data-b.parquet"
+ tbl = seedRowDeltaDataFile(t, tbl, pathA, 1, 2,
3)
+
+ dataB := writeRowDeltaDataFile(t, tbl, pathB,
4, 5, 6)
+ deleteA := format.writeDelete(t, location,
"delete-a"+format.ext, pathA, []int64{0})
+ deleteB := format.writeDelete(t, location,
"delete-b"+format.ext, pathB, []int64{2})
+
+ tx := tbl.NewTransaction()
+ tt.stage(t, tx, dataB, deleteA, deleteB)
+ tbl, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ assert.Equal(t, []int64{2, 3, 4, 5},
idsInTable(t, tbl))
+ })
+ }
+ }
+}
+
+// A delete-file rewrite may target a data file added earlier in the same
transaction.
+//
+// Condition: a RowDelta adds B with a DV for it, then a rewrite replaces that
DV.
+// Assertion: the commit succeeds and only the replacement DV hides B's rows.
+func TestRewriteFilesReplacesDVOfDataFileAddedInSameTransaction(t *testing.T) {
+ tbl := newRowDeltaCommitTestTableVersion(t, 3)
+ location := tbl.Location()
+ pathB := location + "/data/data-b.parquet"
+ tbl = seedRowDeltaDataFile(t, tbl, location+"/data/data-a.parquet", 1,
2, 3)
+
+ dataB := writeRowDeltaDataFile(t, tbl, pathB, 4, 5, 6)
+ dvB := writeDV(t, location, "dv-b.puffin", pathB, []int64{0})
+ replacement := writeDV(t, location, "dv-b-replacement.puffin", pathB,
[]int64{0, 2})
+
+ tx := tbl.NewTransaction()
+ require.NoError(t,
tx.NewRowDelta(nil).AddRows(dataB).AddDeletes(dvB).Commit(t.Context()))
+ require.NoError(t,
tx.NewRewrite(nil).DeleteFile(dvB).AddDeleteFile(replacement).Commit(t.Context()))
+ tbl, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ assert.Equal(t, []int64{1, 2, 3, 5}, idsInTable(t, tbl))
+ assert.Equal(t, 1, liveDVCount(t, tbl))
+}
+
+// Exempting files the transaction adds must not waive the check for others.
+//
+// Condition: a DV targets a file that is neither on the head nor added when
the DV is staged:
+// an unknown file beside a DV for B, B added only by a later RowDelta, which
would leave
+// the DV with the lower data sequence number, or B added by a RowDelta rolled
back before it.
+// Assertion: ErrDataFilesMissing names only that file
+// and the catalog never receives a commit.
+func TestRowDeltaRejectsDeleteForDataFileNotLiveWhenStaged(t *testing.T) {
+ tests := []struct {
+ name string
+ stage func(t *testing.T, tx *table.Transaction, headID
int64, dataB, dvB, dvUnknown iceberg.DataFile)
+ wantMissing string
+ }{
+ {
+ name: "unknown file beside a file the row delta adds",
+ stage: func(t *testing.T, tx *table.Transaction, _
int64, dataB, dvB, dvUnknown iceberg.DataFile) {
+ require.NoError(t,
tx.NewRowDelta(nil).AddRows(dataB).AddDeletes(dvB,
dvUnknown).Commit(t.Context()))
+ },
+ wantMissing: "unknown.parquet",
+ },
+ {
+ name: "file a later row delta adds",
+ stage: func(t *testing.T, tx *table.Transaction, _
int64, dataB, dvB, _ iceberg.DataFile) {
+ require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(dvB).Commit(t.Context()))
+ require.NoError(t,
tx.NewRowDelta(nil).AddRows(dataB).Commit(t.Context()))
+ },
+ wantMissing: "data-b.parquet",
+ },
+ {
+ name: "file a rolled back row delta adds",
+ stage: func(t *testing.T, tx *table.Transaction, headID
int64, dataB, dvB, _ iceberg.DataFile) {
+ require.NoError(t,
tx.NewRowDelta(nil).AddRows(dataB).Commit(t.Context()))
+ require.NoError(t,
tx.RollbackToSnapshot(headID))
+ require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(dvB).Commit(t.Context()))
+ },
+ wantMissing: "data-b.parquet",
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ tbl, cat := newConcurrentRewriteTestTableVersion(t, 3)
+ location := tbl.Location()
+ pathB := location + "/data/data-b.parquet"
+ unknownPath := location + "/data/unknown.parquet"
+
+ tbl = seedRowDeltaDataFile(t, tbl,
location+"/data/data-a.parquet", 1, 2)
+ headBefore := tbl.CurrentSnapshot().SnapshotID
+ attemptsBefore := cat.attempts.Load()
+
+ dataB := writeRowDeltaDataFile(t, tbl, pathB, 3, 4)
+ dvB := writeDV(t, location, "dv-b.puffin", pathB,
[]int64{0})
+ dvUnknown := writeDV(t, location, "dv-unknown.puffin",
unknownPath, []int64{0})
+
+ tx := tbl.NewTransaction()
+ tt.stage(t, tx, headBefore, dataB, dvB, dvUnknown)
+ _, err := tx.Commit(t.Context())
+ require.ErrorIs(t, err, table.ErrDataFilesMissing)
+ assert.ErrorContains(t, err, "1 files missing")
+ assert.ErrorContains(t, err,
location+"/data/"+tt.wantMissing)
+ assert.Equal(t, attemptsBefore, cat.attempts.Load())
+ require.NotNil(t, cat.metadata.CurrentSnapshot())
+ assert.Equal(t, headBefore,
cat.metadata.CurrentSnapshot().SnapshotID)
+ })
+ }
+}
+
+// Rolling back a staged removal makes the file live again for later deletes.
+//
+// Condition: a rewrite removes head file A, the transaction rolls back to the
head,
+// then a RowDelta adds a DV for A.
+// Assertion: the commit succeeds, the DV applies to A, and A carries the only
live DV.
+func TestRowDeltaDeletesDataFileAfterRollingBackItsRemoval(t *testing.T) {
+ tbl := newRowDeltaCommitTestTableVersion(t, 3)
+ location := tbl.Location()
+ pathA := location + "/data/data-a.parquet"
+ tbl = seedRowDeltaDataFile(t, tbl, pathA, 1, 2, 3)
+ headID := tbl.CurrentSnapshot().SnapshotID
+
+ tasks, err := tbl.Scan().PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ dvA := writeDV(t, location, "dv-a.puffin", pathA, []int64{0})
+
+ tx := tbl.NewTransaction()
+ require.NoError(t,
tx.NewRewrite(nil).DeleteFile(tasks[0].File).Commit(t.Context()))
+ require.NoError(t, tx.RollbackToSnapshot(headID))
+ require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(dvA).Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ assert.Equal(t, []int64{2, 3}, idsInTable(t, tbl))
+ assert.Equal(t, 1, liveDVCount(t, tbl))
+}
+
+// A retry re-checks the refreshed head, which still lacks the added B.
+//
+// Condition: a transaction adds B with DVs for head file A and for B, in one
RowDelta
+// or two, and a concurrent append or removal of A commits first.
+// Assertion: the append retries with both DVs; the removal fails naming A.
+func TestRowDeltaDeletesDataFileAddedInSameCommitOnRetry(t *testing.T) {
+ tests := []struct {
+ name string
+ concurrent func(t *testing.T, tbl *table.Table, dataA
iceberg.DataFile)
+ wantIDs []int64
+ wantErr bool
+ }{
+ {
+ name: "concurrent append",
+ concurrent: func(t *testing.T, tbl *table.Table, _
iceberg.DataFile) {
+ seedRowDeltaDataFile(t, tbl,
tbl.Location()+"/data/data-c.parquet", 7)
+ },
+ wantIDs: []int64{2, 3, 4, 5, 7},
+ },
+ {
+ name: "concurrent removal of a head file",
+ concurrent: func(t *testing.T, tbl *table.Table, dataA
iceberg.DataFile) {
+ tx := tbl.NewTransaction()
+ require.NoError(t,
tx.NewRewrite(nil).DeleteFile(dataA).Commit(t.Context()))
+ _, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ },
+ wantErr: true,
+ },
+ }
+
+ for _, stage := range addedDataFileStages {
+ for _, tt := range tests {
+ t.Run(stage.name+"/"+tt.name, func(t *testing.T) {
+ tbl, cat :=
newConcurrentRewriteTestTableVersion(t, 3)
+ location := tbl.Location()
+ pathA := location + "/data/data-a.parquet"
+ pathB := location + "/data/data-b.parquet"
+ tbl = seedRowDeltaDataFile(t, tbl, pathA, 1, 2,
3)
+
+ tasks, err := tbl.Scan().PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+
+ dataB := writeRowDeltaDataFile(t, tbl, pathB,
4, 5, 6)
+ dvA := writeDV(t, location, "dv-a.puffin",
pathA, []int64{0})
+ dvB := writeDV(t, location, "dv-b.puffin",
pathB, []int64{2})
+ tx := tbl.NewTransaction()
+ stage.stage(t, tx, dataB, dvA, dvB)
+
+ tt.concurrent(t, tbl, tasks[0].File)
+ require.NotNil(t,
cat.metadata.CurrentSnapshot())
+ headBefore :=
cat.metadata.CurrentSnapshot().SnapshotID
+ attemptsBefore := cat.attempts.Load()
+
+ committed, err := tx.Commit(t.Context())
+ if tt.wantErr {
+ require.ErrorIs(t, err,
table.ErrDataFilesMissing)
+ assert.ErrorContains(t, err, "1 files
missing")
+ assert.ErrorContains(t, err, pathA)
+ assert.Equal(t, int32(1),
cat.attempts.Load()-attemptsBefore,
+ "the retry must be rejected
before reaching the catalog")
+ assert.Equal(t, headBefore,
cat.metadata.CurrentSnapshot().SnapshotID)
+
+ return
+ }
+
+ require.NoError(t, err)
+ assert.Equal(t, int32(2),
cat.attempts.Load()-attemptsBefore, "the commit must conflict once and retry")
+ assert.Equal(t, tt.wantIDs, idsInTable(t,
committed))
+ })
+ }
+ }
+}
+
+// A rewrite in the same transaction removes the data file a DV targets.
+//
+// Condition: the rewrite removes head file A, or B added earlier in the
transaction,
+// before the RowDelta that adds the DV beside a DV for the other, still live
file,
+// or after the DV together with it.
+// Assertion: only the DV staged after the removal fails, naming that file
alone and writing
+// nothing, as across separate commits; either way the table ends without the
file and
+// without a live DV.
+func TestRowDeltaDeleteForDataFileRemovedInSameTransaction(t *testing.T) {
+ files := []struct {
+ name string
+ pick func(dataA, dataB iceberg.DataFile) (target, other
iceberg.DataFile)
+ wantIDs []int64
+ }{
+ {
+ name: "head file",
+ pick: func(dataA, dataB iceberg.DataFile)
(iceberg.DataFile, iceberg.DataFile) { return dataA, dataB },
+ wantIDs: []int64{4, 5, 6},
+ },
+ {
+ name: "file added earlier in the transaction",
+ pick: func(dataA, dataB iceberg.DataFile)
(iceberg.DataFile, iceberg.DataFile) { return dataB, dataA },
+ wantIDs: []int64{1, 2, 3},
+ },
+ }
+ orders := []struct {
+ name string
+ stage func(t *testing.T, tbl *table.Table, tx
*table.Transaction, target, dv, otherDV iceberg.DataFile)
+ }{
+ {
+ name: "removed before the delete",
+ stage: func(t *testing.T, tbl *table.Table, tx
*table.Transaction, target, dv, otherDV iceberg.DataFile) {
+ before := metadataFileCount(t, tbl)
+ require.NoError(t,
tx.NewRewrite(nil).DeleteFile(target).Commit(t.Context()))
+ staged := metadataFileCount(t, tbl)
+ require.Greater(t, staged, before, "the probe
must see the files a staged operation writes")
+
+ err := tx.NewRowDelta(nil).AddDeletes(otherDV,
dv).Commit(t.Context())
+ require.ErrorIs(t, err,
table.ErrDataFilesMissing)
+ assert.ErrorContains(t, err, "1 files removed
earlier in this transaction")
+ assert.ErrorContains(t, err, target.FilePath())
+ assert.Equal(t, staged, metadataFileCount(t,
tbl), "the rejected row delta must write nothing")
+ },
+ },
+ {
+ name: "removed after the delete",
+ stage: func(t *testing.T, _ *table.Table, tx
*table.Transaction, target, dv, _ iceberg.DataFile) {
+ require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(dv).Commit(t.Context()))
+ require.NoError(t,
tx.NewRewrite(nil).DeleteFile(target).DeleteFile(dv).Commit(t.Context()))
+ },
+ },
+ }
+
+ for _, file := range files {
+ for _, order := range orders {
+ t.Run(file.name+"/"+order.name, func(t *testing.T) {
+ tbl := newRowDeltaCommitTestTableVersion(t, 3)
+ location := tbl.Location()
+ tbl = seedRowDeltaDataFile(t, tbl,
location+"/data/data-a.parquet", 1, 2, 3)
+
+ tasks, err := tbl.Scan().PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+
+ dataB := writeRowDeltaDataFile(t, tbl,
location+"/data/data-b.parquet", 4, 5, 6)
+ target, other := file.pick(tasks[0].File, dataB)
+ dv := writeDV(t, location, "dv-target.puffin",
target.FilePath(), []int64{0})
+ otherDV := writeDV(t, location,
"dv-other.puffin", other.FilePath(), []int64{0})
+
+ tx := tbl.NewTransaction()
+ require.NoError(t,
tx.NewRowDelta(nil).AddRows(dataB).Commit(t.Context()))
+ order.stage(t, tbl, tx, target, dv, otherDV)
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ assert.Equal(t, file.wantIDs, idsInTable(t,
tbl))
+ assert.Zero(t, liveDVCount(t, tbl))
+ })
+ }
+ }
+}
diff --git a/table/table.go b/table/table.go
index 1d3189024..ab06c9c35 100644
--- a/table/table.go
+++ b/table/table.go
@@ -744,9 +744,9 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
// peers (partition-filter overlap, referenced-file removal)
// even when the catalog-side AssertRefSnapshotID would accept
// them. On attempt 0 base == current → no concurrent
- // snapshots → validators short-circuit. Real divergence
- // detection fires on attempts > 0 once `current` is the
- // post-conflict state.
+ // snapshots → validators that inspect them short-circuit, but
+ // validateDataFilesExist still reads the branch head. Real
+ // divergence detection fires on attempts > 0 once `current` is
the post-conflict state.
//
// Skipped when the branch does not exist on `current` — that
// always means "the committer is creating this branch" (e.g.
diff --git a/table/transaction.go b/table/transaction.go
index 3e9bac41f..0966c86b5 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -2137,6 +2137,9 @@ func (t *Transaction) replaceFiles(ctx context.Context,
dataFilesToDelete, dataF
referencedDataFilePaths =
append(referencedDataFilePaths, ref)
}
slices.Sort(referencedDataFilePaths)
+ if referencedDataFilePaths, err = t.headDataFiles(fs,
referencedDataFilePaths, nil); err != nil {
+ return err
+ }
}
if !cfg.skipAutoNameMapping {
@@ -2612,8 +2615,13 @@ func (t *Transaction) performMergeOnReadDeletion(ctx
context.Context, snapshotPr
// validates partition-filter overlap; without this check a
concurrent
// commit that removed a referenced data file would silently
orphan the
// deletes (they would apply to rewritten data or dangle).
Mirrors
- // RowDelta.validate: collect referenced data-file paths and run
- // validateDataFilesExist unconditionally (no isolation gating).
+ // RowDelta: collect referenced data-file paths, leave out the
ones
+ // staged earlier in the transaction, and run
validateDataFilesExist
+ // unconditionally (no isolation gating).
+ referenced, err = t.headDataFiles(fs, referenced, nil)
+ if err != nil {
+ return nil, err
+ }
if len(referenced) > 0 {
t.addValidator(func(cc *conflictContext) error {
return validateDataFilesExist(cc, referenced)
@@ -3291,7 +3299,7 @@ func (t *Transaction) rewriteSingleFile(ctx
context.Context, args rewriteSingleF
// files that match filter, appends them to updater, and returns the distinct
// data-file paths those deletes reference. A position-delete that does not
// record its referenced data file is omitted from the returned paths (it
-// cannot be existence-checked), matching RowDelta.validate and Java.
+// cannot be existence-checked), matching RowDelta.referencedDataFiles.
func (t *Transaction) writePositionDeletesForFiles(ctx context.Context, fs
io.IO, updater *snapshotProducer, dvs liveDeletionVectors, files
[]iceberg.DataFile, filter iceberg.BooleanExpression, caseSensitive bool,
concurrency int, commitUUID uuid.UUID) ([]string, error) {
meta, err := t.txnMeta()
if err != nil {
