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 f3c9f7226 fix(table): skip rows already removed by position deletes in
merge-on-read deletes (#2422)
f3c9f7226 is described below
commit f3c9f72261e34c9365582cfd51479820e004c0c3
Author: Anirudh Sriram <[email protected]>
AuthorDate: Thu Nov 5 09:54:07 2026 -0800
fix(table): skip rows already removed by position deletes in merge-on-read
deletes (#2422)
* fix(table): skip rows already removed by position deletes in
merge-on-read deletes
makePositionDeleteRecordsForFilter built its scan tasks without the
files' existing position delete files, so a v2 merge-on-read delete
matched rows an earlier delete had already removed and wrote position
deletes for them again. Take each file's delete files from scan
planning so those rows are filtered out first.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Anirudh Sriram <[email protected]>
* fix(table): pin the existing-deletes scan to the transaction's branch
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Anirudh Sriram <[email protected]>
---------
Signed-off-by: Anirudh Sriram <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
table/mor_delete_pruning_test.go | 74 ++++++++++++++++++++++++++++++++++++++++
table/transaction_delete.go | 29 ++++++++++++++--
2 files changed, 100 insertions(+), 3 deletions(-)
diff --git a/table/mor_delete_pruning_test.go b/table/mor_delete_pruning_test.go
index b538a02e8..2a77b2267 100644
--- a/table/mor_delete_pruning_test.go
+++ b/table/mor_delete_pruning_test.go
@@ -291,3 +291,77 @@ func idsInTable(t *testing.T, tbl *table.Table) []int64 {
return ids
}
+
+// TestMergeOnReadDeleteSkipsRowsAlreadyDeleted checks that a v2 merge-on-read
+// delete doesn't write position deletes for rows an earlier delete removed.
+// Such deletes change nothing, but they grow the delete files on every update
+// and make the snapshot's total-position-deletes overcount.
+func TestMergeOnReadDeleteSkipsRowsAlreadyDeleted(t *testing.T) {
+ ctx := context.Background()
+ tbl := newMergeOnReadTestTable(t)
+
+ arrowSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false},
+ {Name: "data", Type: arrow.BinaryTypes.String, Nullable: true},
+ }, nil)
+ data, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema,
[]string{
+ `[{"id":1,"data":"a"},{"id":2,"data":"b"},{"id":3,"data":"c"}]`,
+ })
+ require.NoError(t, err)
+ defer data.Release()
+
+ tbl, err = tbl.Append(ctx, array.NewTableReader(data, -1), nil)
+ require.NoError(t, err)
+
+ tbl, err = tbl.Delete(ctx, iceberg.EqualTo(iceberg.Reference("id"),
int64(1)), nil)
+ require.NoError(t, err)
+
+ tbl, err = tbl.Delete(ctx, iceberg.IsIn(iceberg.Reference("id"),
int64(1), int64(2)), nil)
+ require.NoError(t, err)
+
+ assert.Equal(t, []int64{3}, idsInTable(t, tbl))
+ summary := tbl.CurrentSnapshot().Summary.Properties
+ assert.Equal(t, "1", summary["added-position-deletes"], "id=1 was
already deleted")
+ assert.Equal(t, "2", summary["total-position-deletes"])
+
+ t.Run("branch", func(t *testing.T) {
+ const branch = "audit"
+ tbl := newMergeOnReadTestTable(t)
+ tbl, err := tbl.Append(ctx, array.NewTableReader(data, -1), nil)
+ require.NoError(t, err)
+
+ // The branch forks from main here, before main deletes id=1.
+ extra, err := array.TableFromJSON(memory.DefaultAllocator,
arrowSchema, []string{`[{"id":4,"data":"d"}]`})
+ require.NoError(t, err)
+ defer extra.Release()
+ txn, err := tbl.NewTransactionOnBranch(branch)
+ require.NoError(t, err)
+ require.NoError(t, txn.Append(ctx, array.NewTableReader(extra,
-1), nil))
+ tbl, err = txn.Commit(ctx)
+ require.NoError(t, err)
+
+ tbl, err = tbl.Delete(ctx,
iceberg.EqualTo(iceberg.Reference("id"), int64(1)), nil)
+ require.NoError(t, err)
+
+ txn, err = tbl.NewTransactionOnBranch(branch)
+ require.NoError(t, err)
+ require.NoError(t, txn.Delete(ctx,
iceberg.IsIn(iceberg.Reference("id"), int64(1), int64(2)), nil))
+ tbl, err = txn.Commit(ctx)
+ require.NoError(t, err)
+
+ assert.Equal(t, []int64{2, 3}, idsInTable(t, tbl), "main only
deleted id=1")
+ scan, err := tbl.Scan().UseRef(branch)
+ require.NoError(t, err)
+ _, itr, err := scan.ToArrowRecords(ctx)
+ require.NoError(t, err)
+ var ids []int64
+ for rec, err := range itr {
+ require.NoError(t, err)
+ col :=
rec.Column(rec.Schema().FieldIndices("id")[0]).(*array.Int64)
+ ids = append(ids, col.Int64Values()...)
+ rec.Release()
+ }
+ slices.Sort(ids)
+ assert.Equal(t, []int64{3, 4}, ids, "main's delete of id=1
doesn't count on the branch")
+ })
+}
diff --git a/table/transaction_delete.go b/table/transaction_delete.go
index 6ee5dad37..69959fd89 100644
--- a/table/transaction_delete.go
+++ b/table/transaction_delete.go
@@ -953,12 +953,35 @@ func (t *Transaction)
makePositionDeleteRecordsForFilter(ctx context.Context, fs
return nil, err
}
+ // Rows that existing position deletes already removed must not be
deleted
+ // again, so each task carries its file's position delete files from
scan
+ // planning, which matches them by sequence number, partition and path.
The
+ // scan is pinned to the snapshot files came from, so a branch
transaction
+ // doesn't apply main's deletes.
+ scanOpts := []ScanOption{WithRowFilter(filter),
WithCaseSensitive(caseSensitive)}
+ if snap := t.planningSnapshot(meta); snap != nil {
+ scanOpts = append(scanOpts, WithSnapshotID(snap.SnapshotID))
+ }
+ scan, err := t.Scan(scanOpts...)
+ if err != nil {
+ return nil, err
+ }
+ planned, err := scan.PlanFiles(ctx)
+ if err != nil {
+ return nil, err
+ }
+ existingDeletes := make(map[string][]iceberg.DataFile, len(planned))
+ for _, task := range planned {
+ existingDeletes[task.File.FilePath()] = task.DeleteFiles
+ }
+
tasks := make([]FileScanTask, 0, len(files))
for _, f := range files {
tasks = append(tasks, FileScanTask{
- File: f,
- Start: 0,
- Length: f.FileSizeBytes(),
+ File: f,
+ DeleteFiles: existingDeletes[f.FilePath()],
+ Start: 0,
+ Length: f.FileSizeBytes(),
})
}
