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 407a4cdc8 fix(table): remove a removed data file's deletion vector on
every replace path (#2239)
407a4cdc8 is described below
commit 407a4cdc8fd73352364d2ebf0af0f597f24ff9d5
Author: Revanth <[email protected]>
AuthorDate: Tue Oct 20 16:13:35 2026 -0500
fix(table): remove a removed data file's deletion vector on every replace
path (#2239)
---
table/dv_rewrite_test.go | 110 +++++++++++++++++++++++++++++++++++++++++++++++
table/rewrite_files.go | 4 +-
table/transaction.go | 37 ++++++++++++++--
3 files changed, 147 insertions(+), 4 deletions(-)
diff --git a/table/dv_rewrite_test.go b/table/dv_rewrite_test.go
index abbeed898..0da507fc2 100644
--- a/table/dv_rewrite_test.go
+++ b/table/dv_rewrite_test.go
@@ -21,6 +21,7 @@ import (
"context"
"fmt"
"path/filepath"
+ "strconv"
"testing"
"github.com/apache/arrow-go/v18/arrow"
@@ -510,3 +511,112 @@ func deleteEntriesReferencing(t *testing.T, tbl
*table.Table, refs map[string]st
return found
}
+
+// TestRemovingDataFileRemovesItsUnnamedDV covers #2238: the v3 spec requires a
+// removed data file to take its deletion vector with it, so every replace
entry
+// point must expunge the DV even when the caller does not name it. Each case
runs
+// with the files committed in earlier transactions and with everything staged
in
+// one transaction. ReplaceFiles also removes an unrelated equality delete,
which
+// takes the path that resolves delete files instead of the data-only one.
+//
+// Scan planning drops a DV whose data file is gone, so the assertions walk the
+// raw delete manifests.
+func TestRemovingDataFileRemovesItsUnnamedDV(t *testing.T) {
+ removers := []struct {
+ name string
+ remove func(ctx context.Context, tx *table.Transaction, dataB,
dataC, eqDelete iceberg.DataFile) error
+ }{
+ {"RewriteFiles", func(ctx context.Context, tx
*table.Transaction, dataB, dataC, _ iceberg.DataFile) error {
+ return
tx.NewRewrite(nil).DeleteFile(dataB).AddDataFile(dataC).Commit(ctx)
+ }},
+ {"ReplaceDataFilesWithDataFiles", func(ctx context.Context, tx
*table.Transaction, dataB, dataC, _ iceberg.DataFile) error {
+ return tx.ReplaceDataFilesWithDataFiles(ctx,
[]iceberg.DataFile{dataB}, []iceberg.DataFile{dataC}, nil)
+ }},
+ {"ReplaceDataFiles", func(ctx context.Context, tx
*table.Transaction, dataB, _, _ iceberg.DataFile) error {
+ return tx.ReplaceDataFiles(ctx,
[]string{dataB.FilePath()}, nil, nil)
+ }},
+ {"ReplaceFiles", func(ctx context.Context, tx
*table.Transaction, dataB, dataC, eqDelete iceberg.DataFile) error {
+ return tx.ReplaceFiles(ctx, []iceberg.DataFile{dataB},
[]iceberg.DataFile{dataC}, []iceberg.DataFile{eqDelete}, nil)
+ }},
+ }
+
+ for _, rm := range removers {
+ for _, oneTxn := range []bool{false, true} {
+ name := rm.name + "/separate-transactions"
+ if oneTxn {
+ name = rm.name + "/one-transaction"
+ }
+ t.Run(name, func(t *testing.T) {
+ ctx := context.Background()
+ tbl := newRowDeltaCommitTestTableVersion(t, 3)
+ location := tbl.Location()
+ dataA := buildDataFile(t,
location+"/data/a.parquet")
+ dataB := buildDataFile(t,
location+"/data/b.parquet")
+ cBuilder, err :=
iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec, iceberg.EntryContentData,
+ location+"/data/c.parquet",
iceberg.ParquetFile, nil, nil, nil, 10, 1024)
+ require.NoError(t, err)
+ // The non-rewrite replace APIs require v3 data
files to carry a first_row_id.
+ dataC := cBuilder.FirstRowID(100).Build()
+ dvB := buildDVFile(t,
location+"/data/dv-b.puffin", dataB.FilePath())
+ eqDelete := buildEqDeleteFile(t,
location+"/data/eq-delete.parquet", []int{1})
+
+ tx := tbl.NewTransaction()
+ step := func(fn func(tx *table.Transaction)
error) {
+ t.Helper()
+ require.NoError(t, fn(tx))
+ if oneTxn {
+ return
+ }
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+ tx = tbl.NewTransaction()
+ }
+ step(func(tx *table.Transaction) error {
+ return
tx.NewRowDelta(nil).AddRows(dataA, dataB).Commit(ctx)
+ })
+ step(func(tx *table.Transaction) error {
+ return
tx.NewRowDelta(nil).AddDeletes(dvB, eqDelete).Commit(ctx)
+ })
+ require.NoError(t, rm.remove(ctx, tx, dataB,
dataC, eqDelete))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ assert.Empty(t, deleteEntriesReferencing(t,
tbl, map[string]struct{}{dataB.FilePath(): {}}),
+ "a DV must not outlive the data file it
references")
+ snap := tbl.CurrentSnapshot()
+ live, removed := snapshotDeleteEntryFiles(t,
snap, iceio.LocalFS{})
+ assert.Contains(t, pathRefPairs(removed),
[2]string{dvB.FilePath(), dataB.FilePath()})
+ assert.Equal(t, "1",
snap.Summary.Properties["removed-position-delete-files"])
+ assert.Equal(t, strconv.Itoa(len(live)),
snap.Summary.Properties["total-delete-files"])
+ })
+ }
+ }
+}
+
+// TestRewriteFiles_UnnamedDVRemovalKeepsSharedPuffinSibling guards the
removal key:
+// two DVs share one Puffin file, so removing data file A must expunge only
A's DV
+// entry. B's DV stays live and still hides its row.
+func TestRewriteFiles_UnnamedDVRemovalKeepsSharedPuffinSibling(t *testing.T) {
+ ctx := context.Background()
+ tbl, _, pathA, pathB, dvA, dvB := newTableWithSharedPuffinDVs(t)
+
+ tasks, err := tbl.Scan().PlanFiles(ctx)
+ require.NoError(t, err)
+ var dataA iceberg.DataFile
+ for _, task := range tasks {
+ if task.File.FilePath() == pathA {
+ dataA = task.File
+ }
+ }
+ require.NotNil(t, dataA)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.NewRewrite(nil).DeleteFile(dataA).Commit(ctx))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ live, removed := snapshotDeleteEntryFiles(t, tbl.CurrentSnapshot(),
iceio.LocalFS{})
+ assert.Equal(t, [][2]string{{dvB.FilePath(), pathB}},
pathRefPairs(live))
+ assert.Equal(t, [][2]string{{dvA.FilePath(), pathA}},
pathRefPairs(removed))
+ assertRowCount(t, tbl, 1)
+}
diff --git a/table/rewrite_files.go b/table/rewrite_files.go
index 37c14306f..c9d26b863 100644
--- a/table/rewrite_files.go
+++ b/table/rewrite_files.go
@@ -95,7 +95,9 @@ func (t *Transaction) NewRewrite(snapshotProps
iceberg.Properties) *RewriteFiles
// content type: data files are queued as data-file replacements;
// pos/eq-delete files are queued for delete-file removal alongside
// the data rewrite (typical when a delete is fully applied to data
-// files being rewritten and is therefore safe to expunge).
+// files being rewritten and is therefore safe to expunge). On v3 a
+// removed data file's live deletion vector is removed with it, so it
+// does not need to be passed here.
//
// Any other content type stages an error that is returned from the
// next [RewriteFiles.Commit] call.
diff --git a/table/transaction.go b/table/transaction.go
index 0966c86b5..ae1615d04 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -939,6 +939,8 @@ func (t *Transaction) Append(ctx context.Context, rdr
array.RecordReader, snapsh
// operation is only valid if the data is exactly the same as the previous
snapshot.
//
// For now, we'll keep using an overwrite operation.
+//
+// On v3 a removed data file's live deletion vector is removed with it.
func (t *Transaction) ReplaceDataFiles(ctx context.Context, filesToDelete,
filesToAdd []string, snapshotProps iceberg.Properties) error {
meta, err := t.txnMeta()
if err != nil {
@@ -1004,6 +1006,15 @@ func (t *Transaction) ReplaceDataFiles(ctx
context.Context, filesToDelete, files
return errors.New("cannot delete files that do not belong to
the table")
}
+ // On v3 every removed file takes its deletion vector with it.
+ // DVs only exist from v3, so v2 skips the read.
+ var dvs liveDeletionVectors
+ if meta.formatVersion >= 3 && len(markedForDeletion) > 0 {
+ if dvs, err = t.readLiveDeletionVectors(fs); err != nil {
+ return err
+ }
+ }
+
if meta.NameMapping() == nil {
nameMapping := meta.CurrentSchema().NameMapping()
mappingJson, err := json.Marshal(nameMapping)
@@ -1020,7 +1031,7 @@ func (t *Transaction) ReplaceDataFiles(ctx
context.Context, filesToDelete, files
updater := t.updateSnapshot(wfs, snapshotProps,
OpOverwrite).mergeOverwrite(&commitUUID, nil)
for _, df := range markedForDeletion {
- updater.deleteDataFile(df)
+ dvs.removeDataFile(updater, df)
}
// SetProperties above may have staged a name mapping, rebuilding
t.meta;
@@ -1655,6 +1666,8 @@ func (t *Transaction) AddDataFiles(ctx context.Context,
dataFiles []iceberg.Data
// Callers are responsible for ensuring each DataFile is valid and consistent
with the table.
// Supplying incorrect DataFile metadata can produce an invalid snapshot and
break reads.
//
+// On v3 a removed data file's live deletion vector is removed with it.
+//
// This is useful when:
// - Files are written via a separate I/O path and metadata is already known
// - Avoiding file scanning improves performance or reliability
@@ -1732,6 +1745,15 @@ func (t *Transaction) ReplaceDataFilesWithDataFiles(ctx
context.Context, filesTo
return errors.New("cannot delete files that do not belong to
the table")
}
+ // On v3 every removed file takes its deletion vector with it.
+ // DVs only exist from v3, so v2 skips the read.
+ var dvs liveDeletionVectors
+ if meta.formatVersion >= 3 {
+ if dvs, err = t.readLiveDeletionVectors(fs); err != nil {
+ return err
+ }
+ }
+
if !cfg.skipAutoNameMapping {
if err := t.ensureNameMapping(); err != nil {
return err
@@ -1753,7 +1775,7 @@ func (t *Transaction) ReplaceDataFilesWithDataFiles(ctx
context.Context, filesTo
}
for _, df := range markedForDeletion {
- updater.deleteDataFile(df)
+ dvs.removeDataFile(updater, df)
}
for _, df := range filesToAdd {
@@ -1783,7 +1805,8 @@ type DeleteFileAddition struct {
// ReplaceFiles atomically replaces data files and removes associated delete
files
// in a single snapshot. This is the commit primitive for compaction: old data
files
// are replaced with new (compacted) data files, and delete files that are
fully
-// applied are removed.
+// applied are removed. On v3 a removed data file's live deletion vector is
+// removed with it, whether or not it is in deleteFilesToRemove.
func (t *Transaction) ReplaceFiles(ctx context.Context, dataFilesToDelete,
dataFilesToAdd, deleteFilesToRemove []iceberg.DataFile, snapshotProps
iceberg.Properties, opts ...WriteOption) error {
return t.replaceFiles(ctx, dataFilesToDelete, dataFilesToAdd,
deleteFilesToRemove, nil, nil, snapshotProps, opts...)
}
@@ -1941,6 +1964,7 @@ func (t *Transaction) replaceFiles(ctx context.Context,
dataFilesToDelete, dataF
markedDeleteForRemoval := make([]iceberg.DataFile, 0,
len(setDeleteFilesToRemove))
markedAutoDeleteForRemoval := make([]iceberg.DataFile, 0,
len(autoSetDeleteFilesToRemove))
markedDVsForRemoval := make(map[string]iceberg.DataFile,
len(dvsToRemoveByRef))
+ danglingDVs := make(map[string]iceberg.DataFile)
removedDeleteSequenceNumbers := make([]int64, 0,
len(deleteFilesToRemove))
removedDeleteContents := make(map[iceberg.ManifestEntryContent]struct{})
liveDataFiles := make(map[string]rewriteFileState)
@@ -1985,6 +2009,10 @@ func (t *Transaction) replaceFiles(ctx context.Context,
dataFilesToDelete, dataF
}
} else if want, ok :=
autoDVsToRemoveByRef[*ref]; ok && want.FilePath() == path {
markedDVsForRemoval[*ref] = df
+ } else if _, ok := setToDelete[*ref]; ok {
+ // The spec requires a removed data
file to take its DV with it, named or not.
+ // Like automatic removals, it does not
set the sequence number of added delete files.
+ danglingDVs[*ref] = df
}
}
}
@@ -2184,6 +2212,9 @@ func (t *Transaction) replaceFiles(ctx context.Context,
dataFilesToDelete, dataF
for _, df := range markedDVsForRemoval {
updater.removeDeletionVector(df)
}
+ for _, df := range danglingDVs {
+ updater.removeDeletionVector(df)
+ }
updates, reqs, err := updater.commit(ctx)
if err != nil {
