This is an automated email from the ASF dual-hosted git repository.
laskoviymishka 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 c5964d1ba fix(table): validate DV replacement on add (#2054)
c5964d1ba is described below
commit c5964d1ba12897f3697fd36693edfd14b337a87f
Author: Andrei Tserakhau <[email protected]>
AuthorDate: Wed Oct 14 18:29:19 2026 +0200
fix(table): validate DV replacement on add (#2054)
Two paths still let a row delta commit more than one live deletion vector
for a single data file:
- Same-delta duplicate DVs on the fast-append path. findReplacedLiveDVs
collapses added DV references into a set, and when the data file has no
live DV yet the scan returns nothing; with no RemoveDeletes,
validateRemovedDeletes (which held the only duplicate-added-DV check)
never runs, so AddDeletes(dvA, dvB) for one data file committed two live
DVs. Hoist the duplicate-added-DV check into validateAddedDeletionVectors,
run unconditionally on every commit path.
- Concurrent first-DV additions across a commit retry. Two writers based on
a snapshot with no DV for a data file both pass findReplacedLiveDVs. After
one commits, the other's add-only delta is replayable: doCommit rebuilds
the manifest list against the refreshed parent but did not re-check DV
uniqueness, so the retry could inherit the concurrent DV and commit a
second one. Add validateNoConcurrentReplacedDVs to RowDelta.validate: on
retry it walks the concurrently committed delete entries and rejects a
live DV added for a data file this delta also adds a DV for. It is a no-op
on attempt 0 (no concurrent snapshots), and a removal-carrying delta is
non-replayable, so there are no removals to reconcile on retry.
Tests: TestRowDeltaFastAppendRejectsDuplicateAddedDVs covers the fast-append
duplicate, and TestRowDeltaRejectsConcurrentFirstDVOnRetry drives the
two-writer retry through the OCC scenario catalog and asserts the retry is
rejected before a second commit attempt.
---
table/row_delta.go | 184 +++++++++++++++++++++++++++++-------
table/row_delta_test.go | 243 +++++++++++++++++++++++++++++++++++++++---------
2 files changed, 350 insertions(+), 77 deletions(-)
diff --git a/table/row_delta.go b/table/row_delta.go
index a8fee000f..c19a331c2 100644
--- a/table/row_delta.go
+++ b/table/row_delta.go
@@ -222,6 +222,16 @@ func (rd *RowDelta) Commit(ctx context.Context) error {
}
}
+ // A data file may carry at most one live deletion vector, so a single
row
+ // delta may add at most one DV per referenced data file. This must be
+ // checked on every path: on the fast-append path no removals are
present,
+ // so validateRemovedDeletes (which also enforces it) never runs, and
two
+ // DVs added for a data file with no live DV yet would otherwise commit
two
+ // live DVs and make scans reject the table.
+ if err := rd.validateAddedDeletionVectors(); err != nil {
+ return err
+ }
+
fs, err := rd.txn.tbl.fsF(ctx)
if err != nil {
return err
@@ -234,6 +244,27 @@ func (rd *RowDelta) Commit(ctx context.Context) error {
op := rd.Operation()
+ // Find live DVs that would be superseded by added DVs, regardless of
+ // whether removals are present. If any such live DVs exist and are not
+ // being removed, the commit must fail.
+ replacedLive, err := rd.findReplacedLiveDVs(fs, meta)
+ if err != nil {
+ return err
+ }
+
+ // Validate that any live DV whose referenced data file is being
+ // superseded by an added DV is explicitly removed in this delta.
+ if len(replacedLive) > 0 && len(rd.removedDels) == 0 {
+ // No removals were registered, but there are live DVs that
would
+ // be superseded. Return an error with the same message format
as
+ // validateRemovedDeletes to guide the caller.
+ live := replacedLive[0]
+ ref := explicitReferencedDataFile(live)
+
+ return fmt.Errorf("cannot add a replacement deletion vector for
data file %s: live deletion vector %s is not removed by this row delta; the
superseded entry must be removed in the same snapshot",
+ ref, live.FilePath())
+ }
+
var producer *snapshotProducer
if len(rd.removedDels) > 0 {
// Resolve the removed files against the current snapshot's
@@ -245,7 +276,7 @@ func (rd *RowDelta) Commit(ctx context.Context) error {
// live DVs whose referenced data file this delta adds a
// replacement for; validateRemovedDeletes checks each is
// actually removed.
- resolvedRemovals, replacedLive, err :=
rd.resolveRemovedDeletes(fs, meta)
+ resolvedRemovals, err := rd.resolveRemovedDeletes(fs, meta)
if err != nil {
return err
}
@@ -338,13 +369,12 @@ func explicitReferencedDataFile(df iceberg.DataFile)
string {
// removals — otherwise the commit would leave both the old and the
// new DV live on one data file.
func (rd *RowDelta) validateRemovedDeletes(resolved, replacedLive
[]iceberg.DataFile) error {
+ // Duplicate added DVs (two references to one data file) are already
+ // rejected unconditionally by validateAddedDeletionVectors before this
+ // runs; here the set only pairs removals with their replacements.
addedDVs := make(map[string]struct{}, len(rd.delFiles))
for _, f := range rd.delFiles {
if ref := explicitReferencedDataFile(f); IsDeletionVector(f) &&
ref != "" {
- if _, ok := addedDVs[ref]; ok {
- return fmt.Errorf("multiple added deletion
vectors reference data file %s; at most one live deletion vector may exist per
data file",
- ref)
- }
addedDVs[ref] = struct{}{}
}
}
@@ -379,6 +409,66 @@ func (rd *RowDelta) validateRemovedDeletes(resolved,
replacedLive []iceberg.Data
return nil
}
+// validateAddedDeletionVectors rejects a row delta that adds more than one
+// deletion vector referencing the same data file. Two live DVs on one data
+// file violate the v3 spec and make scan planning reject the table. It runs on
+// every commit path; validateRemovedDeletes enforces the same rule but only
+// runs when removals are present, so the fast-append path relies on this.
+func (rd *RowDelta) validateAddedDeletionVectors() error {
+ addedDVs := make(map[string]struct{}, len(rd.delFiles))
+ for _, f := range rd.delFiles {
+ if ref := explicitReferencedDataFile(f); IsDeletionVector(f) &&
ref != "" {
+ if _, ok := addedDVs[ref]; ok {
+ return fmt.Errorf("multiple added deletion
vectors reference data file %s; at most one live deletion vector may exist per
data file",
+ ref)
+ }
+ addedDVs[ref] = struct{}{}
+ }
+ }
+
+ return nil
+}
+
+// findReplacedLiveDVs walks the current snapshot's delete manifests and
+// collects all live deletion vectors whose referenced data file this delta
+// adds a replacement for. This is used to enforce the v3 spec invariant
+// that at most one live DV can exist per data file: if any such live DVs
+// are found and they are not being explicitly removed, the commit must fail.
+func (rd *RowDelta) findReplacedLiveDVs(fs iceio.IO, meta *MetadataBuilder)
([]iceberg.DataFile, error) {
+ snap := rd.txn.planningSnapshot(meta)
+ if snap == nil {
+ // No existing snapshot means no live DVs to worry about.
+ return nil, nil
+ }
+
+ addedRefs := make(map[string]struct{}, len(rd.delFiles))
+ for _, f := range rd.delFiles {
+ if ref := explicitReferencedDataFile(f); IsDeletionVector(f) &&
ref != "" {
+ addedRefs[ref] = struct{}{}
+ }
+ }
+
+ // If no added DVs, there's nothing to replace.
+ if len(addedRefs) == 0 {
+ return nil, nil
+ }
+
+ var replacedLive []iceberg.DataFile
+ for entry, err := range snap.entries(fs,
iceberg.ManifestContentDeletes, true) {
+ if err != nil {
+ return nil, err
+ }
+ df := entry.DataFile()
+ if ref := explicitReferencedDataFile(df); IsDeletionVector(df)
&& ref != "" {
+ if _, ok := addedRefs[ref]; ok {
+ replacedLive = append(replacedLive, df)
+ }
+ }
+ }
+
+ return replacedLive, nil
+}
+
// resolveRemovedDeletes walks the current snapshot's delete manifests
// and resolves each removed file to its live manifest entry, returned
// in the order the removals were registered.
@@ -408,18 +498,10 @@ func (rd *RowDelta) validateRemovedDeletes(resolved,
replacedLive []iceberg.Data
// than one live entry at the path is rejected: the producer would
// tombstone every matching entry while the snapshot summary counts
// only one removal.
-//
-// The walk also collects replacedLive: every live deletion vector
-// whose referenced data file this delta adds a replacement DV for.
-// validateRemovedDeletes requires each to be among the removals, so a
-// delta cannot commit a replacement while leaving the superseded DV
-// live. This piggybacks on the manifest walk the removals already
-// need; deltas without removals do not pay for it (nor get it — see
-// AddDeletes).
-func (rd *RowDelta) resolveRemovedDeletes(fs iceio.IO, meta *MetadataBuilder)
(resolved, replacedLive []iceberg.DataFile, _ error) {
+func (rd *RowDelta) resolveRemovedDeletes(fs iceio.IO, meta *MetadataBuilder)
(resolved []iceberg.DataFile, _ error) {
snap := rd.txn.planningSnapshot(meta)
if snap == nil {
- return nil, nil, errors.New("cannot remove delete files from a
table without an existing snapshot")
+ return nil, errors.New("cannot remove delete files from a table
without an existing snapshot")
}
want := make(map[string]struct{}, len(rd.removedDels))
@@ -427,27 +509,15 @@ func (rd *RowDelta) resolveRemovedDeletes(fs iceio.IO,
meta *MetadataBuilder) (r
want[f.FilePath()] = struct{}{}
}
- addedRefs := make(map[string]struct{}, len(rd.delFiles))
- for _, f := range rd.delFiles {
- if ref := explicitReferencedDataFile(f); IsDeletionVector(f) &&
ref != "" {
- addedRefs[ref] = struct{}{}
- }
- }
-
liveByPath := make(map[string][]iceberg.DataFile, len(rd.removedDels))
for entry, err := range snap.entries(fs,
iceberg.ManifestContentDeletes, true) {
if err != nil {
- return nil, nil, err
+ return nil, err
}
df := entry.DataFile()
if _, ok := want[df.FilePath()]; ok {
liveByPath[df.FilePath()] =
append(liveByPath[df.FilePath()], df)
}
- if ref := explicitReferencedDataFile(df); IsDeletionVector(df)
&& ref != "" {
- if _, ok := addedRefs[ref]; ok {
- replacedLive = append(replacedLive, df)
- }
- }
}
seenKeys := make(map[pathRefKey]struct{}, len(rd.removedDels))
@@ -459,13 +529,13 @@ func (rd *RowDelta) resolveRemovedDeletes(fs iceio.IO,
meta *MetadataBuilder) (r
var live iceberg.DataFile
switch {
case len(candidates) == 0:
- return nil, nil, fmt.Errorf("cannot remove delete files
that do not belong to the table: %s", f.FilePath())
+ return nil, fmt.Errorf("cannot remove delete files that
do not belong to the table: %s", f.FilePath())
case len(candidates) == 1:
live = candidates[0]
default:
ref := explicitReferencedDataFile(f)
if ref == "" {
- return nil, nil, fmt.Errorf("ambiguous removal:
%d live delete entries share path %s; the removed file must declare a
referenced data file to identify one",
+ return nil, fmt.Errorf("ambiguous removal: %d
live delete entries share path %s; the removed file must declare a referenced
data file to identify one",
len(candidates), f.FilePath())
}
matches := 0
@@ -479,10 +549,10 @@ func (rd *RowDelta) resolveRemovedDeletes(fs iceio.IO,
meta *MetadataBuilder) (r
}
switch {
case matches == 0:
- return nil, nil, fmt.Errorf("cannot remove
delete file %s: no live delete entry references data file %s",
+ return nil, fmt.Errorf("cannot remove delete
file %s: no live delete entry references data file %s",
f.FilePath(), ref)
case matches > 1:
- return nil, nil, fmt.Errorf("found %d live
delete entries at %s referencing data file %s; the table has duplicate live
deletion vectors for one data file",
+ return nil, fmt.Errorf("found %d live delete
entries at %s referencing data file %s; the table has duplicate live deletion
vectors for one data file",
matches, f.FilePath(), ref)
}
}
@@ -490,7 +560,7 @@ func (rd *RowDelta) resolveRemovedDeletes(fs iceio.IO, meta
*MetadataBuilder) (r
liveRef := explicitReferencedDataFile(live)
key := pathRefKey{path: live.FilePath(), ref: liveRef}
if _, ok := seenKeys[key]; ok {
- return nil, nil, fmt.Errorf("removed delete files must
be unique: %s (referenced data file %q)",
+ return nil, fmt.Errorf("removed delete files must be
unique: %s (referenced data file %q)",
live.FilePath(), liveRef)
}
seenKeys[key] = struct{}{}
@@ -502,7 +572,7 @@ func (rd *RowDelta) resolveRemovedDeletes(fs iceio.IO, meta
*MetadataBuilder) (r
// error, so they are exempt here.
if liveRef != "" {
if prevPath, ok := seenRefs[liveRef]; ok {
- return nil, nil, fmt.Errorf("removed delete
files %s and %s both reference data file %q; the table has two live deletion
vectors for one data file",
+ return nil, fmt.Errorf("removed delete files %s
and %s both reference data file %q; the table has two live deletion vectors for
one data file",
prevPath, live.FilePath(), liveRef)
}
seenRefs[liveRef] = live.FilePath()
@@ -510,7 +580,7 @@ func (rd *RowDelta) resolveRemovedDeletes(fs iceio.IO, meta
*MetadataBuilder) (r
resolved[i] = live
}
- return resolved, replacedLive, nil
+ return resolved, nil
}
// validate is the client-side conflict check for a RowDelta commit. It
@@ -574,6 +644,15 @@ func (rd *RowDelta) validate(cc *conflictContext) error {
return nil
}
+ // Enforce the v3 single-live-DV invariant against concurrently
committed
+ // snapshots. findReplacedLiveDVs caught DVs already live in the base at
+ // Commit time; this catches a concurrent writer that added the first DV
+ // for the same data file after our base, visible once the branch is
+ // refreshed on retry.
+ if err := rd.validateNoConcurrentReplacedDVs(cc); err != nil {
+ return err
+ }
+
if len(referenced) > 0 {
if err := validateDataFilesExist(cc, referenced); err != nil {
return err
@@ -607,6 +686,41 @@ func (rd *RowDelta) validate(cc *conflictContext) error {
return nil
}
+// validateNoConcurrentReplacedDVs enforces the v3 single-live-DV invariant on
+// commit retries. findReplacedLiveDVs (run once at Commit time) rejects DVs
+// already live in the base snapshot, but two writers can each add the FIRST DV
+// for the same data file: both pass that scan against their shared base, and
an
+// add-only row delta is replayable. On retry cc carries the concurrently
+// committed snapshots, so a concurrent writer that added a DV for a data file
+// this delta also adds a DV for would strand two live DVs on one data file —
+// reject. A removal-carrying delta is non-replayable (see RemoveDeletes), so a
+// retry never has removals to reconcile against the concurrent DV here.
+func (rd *RowDelta) validateNoConcurrentReplacedDVs(cc *conflictContext) error
{
+ addedRefs := make(map[string]struct{}, len(rd.delFiles))
+ for _, f := range rd.delFiles {
+ if ref := explicitReferencedDataFile(f); IsDeletionVector(f) &&
ref != "" {
+ addedRefs[ref] = struct{}{}
+ }
+ }
+ if len(addedRefs) == 0 {
+ return nil
+ }
+
+ return cc.forEachAddedEntry(iceberg.ManifestContentDeletes, func(snap
Snapshot, entry iceberg.ManifestEntry) error {
+ df := entry.DataFile()
+ ref := explicitReferencedDataFile(df)
+ if !IsDeletionVector(df) || ref == "" {
+ return nil
+ }
+ if _, ok := addedRefs[ref]; ok {
+ return fmt.Errorf("cannot add a deletion vector for
data file %s: concurrent snapshot %d committed a live deletion vector %s for
it; reload the table and rewrite the DV to supersede the concurrent one,
removing the superseded DV in the same row delta",
+ ref, snap.SnapshotID, df.FilePath())
+ }
+
+ return nil
+ })
+}
+
// Operation returns the snapshot operation type that will be used when
// this RowDelta is committed:
// - data only → OpAppend
diff --git a/table/row_delta_test.go b/table/row_delta_test.go
index 237f0457c..86e4ddd73 100644
--- a/table/row_delta_test.go
+++ b/table/row_delta_test.go
@@ -963,7 +963,7 @@ func TestRowDeltaRemoveDeletesRequiresReplacement(t
*testing.T) {
// Assertion: Commit fails identifying the surviving live DV (a) or the
// duplicated reference (b).
func TestRowDeltaRemoveDeletesRejectsSurvivingLiveDV(t *testing.T) {
- t.Run("replacement added while live DV not removed", func(t *testing.T)
{
+ t.Run("replacement added while live DV not removed (with
RemoveDeletes)", func(t *testing.T) {
tbl, location, pathA, pathB, dvA, _ :=
newTableWithSharedPuffinDVs(t)
// Supersede A's DV, but also add a replacement for B without
@@ -980,6 +980,20 @@ func TestRowDeltaRemoveDeletesRejectsSurvivingLiveDV(t
*testing.T) {
assert.Contains(t, err.Error(), pathB)
})
+ t.Run("replacement added while live DV not removed (without
RemoveDeletes)", func(t *testing.T) {
+ tbl, location, pathA, _, _, _ := newTableWithSharedPuffinDVs(t)
+
+ // Add a replacement DV for A's live DV without removing the
live one.
+ // This should fail in the fast-append path (no RemoveDeletes
called).
+ dvA2 := writeDV(t, location, "dv-a2.puffin", pathA, []int64{0,
1})
+ rd := tbl.NewTransaction().NewRowDelta(nil).AddDeletes(dvA2)
+
+ err := rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "is not removed by this row
delta")
+ assert.Contains(t, err.Error(), pathA)
+ })
+
t.Run("two replacements for one data file", func(t *testing.T) {
tbl, location, dataPath, dv1 := newTableWithLiveDV(t)
@@ -995,60 +1009,83 @@ func TestRowDeltaRemoveDeletesRejectsSurvivingLiveDV(t
*testing.T) {
})
}
-// Why: a table that already violates the one-live-DV-per-data-file
-// invariant must fail a supersession commit loudly — the producer keys
-// DV removals by referenced data file, so proceeding would either
-// tombstone several entries while counting one removal (same path) or
-// leave the sibling duplicate live next to the replacement (different
-// paths).
-// Condition: v3 tables corrupted through the unvalidated plain
-// AddDeletes path so one data file carries two live DV entries, first
-// as duplicate entries at one Puffin path, then at two distinct paths;
-// a RowDelta then supersedes one of them.
-// Assertion: Commit fails naming the duplicate entries rather than
-// committing.
+// Why: supersession in the fast-append path (no RemoveDeletes called)
+// must still enforce the one-live-DV-per-data-file invariant and require
+// explicit removal of the live DV.
+// Condition: a RowDelta adds a replacement DV for a data file with a live
+// DV, without calling RemoveDeletes.
+// Assertion: Commit fails with the same error as the RemoveDeletes case,
+// directing the caller to remove the superseded DV. With RemoveDeletes,
+// the same operation succeeds.
+func TestRowDeltaFastAppendRejectsDVSupersessionWithoutRemoval(t *testing.T) {
+ tbl, location, dataPath, dv1 := newTableWithLiveDV(t)
+
+ // Add a replacement DV without removing the live one.
+ rep := writeDV(t, location, "dv-002.puffin", dataPath, []int64{0, 1})
+ tx := tbl.NewTransaction()
+ rd := tx.NewRowDelta(nil).AddDeletes(rep)
+
+ err := rd.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "is not removed by this row delta")
+ assert.Contains(t, err.Error(), "superseded entry must be removed in
the same snapshot")
+
+ // With RemoveDeletes, the same operation succeeds.
+ rep2 := writeDV(t, location, "dv-003.puffin", dataPath, []int64{0, 1})
+ tx2 := tbl.NewTransaction()
+ rd2 := tx2.NewRowDelta(nil).AddDeletes(rep2).RemoveDeletes(dv1)
+ require.NoError(t, rd2.Commit(t.Context()))
+
+ result, err := tx2.Commit(t.Context())
+ require.NoError(t, err)
+ snap := result.CurrentSnapshot()
+ require.NotNil(t, snap)
+
+ fs := iceio.LocalFS{}
+ live, removed := snapshotDVEntries(t, snap, fs)
+ assert.Equal(t, []string{rep2.FilePath()}, live)
+ assert.Equal(t, []string{dv1.FilePath()}, removed)
+}
+
+// Why: the fast-append path now prevents tables from being corrupted with
+// duplicate DVs for the same data file. Previously, duplicate entries could
+// be added through the unvalidated plain AddDeletes path, violating the
+// one-live-DV-per-data-file invariant. Now this is detected immediately.
+// Condition: a RowDelta attempts to add a replacement DV for a data file
+// that already has a live DV without removing the live one.
+// Assertion: Commit fails in the fast-append path itself, before any
+// corruption can occur.
func TestRowDeltaRemoveDeletesCorruptDuplicateLiveDVs(t *testing.T) {
- t.Run("duplicate entries at one path", func(t *testing.T) {
- tbl, location, dataPath, dv1 := newTableWithLiveDV(t)
+ t.Run("duplicate entries at one path prevented", func(t *testing.T) {
+ tbl, _, dataPath, dv1 := newTableWithLiveDV(t)
- // Corrupt the table: a second snapshot re-adds the same DV
- // entry (same path, same referenced data file).
+ // Attempt to re-add the same DV without removing the live one.
+ // This now fails in the fast-append validation, preventing
corruption.
tx := tbl.NewTransaction()
- require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(dv1).Commit(t.Context()))
- tbl, err := tx.Commit(t.Context())
- require.NoError(t, err)
-
- replacement := writeDV(t, location, "dv-002.puffin", dataPath,
[]int64{0, 1})
- rd := tbl.NewTransaction().NewRowDelta(nil).
- AddDeletes(replacement).
- RemoveDeletes(dv1)
+ rd := tx.NewRowDelta(nil).AddDeletes(dv1)
- err = rd.Commit(t.Context())
+ err := rd.Commit(t.Context())
require.Error(t, err)
- assert.Contains(t, err.Error(), "duplicate live deletion
vectors")
+ assert.Contains(t, err.Error(), "is not removed by this row
delta",
+ "adding a duplicate DV for the same data file must fail
in the fast-append path")
+ assert.Contains(t, err.Error(), dataPath)
})
- t.Run("duplicate entries at two paths", func(t *testing.T) {
- tbl, location, dataPath, dv1 := newTableWithLiveDV(t)
+ t.Run("duplicate entries at two paths prevented", func(t *testing.T) {
+ tbl, location, dataPath, _ := newTableWithLiveDV(t)
- // Corrupt the table: a second live DV for the same data file
- // at a different Puffin path.
+ // Attempt to add a second live DV for the same data file
+ // at a different Puffin path, without removing the first.
+ // This now fails in the fast-append validation, preventing
corruption.
dv1b := writeDV(t, location, "dv-001b.puffin", dataPath,
[]int64{0})
tx := tbl.NewTransaction()
- require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(dv1b).Commit(t.Context()))
- tbl, err := tx.Commit(t.Context())
- require.NoError(t, err)
+ rd := tx.NewRowDelta(nil).AddDeletes(dv1b)
- replacement := writeDV(t, location, "dv-002.puffin", dataPath,
[]int64{0, 1})
- rd := tbl.NewTransaction().NewRowDelta(nil).
- AddDeletes(replacement).
- RemoveDeletes(dv1)
-
- err = rd.Commit(t.Context())
+ err := rd.Commit(t.Context())
require.Error(t, err)
assert.Contains(t, err.Error(), "is not removed by this row
delta",
- "the sibling duplicate at the other path must be
flagged as surviving")
- assert.Contains(t, err.Error(), dv1b.FilePath())
+ "adding a second DV for the same data file must fail in
the fast-append path")
+ assert.Contains(t, err.Error(), dataPath)
})
}
@@ -1711,3 +1748,125 @@ func writtenManifests(t *testing.T, location string)
[]string {
return paths
}
+
+// Why: two DVs added in one row delta that reference the same data file would
+// commit two live DVs even when that data file has no live DV yet, so the
+// removal path's duplicate check never runs (no RemoveDeletes → fast-append).
+// The fast-append path must reject the duplicate directly.
+// Condition: AddDeletes(dvA, dvB) where both reference the same data file that
+// currently has no live DV, and no RemoveDeletes is called.
+// Assertion: Commit fails with the duplicate-added-DV error before committing.
+func TestRowDeltaFastAppendRejectsDuplicateAddedDVs(t *testing.T) {
+ tbl := newRowDeltaCommitTestTableVersion(t, 3)
+ location := tbl.Location()
+
+ arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
+ require.NoError(t, err)
+
+ dataPath := location + "/data/data-100.parquet"
+ writeParquetFile(t, dataPath, arrowSc, `[
+ {"id": 1, "data": "alpha"},
+ {"id": 2, "data": "beta"}
+ ]`)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(t.Context(), []string{dataPath}, nil,
false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ // The data file has no live DV yet, and no RemoveDeletes is called, so
+ // only the fast-append path can catch the duplicate.
+ dvA := writeDV(t, location, "dv-a.puffin", dataPath, []int64{0})
+ dvB := writeDV(t, location, "dv-b.puffin", dataPath, []int64{1})
+
+ err = tbl.NewTransaction().NewRowDelta(nil).AddDeletes(dvA,
dvB).Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "multiple added deletion vectors
reference data file "+dataPath)
+}
+
+// Why: two writers that each add the FIRST deletion vector for the same data
+// file both pass the pre-commit scan against their shared base (no live DV
+// yet). After one commits, the other's add-only row delta is replayable, so
+// the retry must re-check against the refreshed head, or the table ends up
+// with two live DVs for one data file.
+// Condition: Writer A adds a DV for data file D; a concurrent writer commits
+// the first DV for D before A's commit lands, forcing A to retry.
+// Assertion: A's commit fails on the retry (after one refresh, before a second
+// commit attempt) with the concurrent-DV conflict error.
+func TestRowDeltaRejectsConcurrentFirstDVOnRetry(t *testing.T) {
+ localFS := func(context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }
+
+ location := filepath.ToSlash(t.TempDir())
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2, Name: "data", Type:
iceberg.PrimitiveTypes.String},
+ )
+ props := iceberg.Properties{
+ table.PropertyFormatVersion: "3",
+ table.CommitMinRetryWaitMsKey: "0",
+ table.CommitMaxRetryWaitMsKey: "0",
+ table.CommitTotalRetryTimeoutMsKey: "60000",
+ table.CommitNumRetriesKey: "2",
+ }
+ meta, err := table.NewMetadata(schema, iceberg.UnpartitionedSpec,
table.UnsortedSortOrder, location, props)
+ require.NoError(t, err)
+
+ ident := table.Identifier{"db", "row_delta_concurrent_dv"}
+
+ // Base: one data file D with two rows and no deletion vector.
+ baseCat := &occScenarioCatalog{current: meta, location: location}
+ baseTbl := table.New(ident, meta,
location+"/metadata/v1.metadata.json", localFS, baseCat)
+
+ arrowSc, err := table.SchemaToArrowSchema(baseTbl.Schema(), nil, false,
false)
+ require.NoError(t, err)
+ dataPath := location + "/data/data-001.parquet"
+ writeParquetFile(t, dataPath, arrowSc, `[
+ {"id": 1, "data": "alpha"},
+ {"id": 2, "data": "beta"}
+ ]`)
+
+ tx := baseTbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(t.Context(), []string{dataPath}, nil,
false))
+ _, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ metaBase := baseCat.current
+
+ // A concurrent writer commits the first DV for D just before A's
commit,
+ // advancing the head A will refresh onto.
+ addConcurrentDV := func(current table.Metadata) table.Metadata {
+ peerCat := &occScenarioCatalog{current: current, location:
location}
+ peerTbl := table.New(ident, current,
location+"/metadata/peer.metadata.json", localFS, peerCat)
+
+ dvC := writeDV(t, location, "dv-concurrent.puffin", dataPath,
[]int64{0})
+ ptx := peerTbl.NewTransaction()
+ require.NoError(t,
ptx.NewRowDelta(nil).AddDeletes(dvC).Commit(t.Context()))
+ _, perr := ptx.Commit(t.Context())
+ require.NoError(t, perr)
+
+ return peerCat.current
+ }
+
+ // Writer A starts from the DV-free base and adds its own first DV for
D.
+ // Its catalog conflicts once (landing the concurrent DV), then would
accept
+ // the retry — which the validator must reject before it commits.
+ catA := &occScenarioCatalog{
+ current: metaBase,
+ conflictsLeft: 1,
+ location: location,
+ onConflict: addConcurrentDV,
+ }
+ writerA := table.New(ident, metaBase,
location+"/metadata/base.metadata.json", localFS, catA)
+
+ dvA := writeDV(t, location, "dv-writer-a.puffin", dataPath, []int64{1})
+ txA := writerA.NewTransaction()
+ require.NoError(t,
txA.NewRowDelta(nil).AddDeletes(dvA).Commit(t.Context()))
+
+ _, err = txA.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "concurrent snapshot")
+ assert.Contains(t, err.Error(), dataPath)
+ assert.Equal(t, int32(1), catA.loadTableCalls.Load(),
+ "the commit must refresh once for the retry before rejecting")
+ assert.Equal(t, int32(1), catA.commitTableCalls.Load(),
+ "the validator must reject on the refreshed retry, before a
second commit attempt")
+}
