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 b00fada37 feat(table): remove dangling delete files (#2408)
b00fada37 is described below
commit b00fada37e8238e4538ec409eef822850fe32053
Author: Minh Vu <[email protected]>
AuthorDate: Wed Nov 4 19:20:30 2026 +0100
feat(table): remove dangling delete files (#2408)
* feat(table): remove dangling delete files
Signed-off-by: Minh Vu <[email protected]>
* fix(table): preserve partition-scoped deletes during dangling cleanup
* perf(table): avoid scanning data manifests without dangling delete
candidates
* test(table): cover scoped delete references and fix nlreturn lint
* perf(table): index only explicitly referenced data paths
* test(table): verify dangling delete cleanup without delete manifests is a
no-op
* test(table): fence dangling delete cleanup with retries enabled
* fix(table): include unpartitioned tables in dangling delete cleanup
* test(table): verify unpartitioned dangling delete cleanup
* fix(table): match Java scan-based dangling delete cleanup
* style(table): remove trailing blank lines in dangling delete tests
* refactor(table): stage dangling delete cleanup in transactions
---------
Signed-off-by: Minh Vu <[email protected]>
---
table/remove_dangling_delete_files.go | 269 +++++++++++++++++++
...emove_dangling_delete_files_integration_test.go | 286 +++++++++++++++++++++
table/remove_dangling_delete_files_test.go | 94 +++++++
3 files changed, 649 insertions(+)
diff --git a/table/remove_dangling_delete_files.go
b/table/remove_dangling_delete_files.go
new file mode 100644
index 000000000..4bac00ac2
--- /dev/null
+++ b/table/remove_dangling_delete_files.go
@@ -0,0 +1,269 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package table
+
+import (
+ "context"
+
+ "github.com/apache/iceberg-go"
+)
+
+// danglingDeleteKey follows the scan-planning identity of a delete file.
+// Regular position/equality deletes are identified by path. Multiple deletion
+// vectors can share a Puffin file, so each vector needs its own blob identity.
+type danglingDeleteKey struct {
+ path string
+ referenced string
+ offset int64
+ length int64
+ vector bool
+}
+
+func keyForDanglingDelete(df iceberg.DataFile) danglingDeleteKey {
+ key := danglingDeleteKey{path: df.FilePath()}
+ if !IsDeletionVector(df) {
+ return key
+ }
+
+ key.vector = true
+ key.referenced = explicitReferencedDataFile(df)
+ key.offset, key.length = -1, -1
+ if offset := df.ContentOffset(); offset != nil {
+ key.offset = *offset
+ }
+ if length := df.ContentSizeInBytes(); length != nil {
+ key.length = *length
+ }
+
+ return key
+}
+
+// decideDanglingDeleteFiles mirrors Java's RemoveDanglingDeleteFilesAction:
+// delete files that are referenced by at least one scan task are still live.
+// This includes metrics pruning and DV supersession, not only partition and
+// sequence checks. Do not inspect physical delete-file contents.
+func decideDanglingDeleteFiles(candidates []iceberg.DataFile, tasks
[]FileScanTask) []iceberg.DataFile {
+ referenced := make(map[danglingDeleteKey]struct{})
+ for _, task := range tasks {
+ for _, df := range task.DeleteFiles {
+ referenced[keyForDanglingDelete(df)] = struct{}{}
+ }
+ for _, df := range task.EqualityDeleteFiles {
+ referenced[keyForDanglingDelete(df)] = struct{}{}
+ }
+ for _, df := range task.DeletionVectorFiles {
+ referenced[keyForDanglingDelete(df)] = struct{}{}
+ }
+ }
+
+ var dangling []iceberg.DataFile
+ for _, df := range candidates {
+ if _, ok := referenced[keyForDanglingDelete(df)]; !ok {
+ dangling = append(dangling, df)
+ }
+ }
+
+ return dangling
+}
+
+// RemoveDanglingDeleteFilesResult reports delete-file references removed by
cleanup.
+type RemoveDanglingDeleteFilesResult struct {
+ // Table is set by the table-level action after a successful commit.
+ // It is nil when the cleanup is only staged on a transaction.
+ Table *Table
+
+ // RemovedDeleteFiles counts removed position/equality delete files and
+ // deletion vectors. Physical files are never deleted by this action.
+ RemovedDeleteFiles int
+}
+
+// PlanDanglingDeleteFiles identifies delete files that are not attached to any
+// data file by an unfiltered scan of the current snapshot. It reads metadata
+// only; no physical data, delete files, or Puffin blobs are opened.
+//
+// Position and equality deletes use the scan planner's partition, sequence,
+// and metrics checks. DVs are identified by their blob, not just Puffin path.
+// Unknown delete sequence numbers and incomplete DV metadata are retained.
+func (t Table) PlanDanglingDeleteFiles(ctx context.Context)
([]iceberg.DataFile, error) {
+ snap := t.CurrentSnapshot()
+ if snap == nil || t.metadata.FormatVersion() < 2 {
+ return nil, nil
+ }
+
+ return t.planDanglingDeleteFiles(ctx, snap, t.Scan())
+}
+
+// PlanDanglingDeleteFiles plans cleanup against the transaction's target
branch.
+// It also sees snapshots already staged in the transaction, not just the
+// original table head. The scan is local so an external planner cannot read
+// committed catalog state instead of the staged metadata.
+func (tx *Transaction) PlanDanglingDeleteFiles(ctx context.Context)
([]iceberg.DataFile, error) {
+ meta, err := tx.txnMeta()
+ if err != nil {
+ return nil, err
+ }
+ snap := tx.planningSnapshot(meta)
+ if snap == nil || meta.formatVersion < 2 {
+ return nil, nil
+ }
+
+ built, err := meta.Build()
+ if err != nil {
+ return nil, err
+ }
+ view := *tx.tbl
+ view.metadata = built
+ scan := view.Scan(WithScanPlanningMode(ScanPlanningLocal))
+ // Missing branches fork from main, matching
Transaction.planningSnapshot.
+ // Existing branches must scan their own head using the current schema.
+ if tx.branch != "" && tx.branch != MainBranch &&
built.SnapshotByName(tx.branch) != nil {
+ branchScan, err := scan.UseRef(tx.branch)
+ _ = scan.Close()
+ if err != nil {
+ return nil, err
+ }
+ scan = branchScan
+ }
+
+ return view.planDanglingDeleteFiles(ctx, snap, scan)
+}
+
+func (t Table) planDanglingDeleteFiles(ctx context.Context, snap *Snapshot,
scan *Scan) ([]iceberg.DataFile, error) {
+ defer scan.Close()
+
+ fs, err := t.FS(ctx)
+ if err != nil {
+ return nil, err
+ }
+ manifests, err := snap.Manifests(fs)
+ if err != nil {
+ return nil, err
+ }
+
+ seen := make(map[danglingDeleteKey]struct{})
+ var candidates []iceberg.DataFile
+ for _, manifest := range manifests {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ if manifest.ManifestContent() != iceberg.ManifestContentDeletes
{
+ continue
+ }
+ for entry, err := range manifest.Entries(fs, true) {
+ if err != nil {
+ return nil, err
+ }
+ df := entry.DataFile()
+ if df.ContentType() != iceberg.EntryContentPosDeletes &&
+ df.ContentType() !=
iceberg.EntryContentEqDeletes {
+ continue
+ }
+ if entry.DataSequenceNumber() < 0 {
+ continue
+ }
+ if IsDeletionVector(df) &&
+ (df.ReferencedDataFile() == nil ||
df.ContentOffset() == nil || df.ContentSizeInBytes() == nil) {
+ continue
+ }
+
+ key := keyForDanglingDelete(df)
+ if _, ok := seen[key]; ok {
+ continue
+ }
+ seen[key] = struct{}{}
+ candidates = append(candidates, df)
+ }
+ }
+
+ // No delete files means no scan planning, which matters for
append-only tables.
+ if len(candidates) == 0 {
+ return nil, nil
+ }
+
+ tasks, err := scan.PlanFiles(ctx)
+ if err != nil {
+ return nil, err
+ }
+
+ return decideDanglingDeleteFiles(candidates, tasks), nil
+}
+
+// RemoveDanglingDeleteFiles removes metadata references to whole dangling
+// delete files in one replace snapshot. It never deletes physical files,
+// which may still be reachable through older snapshots.
+//
+// The returned result contains the committed table. A no-op returns the input
+// table without writing a new snapshot. Use
Transaction.RemoveDanglingDeleteFiles
+// to stage cleanup alongside other updates or target a non-main branch.
+func (t Table) RemoveDanglingDeleteFiles(ctx context.Context)
(*RemoveDanglingDeleteFilesResult, error) {
+ tx, err := t.NewTransactionOnBranch(MainBranch)
+ if err != nil {
+ return nil, err
+ }
+ result, err := tx.RemoveDanglingDeleteFiles(ctx)
+ if err != nil {
+ return nil, err
+ }
+ if result.RemovedDeleteFiles == 0 {
+ result.Table = &t
+
+ return result, nil
+ }
+
+ result.Table, err = tx.Commit(ctx)
+
+ return result, err
+}
+
+// RemoveDanglingDeleteFiles stages the removal of entire dangling position and
+// equality delete files on the transaction's branch. It does not commit the
+// transaction or delete physical files. Unpartitioned tables are eligible;
+// partially dangling position-delete files require a separate rewrite.
+//
+// The branch head used for planning is pinned for the eventual commit. If it
+// changes concurrently, the commit fails rather than replaying stale removals.
+func (tx *Transaction) RemoveDanglingDeleteFiles(ctx context.Context)
(*RemoveDanglingDeleteFilesResult, error) {
+ candidates, err := tx.PlanDanglingDeleteFiles(ctx)
+ if err != nil {
+ return nil, err
+ }
+ result := &RemoveDanglingDeleteFilesResult{RemovedDeleteFiles:
len(candidates)}
+ if len(candidates) == 0 {
+ return result, nil
+ }
+
+ // A transaction commit already asserts its base branch head. Pin that
+ // assertion against retries without revalidating it against snapshots
+ // staged earlier in this same transaction.
+ branch := tx.branch
+ if branch == "" {
+ branch = MainBranch
+ }
+ if err := tx.applyPinned(nil, nil, []string{branch}); err != nil {
+ return nil, err
+ }
+
+ rewrite := tx.NewRewrite(nil)
+ for _, df := range candidates {
+ rewrite.DeleteFile(df)
+ }
+ if err := rewrite.Commit(ctx); err != nil {
+ return nil, err
+ }
+
+ return result, nil
+}
diff --git a/table/remove_dangling_delete_files_integration_test.go
b/table/remove_dangling_delete_files_integration_test.go
new file mode 100644
index 000000000..486b2eda5
--- /dev/null
+++ b/table/remove_dangling_delete_files_integration_test.go
@@ -0,0 +1,286 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table_test
+
+import (
+ "context"
+ "path/filepath"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/require"
+)
+
+func TestRemoveDanglingDeleteFilesCommit(t *testing.T) {
+ location := filepath.ToSlash(t.TempDir())
+ schema := iceberg.NewSchema(0, iceberg.NestedField{ID: 1, Name: "id",
Type: iceberg.PrimitiveTypes.Int64, Required: true})
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{SourceIDs:
[]int{1}, FieldID: 1000, Name: "id", Transform: iceberg.IdentityTransform{}})
+ meta, err := table.NewMetadata(schema, &spec, table.UnsortedSortOrder,
location, iceberg.Properties{
+ table.PropertyFormatVersion: "2",
+ table.CommitNumRetriesKey: "2",
+ table.CommitMinRetryWaitMsKey: "1",
+ table.CommitMaxRetryWaitMsKey: "2",
+ table.CommitTotalRetryTimeoutMsKey: "1000",
+ })
+ require.NoError(t, err)
+ fsF := func(context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }
+ cat := &concurrentTestCatalog{metadata: meta, location: location +
"/metadata/v1.metadata.json", fsF: fsF}
+ tbl := table.New(table.Identifier{"db", "test"}, meta, cat.location,
fsF, cat)
+ partition := map[int]any{1000: int64(7)}
+ data, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData,
location+"/data/live.parquet", iceberg.ParquetFile, partition, nil, nil, 1, 128)
+ require.NoError(t, err)
+ tx := tbl.NewTransaction()
+ require.NoError(t,
tx.NewRowDelta(nil).AddRows(data.Build()).Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ // With no delete files, planning must leave the table unchanged.
+ before, err := tbl.PlanDanglingDeleteFiles(t.Context())
+ require.NoError(t, err)
+ require.Empty(t, before)
+ noOp, err := tbl.RemoveDanglingDeleteFiles(t.Context())
+ require.NoError(t, err)
+ require.Zero(t, noOp.RemovedDeleteFiles)
+ require.Equal(t, tbl.Metadata(), noOp.Table.Metadata())
+
+ live, err := iceberg.NewDataFileBuilder(spec,
iceberg.EntryContentPosDeletes, location+"/data/live-delete.parquet",
iceberg.ParquetFile, partition, nil, nil, 1, 128)
+ require.NoError(t, err)
+ dead, err := iceberg.NewDataFileBuilder(spec,
iceberg.EntryContentPosDeletes, location+"/data/dead-delete.parquet",
iceberg.ParquetFile, map[int]any{1000: int64(8)}, nil, nil, 1, 128)
+ require.NoError(t, err)
+ liveFile := live.ReferencedDataFile(data.Build().FilePath()).Build()
+ deadFile := dead.Build()
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(liveFile,
deadFile).Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ planned, err := tbl.PlanDanglingDeleteFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, planned, 1)
+ require.Equal(t, deadFile.FilePath(), planned[0].FilePath())
+ result, err := tbl.RemoveDanglingDeleteFiles(t.Context())
+ require.NoError(t, err)
+ require.Equal(t, 1, result.RemovedDeleteFiles)
+ updated := result.Table
+ require.Empty(t, mustPlanDangling(t, updated))
+ tasks, err := updated.Scan().PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ require.Len(t, tasks[0].DeleteFiles, 1)
+ require.Equal(t, liveFile.FilePath(),
tasks[0].DeleteFiles[0].FilePath())
+ // The stale table must not replay the same cleanup on a moved branch,
+ // even with retries enabled.
+ attemptsBefore := cat.attempts.Load()
+ _, err = tbl.RemoveDanglingDeleteFiles(t.Context())
+ require.ErrorIs(t, err, table.ErrCommitFailed)
+ require.Equal(t, attemptsBefore+1, cat.attempts.Load(),
+ "cleanup must not retry after a pinned branch head moves")
+}
+
+func TestRemoveDanglingDeleteFilesUnpartitionedCleanup(t *testing.T) {
+ ctx := t.Context()
+ tbl, _ := newConcurrentRewriteTestTable(t)
+ dataPath := tbl.Location() + "/data/live.parquet"
+ missingPath := tbl.Location() + "/data/missing.parquet"
+ posPath := tbl.Location() + "/data/dangling-pos.parquet"
+ eqPath := tbl.Location() + "/data/dangling-eq.parquet"
+
+ data, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec,
iceberg.EntryContentData,
+ dataPath, iceberg.ParquetFile, nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+ pos, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec,
iceberg.EntryContentPosDeletes,
+ posPath, iceberg.ParquetFile, nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+ eq, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec,
iceberg.EntryContentEqDeletes,
+ eqPath, iceberg.ParquetFile, nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+
+ // An explicit position-delete target is absent, and an equality delete
+ // does not apply to data from the same sequence number.
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).
+ AddRows(data.Build()).
+ AddDeletes(pos.ReferencedDataFile(missingPath).Build(),
eq.EqualityFieldIDs([]int{1}).Build()).
+ Commit(ctx))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ planned, err := tbl.PlanDanglingDeleteFiles(ctx)
+ require.NoError(t, err)
+ var paths []string
+ for _, file := range planned {
+ paths = append(paths, file.FilePath())
+ }
+ require.ElementsMatch(t, []string{posPath, eqPath}, paths)
+
+ result, err := tbl.RemoveDanglingDeleteFiles(ctx)
+ require.NoError(t, err)
+ require.Equal(t, 2, result.RemovedDeleteFiles)
+ updated := result.Table
+ require.Empty(t, mustPlanDangling(t, updated))
+
+ tasks, err := updated.Scan().PlanFiles(ctx)
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ require.Equal(t, dataPath, tasks[0].File.FilePath())
+ require.Empty(t, tasks[0].DeleteFiles)
+ require.Empty(t, tasks[0].EqualityDeleteFiles)
+}
+
+func mustPlanDangling(t *testing.T, tbl *table.Table) []iceberg.DataFile {
+ t.Helper()
+ files, err := tbl.PlanDanglingDeleteFiles(t.Context())
+ require.NoError(t, err)
+
+ return files
+}
+
+func TestRemoveDanglingDeleteFilesPrunesEqualityBounds(t *testing.T) {
+ ctx := t.Context()
+ tbl, _ := newConcurrentRewriteTestTable(t)
+ makeFile := func(content iceberg.ManifestEntryContent, name, value
string) iceberg.DataFile {
+ t.Helper()
+ builder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, content,
tbl.Location()+"/data/"+name,
+ iceberg.ParquetFile, nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+ builder.LowerBoundValues(map[int][]byte{2: []byte(value)})
+ builder.UpperBoundValues(map[int][]byte{2: []byte(value)})
+ builder.ValueCounts(map[int]int64{2: 1})
+ builder.NullValueCounts(map[int]int64{2: 0})
+ if content == iceberg.EntryContentEqDeletes {
+ builder.EqualityFieldIDs([]int{2})
+ }
+
+ return builder.Build()
+ }
+
+ data := makeFile(iceberg.EntryContentData, "live.parquet", "a")
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddRows(data).Commit(ctx))
+ var err error
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ liveEquality := makeFile(iceberg.EntryContentEqDeletes,
"live-eq.parquet", "a")
+ danglingEquality := makeFile(iceberg.EntryContentEqDeletes,
"dangling-eq.parquet", "z")
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(liveEquality,
danglingEquality).Commit(ctx))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ tasks, err := tbl.Scan().PlanFiles(ctx)
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ require.Len(t, tasks[0].EqualityDeleteFiles, 1)
+ require.Equal(t, liveEquality.FilePath(),
tasks[0].EqualityDeleteFiles[0].FilePath())
+
+ planned, err := tbl.PlanDanglingDeleteFiles(ctx)
+ require.NoError(t, err)
+ require.Len(t, planned, 1)
+ require.Equal(t, danglingEquality.FilePath(), planned[0].FilePath())
+
+ result, err := tbl.RemoveDanglingDeleteFiles(ctx)
+ require.NoError(t, err)
+ require.Equal(t, 1, result.RemovedDeleteFiles)
+ updated := result.Table
+ require.Empty(t, mustPlanDangling(t, updated))
+}
+
+func TestRemoveDanglingDeleteFilesTransactionStagesAfterRowDelta(t *testing.T)
{
+ ctx := t.Context()
+ tbl, _ := newConcurrentRewriteTestTable(t)
+ data := newDataFile(t, tbl.Location()+"/data/live.parquet")
+ posBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
+ tbl.Location()+"/data/dangling.parquet", iceberg.ParquetFile,
nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).
+ AddRows(data).
+
AddDeletes(posBuilder.ReferencedDataFile(tbl.Location()+"/data/missing.parquet").Build()).
+ Commit(ctx))
+
+ result, err := tx.RemoveDanglingDeleteFiles(ctx)
+ require.NoError(t, err)
+ require.Nil(t, result.Table, "transaction action stages rather than
commits")
+ require.Equal(t, 1, result.RemovedDeleteFiles)
+ require.Nil(t, tbl.CurrentSnapshot(), "base table must remain
unchanged")
+
+ committed, err := tx.Commit(ctx)
+ require.NoError(t, err)
+ require.Empty(t, mustPlanDangling(t, committed))
+ tasks, err := committed.Scan().PlanFiles(ctx)
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ require.Equal(t, data.FilePath(), tasks[0].File.FilePath())
+ require.Empty(t, tasks[0].DeleteFiles)
+}
+
+func TestRemoveDanglingDeleteFilesTransactionTargetsBranch(t *testing.T) {
+ ctx := t.Context()
+ tbl, _ := newConcurrentRewriteTestTable(t)
+ data := newDataFile(t, tbl.Location()+"/data/main.parquet")
+ posBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
+ tbl.Location()+"/data/main-delete.parquet",
iceberg.ParquetFile, nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+ liveDelete := posBuilder.ReferencedDataFile(data.FilePath()).Build()
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddRows(data).Commit(ctx))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+ tx = tbl.NewTransaction()
+ require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(liveDelete).Commit(ctx))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ // Fork from main, then remove the target data only on the branch.
+ branchTx, err := tbl.NewTransactionOnBranch("cleanup")
+ require.NoError(t, err)
+ rewrite := branchTx.NewRewrite(nil)
+ rewrite.DeleteFile(data)
+ require.NoError(t, rewrite.Commit(ctx))
+ tbl, err = branchTx.Commit(ctx)
+ require.NoError(t, err)
+ require.Empty(t, mustPlanDangling(t, tbl), "delete remains live on
main")
+
+ branchTx, err = tbl.NewTransactionOnBranch("cleanup")
+ require.NoError(t, err)
+ require.NoError(t,
branchTx.SetProperties(iceberg.Properties{"dangling-cleanup-test": "yes"}))
+ planned, err := branchTx.PlanDanglingDeleteFiles(ctx)
+ require.NoError(t, err)
+ require.Len(t, planned, 1)
+ require.Equal(t, liveDelete.FilePath(), planned[0].FilePath())
+ result, err := branchTx.RemoveDanglingDeleteFiles(ctx)
+ require.NoError(t, err)
+ require.Nil(t, result.Table)
+ require.Equal(t, 1, result.RemovedDeleteFiles)
+
+ updated, err := branchTx.Commit(ctx)
+ require.NoError(t, err)
+ require.Equal(t, "yes", updated.Properties()["dangling-cleanup-test"])
+ mainData, mainDeletes := liveFilesOnRef(t, updated, table.MainBranch)
+ require.Contains(t, mainData, data.FilePath())
+ require.Contains(t, mainDeletes, liveDelete.FilePath())
+ _, branchDeletes := liveFilesOnRef(t, updated, "cleanup")
+ require.Empty(t, branchDeletes)
+}
diff --git a/table/remove_dangling_delete_files_test.go
b/table/remove_dangling_delete_files_test.go
new file mode 100644
index 000000000..5634f4015
--- /dev/null
+++ b/table/remove_dangling_delete_files_test.go
@@ -0,0 +1,94 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package table
+
+import (
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/require"
+)
+
+func TestDecideDanglingDeleteFilesAllTypes(t *testing.T) {
+ makeDelete := func(path string, content iceberg.ManifestEntryContent)
iceberg.DataFile {
+ t.Helper()
+ builder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, content, path,
+ iceberg.ParquetFile, nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+ if content == iceberg.EntryContentEqDeletes {
+ builder.EqualityFieldIDs([]int{1})
+ }
+
+ return builder.Build()
+ }
+
+ livePosition := makeDelete("live-pos", iceberg.EntryContentPosDeletes)
+ danglingPosition := makeDelete("dangling-pos",
iceberg.EntryContentPosDeletes)
+ liveEquality := makeDelete("live-eq", iceberg.EntryContentEqDeletes)
+ danglingEquality := makeDelete("dangling-eq",
iceberg.EntryContentEqDeletes)
+
+ got := decideDanglingDeleteFiles(
+ []iceberg.DataFile{livePosition, danglingPosition,
liveEquality, danglingEquality},
+ []FileScanTask{
+ {DeleteFiles: []iceberg.DataFile{livePosition},
EqualityDeleteFiles: []iceberg.DataFile{liveEquality}},
+ {DeleteFiles: []iceberg.DataFile{livePosition}},
+ })
+ require.Len(t, got, 2)
+ require.Equal(t, danglingPosition.FilePath(), got[0].FilePath())
+ require.Equal(t, danglingEquality.FilePath(), got[1].FilePath())
+}
+
+func TestDecideDanglingDeleteFilesSharedPuffin(t *testing.T) {
+ // A Puffin container can hold multiple DV blobs. Seeing one live vector
+ // must not preserve a different, dangling vector in the same container.
+ makeVector := func(reference string, offset, length int64)
iceberg.DataFile {
+ return &dvMockDataFile{
+ mockDataFile: mockDataFile{
+ path: "shared.puffin",
+ contentType: iceberg.EntryContentPosDeletes,
+ format: iceberg.PuffinFile,
+ },
+ referencedDataFile: &reference,
+ contentOffset: &offset,
+ contentSizeInBytes: &length,
+ }
+ }
+
+ live := makeVector("live.parquet", 0, 100)
+ dangling := makeVector("missing.parquet", 100, 100)
+ got := decideDanglingDeleteFiles(
+ []iceberg.DataFile{live, dangling},
+ []FileScanTask{{DeletionVectorFiles: []iceberg.DataFile{live}}},
+ )
+ require.Len(t, got, 1)
+ require.Equal(t, dangling.ReferencedDataFile(),
got[0].ReferencedDataFile())
+ require.Equal(t, dangling.ContentOffset(), got[0].ContentOffset())
+}
+
+func TestRemoveDanglingDeleteFilesUnpartitionedNoop(t *testing.T) {
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ )
+ meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec,
UnsortedSortOrder, t.TempDir(), nil)
+ require.NoError(t, err)
+ tbl := New([]string{"db", "example"}, meta, "", nil, nil)
+ result, err := tbl.RemoveDanglingDeleteFiles(t.Context())
+ require.NoError(t, err)
+ require.Zero(t, result.RemovedDeleteFiles)
+ require.Equal(t, tbl.Metadata(), result.Table.Metadata())
+}
