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 1df76d8c9 fix(table): apply existing deletes when cow rewrites a data
file (#2046)
1df76d8c9 is described below
commit 1df76d8c901f63b22be4d6e98fbc8414880ced9a
Author: Badal Prasad Singh <[email protected]>
AuthorDate: Thu Oct 15 23:51:23 2026 +0530
fix(table): apply existing deletes when cow rewrites a data file (#2046)
* fix: apply existing deletes
Signed-off-by: badalprasadsingh <[email protected]>
* minor
Signed-off-by: badalprasadsingh <[email protected]>
* minor
Signed-off-by: badalprasadsingh <[email protected]>
* fix: merge
Signed-off-by: badalprasadsingh <[email protected]>
* minor
Signed-off-by: badalprasadsingh <[email protected]>
* minor
Signed-off-by: badalprasadsingh <[email protected]>
* minor
Signed-off-by: badalprasadsingh <[email protected]>
* fix: minor
Signed-off-by: badalprasadsingh <[email protected]>
---------
Signed-off-by: badalprasadsingh <[email protected]>
---
table/cow_rewrite_deletes_test.go | 457 ++++++++++++++++++++++++++++++++++++++
table/mor_delete_pruning_test.go | 4 +-
table/no_replay_removals_test.go | 52 +++++
table/table.go | 16 ++
table/transaction.go | 335 ++++++++++++++++------------
5 files changed, 719 insertions(+), 145 deletions(-)
diff --git a/table/cow_rewrite_deletes_test.go
b/table/cow_rewrite_deletes_test.go
new file mode 100644
index 000000000..aafad6493
--- /dev/null
+++ b/table/cow_rewrite_deletes_test.go
@@ -0,0 +1,457 @@
+// 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 (
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+type cowRewriteOp struct {
+ name string
+ run func(*testing.T, *table.Table) *table.Table
+ wantIDs []int64
+ wantRecords int64
+}
+
+// TestCoWRewriteKeepsRowsDeleted rewrites a data file that already has a
delete on id=1.
+// Once the file is replaced that delete no longer reaches the copied rows, so
the rewrite itself must drop id=1.
+func TestCoWRewriteKeepsRowsDeleted(t *testing.T) {
+ deleteID1 := func(t *testing.T, tbl *table.Table) *table.Table {
+ t.Helper()
+
+ return mergeOnReadDelete(t, tbl, 1)
+ }
+ equalityDeleteID1 := func(t *testing.T, tbl *table.Table) *table.Table {
+ t.Helper()
+
+ return appendEqualityDelete(t, tbl, []int{1}, `[{"id": 1}]`)
+ }
+
+ deletes := []struct {
+ name string
+ formatVersion string
+ seed func(*testing.T, *table.Table) *table.Table
+ seededDVs int
+ }{
+ {name: "position delete", formatVersion: "2", seed: deleteID1},
+ {name: "deletion vector", formatVersion: "3", seed: deleteID1,
seededDVs: 1},
+ {name: "equality delete on v2", formatVersion: "2", seed:
equalityDeleteID1},
+ {name: "equality delete on v3", formatVersion: "3", seed:
equalityDeleteID1},
+ }
+
+ matchID2 := iceberg.EqualTo(iceberg.Reference("id"), int64(2))
+ operations := []cowRewriteOp{
+ {
+ name: "copy-on-write delete",
+ run: func(t *testing.T, tbl *table.Table)
*table.Table { return copyOnWriteDelete(t, tbl, matchID2) },
+ wantIDs: []int64{3, 4, 5},
+ wantRecords: 3,
+ },
+ {
+ name: "filtered overwrite",
+ run: func(t *testing.T, tbl *table.Table) *table.Table {
+ return filteredOverwrite(t, tbl, matchID2,
`[{"id": 6}]`)
+ },
+ wantIDs: []int64{3, 4, 5, 6},
+ wantRecords: 4,
+ },
+ }
+
+ for _, del := range deletes {
+ for _, op := range operations {
+ t.Run(del.name+"/"+op.name, func(t *testing.T) {
+ tbl := newMergeOnReadTestTableVersion(t,
del.formatVersion)
+ tbl = appendRowsOnRef(t, tbl, table.MainBranch,
1, 2, 3, 4, 5)
+ tbl = del.seed(t, tbl)
+ require.Equal(t, []int64{2, 3, 4, 5},
idsInTable(t, tbl))
+ require.Equal(t, del.seededDVs, liveDVCount(t,
tbl))
+
+ var rowIDsBefore map[int64]int64
+ if del.formatVersion == "3" {
+ rowIDsBefore = readRowIDsByID(t,
t.Context(), tbl)
+ require.Len(t, rowIDsBefore, 4)
+ }
+
+ tbl = op.run(t, tbl)
+
+ assert.Equal(t, op.wantIDs, idsInTable(t, tbl),
"id=1 was deleted before the rewrite and must stay deleted")
+ assert.Equal(t, op.wantRecords,
liveDataRecordCount(t, tbl), "the rewritten file must not carry the
already-deleted row")
+ assert.Zero(t, liveDVCount(t, tbl), "the
deletion vector of the replaced file must be removed with it")
+
+ if rowIDsBefore != nil {
+ rowIDsAfter := readRowIDsByID(t,
t.Context(), tbl)
+ for _, id := range []int64{3, 4, 5} {
+ assert.Equal(t,
rowIDsBefore[id], rowIDsAfter[id],
+ "id=%d must keep its
_row_id through the rewrite", id)
+ }
+ }
+ })
+ }
+ }
+}
+
+// TestCoWRewriteKeepsEachFilesOwnRowsDeleted rewrites two partitioned files
in one operation.
+// Each carrying a deletion vector on a different row position.
+// Every rewritten file must drop the row its own vector removed, and no other.
+func TestCoWRewriteKeepsEachFilesOwnRowsDeleted(t *testing.T) {
+ matchID2Or5 := iceberg.NewOr(
+ iceberg.EqualTo(iceberg.Reference("id"), int64(2)),
+ iceberg.EqualTo(iceberg.Reference("id"), int64(5)),
+ )
+
+ operations := []cowRewriteOp{
+ {
+ name: "copy-on-write delete",
+ run: func(t *testing.T, tbl *table.Table)
*table.Table { return copyOnWriteDelete(t, tbl, matchID2Or5) },
+ wantIDs: []int64{3, 4},
+ wantRecords: 2,
+ },
+ {
+ name: "filtered overwrite",
+ run: func(t *testing.T, tbl *table.Table) *table.Table {
+ return filteredOverwrite(t, tbl, matchID2Or5,
`[{"id": 7, "category": "x"}]`)
+ },
+ wantIDs: []int64{3, 4, 7},
+ wantRecords: 3,
+ },
+ }
+
+ recordsJSON := `[
+ {"id": 1, "category": "x"}, {"id": 2, "category": "x"}, {"id":
3, "category": "x"},
+ {"id": 4, "category": "y"}, {"id": 5, "category": "y"}, {"id":
6, "category": "y"}
+ ]`
+
+ for _, op := range operations {
+ t.Run(op.name, func(t *testing.T) {
+ tbl := newPartitionedMergeOnReadTestTable(t)
+ tbl = appendRows(t, tbl, recordsJSON)
+
+ // Distinct positions per file: id=1 is first in its
file, id=6 is last in its own.
+ // So applying one file's vector to the other changes
the result.
+ tbl = mergeOnReadDelete(t, tbl, 1)
+ tbl = mergeOnReadDelete(t, tbl, 6)
+ require.Equal(t, 2, liveDVCount(t, tbl))
+ require.Equal(t, []int64{2, 3, 4, 5}, idsInTable(t,
tbl))
+
+ tbl = op.run(t, tbl)
+
+ assert.Equal(t, op.wantIDs, idsInTable(t, tbl), "each
rewritten file must drop only the row its own deletion vector removed")
+ assert.Equal(t, op.wantRecords, liveDataRecordCount(t,
tbl))
+ assert.Zero(t, liveDVCount(t, tbl), "both replaced
files must take their deletion vectors with them")
+ })
+ }
+}
+
+func TestCoWRewriteAppliesStackedDeletes(t *testing.T) {
+ matchID2 := iceberg.EqualTo(iceberg.Reference("id"), int64(2))
+ operations := []cowRewriteOp{
+ {
+ name: "copy-on-write delete",
+ run: func(t *testing.T, tbl *table.Table)
*table.Table { return copyOnWriteDelete(t, tbl, matchID2) },
+ wantIDs: []int64{4, 5},
+ wantRecords: 2,
+ },
+ {
+ name: "filtered overwrite",
+ run: func(t *testing.T, tbl *table.Table) *table.Table {
+ return filteredOverwrite(t, tbl, matchID2,
`[{"id": 6}]`)
+ },
+ wantIDs: []int64{4, 5, 6},
+ wantRecords: 3,
+ },
+ }
+
+ for _, op := range operations {
+ t.Run(op.name, func(t *testing.T) {
+ tbl := newMergeOnReadTestTableVersion(t, "3")
+ tbl = appendRowsOnRef(t, tbl, table.MainBranch, 1, 2,
3, 4, 5)
+ tbl = mergeOnReadDelete(t, tbl, 1)
+ tbl = appendEqualityDelete(t, tbl, []int{1}, `[{"id":
3}]`)
+ require.Equal(t, 1, liveDVCount(t, tbl))
+ require.Equal(t, []int64{2, 4, 5}, idsInTable(t, tbl))
+
+ tbl = op.run(t, tbl)
+
+ assert.Equal(t, op.wantIDs, idsInTable(t, tbl), "id=1
and id=3 were deleted before the rewrite and must stay deleted")
+ assert.Equal(t, op.wantRecords, liveDataRecordCount(t,
tbl), "the rewritten file must carry neither already-deleted row")
+ assert.Zero(t, liveDVCount(t, tbl))
+ })
+ }
+}
+
+func TestDroppedFilesTakeTheirDeletionVectors(t *testing.T) {
+ matchY := iceberg.EqualTo(iceberg.Reference("category"), "y")
+ matchYOrID2 := iceberg.NewOr(matchY,
iceberg.EqualTo(iceberg.Reference("id"), int64(2)))
+ mergeOnReadDeleteWhere := func(filter iceberg.BooleanExpression)
func(*testing.T, *table.Table) *table.Table {
+ return func(t *testing.T, tbl *table.Table) *table.Table {
+ tbl, err := tbl.Delete(t.Context(), filter, nil)
+ require.NoError(t, err)
+
+ return tbl
+ }
+ }
+
+ cases := []struct {
+ name string
+ run func(*testing.T, *table.Table) *table.Table
+ wantIDs []int64
+ wantDVs int
+ }{
+ {
+ name: "copy-on-write delete drops a file",
+ run: func(t *testing.T, tbl *table.Table)
*table.Table { return copyOnWriteDelete(t, tbl, matchY) },
+ wantIDs: []int64{2, 3},
+ wantDVs: 1,
+ },
+ {
+ name: "filtered overwrite drops a file",
+ run: func(t *testing.T, tbl *table.Table) *table.Table {
+ return filteredOverwrite(t, tbl, matchY,
`[{"id": 7, "category": "y"}]`)
+ },
+ wantIDs: []int64{2, 3, 7},
+ wantDVs: 1,
+ },
+ {
+ name: "copy-on-write delete drops a file and
rewrites another",
+ run: func(t *testing.T, tbl *table.Table)
*table.Table { return copyOnWriteDelete(t, tbl, matchYOrID2) },
+ wantIDs: []int64{3},
+ wantDVs: 0,
+ },
+ {
+ name: "filtered overwrite drops a file and rewrites
another",
+ run: func(t *testing.T, tbl *table.Table) *table.Table {
+ return filteredOverwrite(t, tbl, matchYOrID2,
`[{"id": 7, "category": "y"}]`)
+ },
+ wantIDs: []int64{3, 7},
+ wantDVs: 0,
+ },
+ {
+ name: "merge-on-read delete drops a file",
+ run: mergeOnReadDeleteWhere(matchY),
+ wantIDs: []int64{2, 3},
+ wantDVs: 1,
+ },
+ {
+ // The partially deleted file's vector is replaced by a
merged one, so one DV stays live.
+ name: "merge-on-read delete drops a file and deletes
rows from another",
+ run: mergeOnReadDeleteWhere(matchYOrID2),
+ wantIDs: []int64{3},
+ wantDVs: 1,
+ },
+ }
+
+ recordsJSON := `[
+ {"id": 1, "category": "x"}, {"id": 2, "category": "x"}, {"id":
3, "category": "x"},
+ {"id": 4, "category": "y"}, {"id": 5, "category": "y"}, {"id":
6, "category": "y"}
+ ]`
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ tbl := newPartitionedMergeOnReadTestTable(t)
+ tbl = appendRows(t, tbl, recordsJSON)
+ tbl = mergeOnReadDelete(t, tbl, 1)
+ tbl = mergeOnReadDelete(t, tbl, 6)
+ require.Equal(t, 2, liveDVCount(t, tbl))
+ require.Equal(t, []int64{2, 3, 4, 5}, idsInTable(t,
tbl))
+
+ tbl = tc.run(t, tbl)
+
+ assert.Equal(t, tc.wantIDs, idsInTable(t, tbl))
+ assert.Equal(t, tc.wantDVs, liveDVCount(t, tbl), "the
deletion vector of every dropped file must be removed with it")
+ })
+ }
+}
+
+// TestCoWRewriteToEmptyWritesNoDataFile rewrites a file whose only surviving
row is the one the operation removes,
+// so the rewrite has no rows to write.
+// No empty data file may stay live, and the file's deletion vector must go
too.
+func TestCoWRewriteToEmptyWritesNoDataFile(t *testing.T) {
+ matchID2 := iceberg.EqualTo(iceberg.Reference("id"), int64(2))
+ cases := []struct {
+ name string
+ run func(*testing.T, *table.Table) *table.Table
+ wantIDs []int64
+ wantDataFiles int
+ }{
+ {
+ name: "copy-on-write delete",
+ run: func(t *testing.T, tbl *table.Table)
*table.Table { return copyOnWriteDelete(t, tbl, matchID2) },
+ wantIDs: nil,
+ wantDataFiles: 0,
+ },
+ {
+ name: "filtered overwrite",
+ run: func(t *testing.T, tbl *table.Table) *table.Table {
+ return filteredOverwrite(t, tbl, matchID2,
`[{"id": 6}]`)
+ },
+ wantIDs: []int64{6},
+ wantDataFiles: 1,
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ tbl := newMergeOnReadTestTableVersion(t, "3")
+ tbl = appendRowsOnRef(t, tbl, table.MainBranch, 1, 2)
+ tbl = mergeOnReadDelete(t, tbl, 1)
+ require.Equal(t, 1, liveDVCount(t, tbl))
+ require.Equal(t, []int64{2}, idsInTable(t, tbl))
+
+ tbl = tc.run(t, tbl)
+
+ assert.Equal(t, tc.wantIDs, idsInTable(t, tbl))
+ dataPaths, _ := liveFilesOnRef(t, tbl, table.MainBranch)
+ assert.Len(t, dataPaths, tc.wantDataFiles, "a rewrite
with no surviving rows must not leave an empty data file live")
+ assert.Zero(t, liveDVCount(t, tbl))
+ })
+ }
+}
+
+// TestCoWRewriteSkipsDeletesThatDoNotApply re-inserts id=1 after an equality
delete on it.
+// The delete has a lower sequence number than the new file, so rewriting that
file must keep the row,
+// while the untouched older file must keep its own id=1 hidden.
+func TestCoWRewriteSkipsDeletesThatDoNotApply(t *testing.T) {
+ matchID4 := iceberg.EqualTo(iceberg.Reference("id"), int64(4))
+ operations := []cowRewriteOp{
+ {
+ name: "copy-on-write delete",
+ run: func(t *testing.T, tbl *table.Table)
*table.Table { return copyOnWriteDelete(t, tbl, matchID4) },
+ wantIDs: []int64{1, 2, 3, 5},
+ wantRecords: 5,
+ },
+ {
+ name: "filtered overwrite",
+ run: func(t *testing.T, tbl *table.Table) *table.Table {
+ return filteredOverwrite(t, tbl, matchID4,
`[{"id": 6}]`)
+ },
+ wantIDs: []int64{1, 2, 3, 5, 6},
+ wantRecords: 6,
+ },
+ }
+
+ for _, op := range operations {
+ t.Run(op.name, func(t *testing.T) {
+ tbl := newMergeOnReadTestTableVersion(t, "2")
+ tbl = appendRowsOnRef(t, tbl, table.MainBranch, 1, 2, 3)
+ tbl = appendEqualityDelete(t, tbl, []int{1}, `[{"id":
1}]`)
+ tbl = appendRowsOnRef(t, tbl, table.MainBranch, 1, 4, 5)
+ require.Equal(t, []int64{1, 2, 3, 4, 5}, idsInTable(t,
tbl))
+
+ tbl = op.run(t, tbl)
+
+ assert.Equal(t, op.wantIDs, idsInTable(t, tbl), "the
re-inserted id=1 is newer than the equality delete and must survive the
rewrite")
+ assert.Equal(t, op.wantRecords, liveDataRecordCount(t,
tbl), "only the file holding id=4 is rewritten; the other keeps all 3 of its
rows")
+ })
+ }
+}
+
+// mergeOnReadDelete deletes id on a table configured for merge-on-read
deletes.
+func mergeOnReadDelete(t *testing.T, tbl *table.Table, id int64) *table.Table {
+ t.Helper()
+
+ tbl, err := tbl.Delete(t.Context(),
iceberg.EqualTo(iceberg.Reference("id"), id), nil)
+ require.NoError(t, err)
+
+ return tbl
+}
+
+// copyOnWriteDelete switches the table to copy-on-write in its own commit,
then deletes the rows matching filter.
+func copyOnWriteDelete(t *testing.T, tbl *table.Table, filter
iceberg.BooleanExpression) *table.Table {
+ t.Helper()
+
+ tx := tbl.NewTransaction()
+ require.NoError(t,
tx.SetProperties(iceberg.Properties{table.WriteDeleteModeKey:
table.WriteModeCopyOnWrite}))
+ tbl, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ tbl, err = tbl.Delete(t.Context(), filter, nil)
+ require.NoError(t, err)
+
+ return tbl
+}
+
+// filteredOverwrite replaces the rows matching filter with the given records.
+func filteredOverwrite(t *testing.T, tbl *table.Table, filter
iceberg.BooleanExpression, recordsJSON string) *table.Table {
+ t.Helper()
+
+ rdr := rowsReader(t, tbl, recordsJSON)
+ defer rdr.Release()
+
+ tbl, err := tbl.Overwrite(t.Context(), rdr, nil,
table.WithOverwriteFilter(filter))
+ require.NoError(t, err)
+
+ return tbl
+}
+
+func appendRows(t *testing.T, tbl *table.Table, recordsJSON string)
*table.Table {
+ t.Helper()
+
+ rdr := rowsReader(t, tbl, recordsJSON)
+ defer rdr.Release()
+
+ tbl, err := tbl.Append(t.Context(), rdr, nil)
+ require.NoError(t, err)
+
+ return tbl
+}
+
+func rowsReader(t *testing.T, tbl *table.Table, recordsJSON string)
array.RecordReader {
+ t.Helper()
+
+ arrowSchema, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
+ require.NoError(t, err)
+
+ data, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema,
[]string{recordsJSON})
+ require.NoError(t, err)
+ t.Cleanup(data.Release)
+
+ return array.NewTableReader(data, -1)
+}
+
+// liveDataRecordCount sums record_count over the live data files of the
current snapshot,
+// including rows that deletes hide from a scan.
+func liveDataRecordCount(t *testing.T, tbl *table.Table) int64 {
+ t.Helper()
+
+ fs, err := tbl.FS(t.Context())
+ require.NoError(t, err)
+ snapshot := tbl.CurrentSnapshot()
+ require.NotNil(t, snapshot)
+ manifests, err := snapshot.Manifests(fs)
+ require.NoError(t, err)
+
+ var total int64
+ for _, m := range manifests {
+ if m.ManifestContent() != iceberg.ManifestContentData {
+ continue
+ }
+ for entry, err := range m.Entries(fs, true) {
+ require.NoError(t, err)
+ total += entry.DataFile().Count()
+ }
+ }
+
+ return total
+}
diff --git a/table/mor_delete_pruning_test.go b/table/mor_delete_pruning_test.go
index 01dc5585d..b538a02e8 100644
--- a/table/mor_delete_pruning_test.go
+++ b/table/mor_delete_pruning_test.go
@@ -147,7 +147,7 @@ func TestV3MergeOnReadDoubleDeleteMergesDeletionVector(t
*testing.T) {
// TestV3MergeOnReadTripleDeleteMergesDeletionVector extends the double-delete
// guard to a third delete. The second delete supersedes DV #1, leaving it as a
// DELETED-status entry in the deleted-files manifest against the same data
file.
-// The third delete's collectExistingDVs must skip that ghost entry; otherwise
it
+// The third delete's readPlanningDeletes must skip that ghost entry;
otherwise it
// seeds the merged DV from the stale first bitmap (resurrecting id=4, removed
by
// the second delete) and tries to remove a DV that no longer exists. See
#1372.
func TestV3MergeOnReadTripleDeleteMergesDeletionVector(t *testing.T) {
@@ -175,7 +175,7 @@ func TestV3MergeOnReadTripleDeleteMergesDeletionVector(t
*testing.T) {
require.NoError(t, err)
require.Equal(t, 1, liveDVCount(t, tbl), "second delete must merge into
the single DV")
- // Third delete: DV #1 is now a superseded DELETED entry.
collectExistingDVs
+ // Third delete: DV #1 is now a superseded DELETED entry.
readPlanningDeletes
// must ignore it and seed only from the live merged DV.
tbl, err = tbl.Delete(ctx, iceberg.EqualTo(iceberg.Reference("id"),
int64(1)), nil)
require.NoError(t, err)
diff --git a/table/no_replay_removals_test.go b/table/no_replay_removals_test.go
index ec8de8074..34e3b181b 100644
--- a/table/no_replay_removals_test.go
+++ b/table/no_replay_removals_test.go
@@ -297,3 +297,55 @@ func TestReplaceFilesDataOnlyStillReplays(t *testing.T) {
assert.Equal(t, []int64{1, 2, 100}, idsInTable(t, committed),
"the replay must inherit the peer's append alongside the
compacted file")
}
+
+// A copy-on-write delete that removes a deletion vector must fail on a
concurrent append instead of retrying.
+// Without a deletion vector, the same delete still retries and both writes
land.
+func TestCoWDeleteRemovingDVFailsInsteadOfReplaying(t *testing.T) {
+ cases := []struct {
+ name string
+ seededDVs int
+ wantAttempts int32
+ }{
+ {name: "file with a deletion vector fails on the first
conflict", seededDVs: 1, wantAttempts: 1},
+ {name: "file without a deletion vector still retries",
seededDVs: 0, wantAttempts: 2},
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ ctx := t.Context()
+ tbl, cat := newNoReplayV3Table(t)
+ tbl = appendTenRows(t, tbl)
+ if tc.seededDVs > 0 {
+ tbl = mergeOnReadDelete(t, tbl, 7)
+ }
+ require.Equal(t, tc.seededDVs, liveDVCount(t, tbl))
+
+ tx := tbl.NewTransaction()
+ require.NoError(t,
tx.SetProperties(iceberg.Properties{table.WriteDeleteModeKey:
table.WriteModeCopyOnWrite}))
+ tbl, err := tx.Commit(ctx)
+ require.NoError(t, err)
+
+ staleTxn := tbl.NewTransaction()
+ require.NoError(t, staleTxn.Delete(ctx,
iceberg.EqualTo(iceberg.Reference("id"), int64(2)), nil))
+
+ appendRowsOnRef(t, tbl, table.MainBranch, 100)
+
+ attemptsBefore := cat.attempts.Load()
+ _, err = staleTxn.Commit(ctx)
+ assert.Equal(t, attemptsBefore+tc.wantAttempts,
cat.attempts.Load())
+
+ committed, loadErr := cat.LoadTable(ctx,
table.Identifier{"db", "no_replay_removals"})
+ require.NoError(t, loadErr)
+ if tc.seededDVs > 0 {
+ require.Error(t, err)
+ assert.ErrorIs(t, err, table.ErrCommitFailed)
+ assert.Equal(t, []int64{1, 2, 3, 4, 5, 6, 8, 9,
10, 100}, idsInTable(t, committed),
+ "only the peer's append landed; the
stale delete of id 2 must not have")
+ } else {
+ require.NoError(t, err)
+ assert.Equal(t, []int64{1, 3, 4, 5, 6, 7, 8, 9,
10, 100}, idsInTable(t, committed),
+ "the delete replays over the peer's
append, so both writers land")
+ }
+ })
+ }
+}
diff --git a/table/table.go b/table/table.go
index 27950c12c..1d3189024 100644
--- a/table/table.go
+++ b/table/table.go
@@ -310,6 +310,11 @@ func (t Table) Append(ctx context.Context, rdr
array.RecordReader, snapshotProps
//
// New data from the provided table is written to the table regardless of the
filter.
//
+// On a v3 table, every data file the overwrite removes or rewrites takes its
deletion vector with it.
+// A commit that removes a deletion vector is not replayed: if another writer
commits first,
+// it fails with ErrCommitFailed instead of refreshing and retrying.
+// Reload the table and run the overwrite again.
+//
// The batchSize parameter refers to the batch size for reading the input
data, not the batch size for writes.
// The concurrency parameter controls the level of parallelism for manifest
processing and file rewriting and
// can be overridden using the WithOverwriteConcurrency option. Defaults to
runtime.GOMAXPROCS(0).
@@ -338,6 +343,11 @@ func (t Table) OverwriteTable(ctx context.Context, tbl
arrow.Table, batchSize in
//
// New data from the provided RecordReader is written to the table regardless
of the filter.
//
+// On a v3 table, every data file the overwrite removes or rewrites takes its
deletion vector with it.
+// A commit that removes a deletion vector is not replayed: if another writer
commits first,
+// it fails with ErrCommitFailed instead of refreshing and retrying.
+// Reload the table and run the overwrite again.
+//
// The concurrency parameter controls the level of parallelism for manifest
processing and file rewriting and
// can be overridden using the WithOverwriteConcurrency option. Defaults to
runtime.GOMAXPROCS(0).
func (t Table) Overwrite(ctx context.Context, rdr array.RecordReader,
snapshotProps iceberg.Properties, opts ...OverwriteOption) (*Table, error) {
@@ -364,6 +374,12 @@ func (t Table) Overwrite(ctx context.Context, rdr
array.RecordReader, snapshotPr
// - Strict evaluator determines if all rows in a file must match the filter
// - Files that pass inclusive but not strict evaluation are rewritten with
filtered data
//
+// On a v3 table, every data file the delete removes or rewrites takes its
deletion vector with it.
+// A merge-on-read delete replaces the existing deletion vector of a file it
deletes rows from.
+//
+// A commit that removes a deletion vector is not replayed: if another writer
commits first,
+// it fails with ErrCommitFailed instead of refreshing and retrying, so reload
the table and run the delete again.
+//
// The concurrency parameter controls the level of parallelism for manifest
processing and file rewriting and
// can be overridden using the WithOverwriteConcurrency option. Defaults to
runtime.GOMAXPROCS(0).
func (t Table) Delete(ctx context.Context, filter iceberg.BooleanExpression,
snapshotProps iceberg.Properties, opts ...DeleteOption) (*Table, error) {
diff --git a/table/transaction.go b/table/transaction.go
index 3edadcf89..103ed5ba7 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -25,7 +25,6 @@ import (
"fmt"
"iter"
"log/slog"
- "maps"
"runtime"
"slices"
"strconv"
@@ -2310,6 +2309,11 @@ func (t *Transaction) AddFiles(ctx context.Context,
filePaths []string, snapshot
//
// New data from the provided table is written to the table regardless of the
filter.
//
+// On a v3 table, every data file the overwrite removes or rewrites takes its
deletion vector with it.
+// A commit that removes a deletion vector is not replayed: if another writer
commits first,
+// it fails with ErrCommitFailed instead of refreshing and retrying.
+// Reload the table and run the overwrite again.
+//
// The batchSize parameter refers to the batch size for reading the input
data, not the batch size for writes.
// The concurrency parameter controls the level of parallelism for manifest
processing and file rewriting and
// can be overridden using the WithOverwriteConcurrency option.
@@ -2376,6 +2380,11 @@ func WithOverwriteCaseInsensitive() OverwriteOption {
//
// New data from the provided RecordReader is written to the table regardless
of the filter.
//
+// On a v3 table, every data file the overwrite removes or rewrites takes its
deletion vector with it.
+// A commit that removes a deletion vector is not replayed: if another writer
commits first,
+// it fails with ErrCommitFailed instead of refreshing and retrying.
+// Reload the table and run the overwrite again.
+//
// The concurrency parameter controls the level of parallelism for manifest
processing and file rewriting and
// can be overridden using the WithOverwriteConcurrency option.
// If concurrency <= 0, defaults to runtime.GOMAXPROCS(0).
@@ -2475,17 +2484,40 @@ func (t *Transaction) performCopyOnWriteDeletion(ctx
context.Context, operation
updater := t.updateSnapshot(wfs, snapshotProps,
operation).mergeOverwrite(&commitUUID, filter)
updater.setManifestConcurrency(concurrency)
- filesToDelete, filesToRewrite, fileSeqByPath, err :=
t.classifyFilesForDeletions(ctx, fs, filter, caseSensitive, concurrency)
+ filesToDelete, entriesToRewrite, err :=
t.classifyFilesForDeletions(ctx, fs, filter, caseSensitive, concurrency)
+ if err != nil {
+ return nil, nil, err
+ }
+
+ // SetProperties above may have staged a name mapping, which replaces
t.meta, so re-read it.
+ meta, err = t.txnMeta()
if err != nil {
return nil, nil, err
}
+ // Rewrites need the live deletes to drop rows already deleted, and
every removed file must take its deletion vector with it.
+ // DVs only exist from v3, so a v2 delete that only drops whole files
skips the read.
+ var (
+ deletes planningDeletes
+ builtMeta Metadata
+ )
+ if len(entriesToRewrite) > 0 || (len(filesToDelete) > 0 &&
meta.formatVersion >= 3) {
+ builtMeta, err = meta.Build()
+ if err != nil {
+ return nil, nil, fmt.Errorf("failed to build metadata:
%w", err)
+ }
+ deletes, err = t.readPlanningDeletes(fs, builtMeta)
+ if err != nil {
+ return nil, nil, err
+ }
+ }
+
for _, df := range filesToDelete {
- updater.deleteDataFile(df)
+ deletes.removeDataFile(updater, df)
}
- if len(filesToRewrite) > 0 {
- if err := t.rewriteFilesWithFilter(ctx, fs, updater,
filesToRewrite, fileSeqByPath, filter, caseSensitive, concurrency); err != nil {
+ if len(entriesToRewrite) > 0 {
+ if err := t.rewriteFilesWithFilter(ctx, fs, updater, builtMeta,
deletes, entriesToRewrite, filter, caseSensitive, concurrency); err != nil {
return nil, nil, err
}
}
@@ -2500,7 +2532,10 @@ func (t *Transaction) performCopyOnWriteDeletion(ctx
context.Context, operation
// shared with RewriteDataFiles via rewriteValidator, so changes there
// apply here too. A concurrent removal of one of these files is caught
// earlier, by the retry rebuild's checkRemovedFiles
(ErrCommitDiverged).
- removed := slices.Concat(filesToDelete, filesToRewrite)
+ removed := slices.Clip(filesToDelete)
+ for _, entry := range entriesToRewrite {
+ removed = append(removed, entry.DataFile())
+ }
if len(removed) > 0 {
t.addValidator(rewriteValidator(removed))
}
@@ -2538,17 +2573,41 @@ func (t *Transaction) performMergeOnReadDeletion(ctx
context.Context, snapshotPr
updater := t.updateSnapshot(wfs, snapshotProps,
OpDelete).mergeOverwrite(&commitUUID, filter)
updater.setManifestConcurrency(concurrency)
- filesToDelete, withPartialDeletions, _, err :=
t.classifyFilesForDeletions(ctx, fs, filter, caseSensitive, concurrency)
+ filesToDelete, withPartialDeletions, err :=
t.classifyFilesForDeletions(ctx, fs, filter, caseSensitive, concurrency)
+ if err != nil {
+ return nil, err
+ }
+
+ // SetProperties above may have staged a name mapping, which replaces
t.meta, so re-read it.
+ meta, err = t.txnMeta()
if err != nil {
return nil, err
}
+ // On v3 every removed file takes its deletion vector with it, and a
partially deleted file's vector is
+ // merged into its replacement. One read serves both. DVs only exist
from v3, so v2 skips it.
+ var deletes planningDeletes
+ if meta.formatVersion >= 3 && (len(filesToDelete) > 0 ||
len(withPartialDeletions) > 0) {
+ builtMeta, err := meta.Build()
+ if err != nil {
+ return nil, fmt.Errorf("failed to build metadata: %w",
err)
+ }
+ deletes, err = t.readPlanningDeletes(fs, builtMeta)
+ if err != nil {
+ return nil, err
+ }
+ }
+
for _, df := range filesToDelete {
- updater.deleteDataFile(df)
+ deletes.removeDataFile(updater, df)
}
if len(withPartialDeletions) > 0 {
- referenced, err := t.writePositionDeletesForFiles(ctx, fs,
updater, withPartialDeletions, filter, caseSensitive, concurrency, commitUUID)
+ partialFiles := make([]iceberg.DataFile, 0,
len(withPartialDeletions))
+ for _, entry := range withPartialDeletions {
+ partialFiles = append(partialFiles, entry.DataFile())
+ }
+ referenced, err := t.writePositionDeletesForFiles(ctx, fs,
updater, deletes.dvs, partialFiles, filter, caseSensitive, concurrency,
commitUUID)
if err != nil {
return nil, err
}
@@ -2614,6 +2673,12 @@ func WithDeleteCaseInsensitive() DeleteOption {
// - Strict evaluator determines if all rows in a file must match the filter
// - Files that pass inclusive but not strict evaluation are rewritten with
filtered data
//
+// On a v3 table, every data file the delete removes or rewrites takes its
deletion vector with it.
+// A merge-on-read delete replaces the existing deletion vector of a file it
deletes rows from.
+//
+// A commit that removes a deletion vector is not replayed: if another writer
commits first,
+// it fails with ErrCommitFailed instead of refreshing and retrying, so reload
the table and run the delete again.
+//
// The concurrency parameter controls the level of parallelism for manifest
processing and file rewriting and
// can be overridden using the WithOverwriteConcurrency option. Defaults to
runtime.GOMAXPROCS(0).
func (t *Transaction) Delete(ctx context.Context, filter
iceberg.BooleanExpression, snapshotProps iceberg.Properties, opts
...DeleteOption) (err error) {
@@ -2660,31 +2725,30 @@ func (t *Transaction) Delete(ctx context.Context,
filter iceberg.BooleanExpressi
}
// classifyFilesForDeletions classifies existing data files based on the
provided filter.
-// Returns files to delete completely, files to rewrite partially, the
-// per-file file_sequence_number map for rewrite candidates (used to
-// synthesize row-lineage columns when reading), and any error.
-func (t *Transaction) classifyFilesForDeletions(ctx context.Context, fs io.IO,
filter iceberg.BooleanExpression, caseSensitive bool, concurrency int)
(filesToDelete, filesWithPartialDeletions []iceberg.DataFile, fileSeqByPath
map[string]*int64, err error) {
+// Returns files to delete completely, the manifest entries of files that
partially match, and any error.
+// Partial matches stay entries because their sequence numbers decide which
deletes apply and seed row-lineage synthesis.
+func (t *Transaction) classifyFilesForDeletions(ctx context.Context, fs io.IO,
filter iceberg.BooleanExpression, caseSensitive bool, concurrency int)
(filesToDelete []iceberg.DataFile, filesWithPartialDeletions
[]iceberg.ManifestEntry, err error) {
meta, err := t.txnMeta()
if err != nil {
- return nil, nil, nil, err
+ return nil, nil, err
}
s := t.planningSnapshot(meta)
if s == nil {
- return nil, nil, nil, nil
+ return nil, nil, nil
}
if filter == nil || filter.Equals(iceberg.AlwaysTrue{}) {
for df, err := range s.dataFiles(fs, nil) {
if err != nil {
- return nil, nil, nil, err
+ return nil, nil, err
}
if df.ContentType() == iceberg.EntryContentData {
filesToDelete = append(filesToDelete, df)
}
}
- return filesToDelete, filesWithPartialDeletions, nil, nil
+ return filesToDelete, nil, nil
}
return t.classifyFilesForFilteredDeletions(ctx, fs, filter,
caseSensitive, concurrency)
@@ -2717,23 +2781,22 @@ func (t *fileClassificationTask)
buildPartitionProjection(specID int) (iceberg.B
}
// classifyFilesForFilteredDeletions classifies files for filtered overwrite
operations.
-// Returns files to delete completely, files to rewrite partially, the
-// per-file file_sequence_number map for rewrite candidates, and any error.
-func (t *Transaction) classifyFilesForFilteredDeletions(ctx context.Context,
fs io.IO, filter iceberg.BooleanExpression, caseSensitive bool, concurrency
int) (filesToDelete, filesWithPartialDeletes []iceberg.DataFile, fileSeqByPath
map[string]*int64, err error) {
+// Returns files to delete completely, the manifest entries of files that
partially match, and any error.
+func (t *Transaction) classifyFilesForFilteredDeletions(ctx context.Context,
fs io.IO, filter iceberg.BooleanExpression, caseSensitive bool, concurrency
int) (filesToDelete []iceberg.DataFile, filesWithPartialDeletes
[]iceberg.ManifestEntry, err error) {
meta, err := t.txnMeta()
if err != nil {
- return nil, nil, nil, err
+ return nil, nil, err
}
schema := meta.CurrentSchema()
builtMeta, err := meta.Build()
if err != nil {
- return nil, nil, nil, err
+ return nil, nil, err
}
inclusiveEvaluator, err := newInclusiveMetricsEvaluator(schema, filter,
caseSensitive, false)
if err != nil {
- return nil, nil, nil, fmt.Errorf("failed to create inclusive
metrics evaluator: %w", err)
+ return nil, nil, fmt.Errorf("failed to create inclusive metrics
evaluator: %w", err)
}
// A file is deleted as a whole only if the filter is true for all of
its
@@ -2742,12 +2805,12 @@ func (t *Transaction)
classifyFilesForFilteredDeletions(ctx context.Context, fs
// negation of the rows the rewrite keeps instead, so that both paths
agree.
notTrue, err := isNotTrueExpr(filter)
if err != nil {
- return nil, nil, nil, fmt.Errorf("failed to build negated
filter: %w", err)
+ return nil, nil, fmt.Errorf("failed to build negated filter:
%w", err)
}
strictEvaluator, err := newStrictMetricsEvaluator(schema,
iceberg.NewNot(notTrue), caseSensitive, false)
if err != nil {
- return nil, nil, nil, fmt.Errorf("failed to create strict
metrics evaluator: %w", err)
+ return nil, nil, fmt.Errorf("failed to create strict metrics
evaluator: %w", err)
}
classificationTask := newFileClassificationTask(builtMeta, filter,
caseSensitive)
@@ -2758,12 +2821,11 @@ func (t *Transaction)
classifyFilesForFilteredDeletions(ctx context.Context, fs
if s != nil {
manifests, err = s.Manifests(fs)
if err != nil {
- return nil, nil, nil, fmt.Errorf("failed to get
manifests: %w", err)
+ return nil, nil, fmt.Errorf("failed to get manifests:
%w", err)
}
}
var mu sync.Mutex
- fileSeqByPath = make(map[string]*int64)
g, _ := errgroup.WithContext(ctx)
g.SetLimit(min(concurrency, len(manifests)))
@@ -2783,8 +2845,7 @@ func (t *Transaction)
classifyFilesForFilteredDeletions(ctx context.Context, fs
}
localDelete := make([]iceberg.DataFile, 0)
- localRewrite := make([]iceberg.DataFile, 0)
- localSeqByPath := make(map[string]*int64)
+ localRewrite := make([]iceberg.ManifestEntry, 0)
for entry, err := range manifest.Entries(fs, false) {
if err != nil {
@@ -2816,20 +2877,7 @@ func (t *Transaction)
classifyFilesForFilteredDeletions(ctx context.Context, fs
if strict {
localDelete = append(localDelete, df)
} else {
- localRewrite = append(localRewrite, df)
- // Capture the entry's data sequence
number so the
- // rewrite path can synthesize
- // _last_updated_sequence_number for
source rows that
- // have a null value (or no column) in
the file. Per
- // spec the synthesized value is the
manifest entry's
- // sequence_number (field id 3, the
data sequence
- // number) — not file_sequence_number
(field id 4) —
- // so back-dated EXISTING entries
assign the correct
- // value to surviving rows.
- if seq := entry.SequenceNum(); seq >= 0
{
- s := seq
- localSeqByPath[df.FilePath()] =
&s
- }
+ localRewrite = append(localRewrite,
entry)
}
}
@@ -2837,7 +2885,6 @@ func (t *Transaction)
classifyFilesForFilteredDeletions(ctx context.Context, fs
mu.Lock()
filesToDelete = append(filesToDelete,
localDelete...)
filesWithPartialDeletes =
append(filesWithPartialDeletes, localRewrite...)
- maps.Copy(fileSeqByPath, localSeqByPath)
mu.Unlock()
}
@@ -2846,17 +2893,78 @@ func (t *Transaction)
classifyFilesForFilteredDeletions(ctx context.Context, fs
}
if err := g.Wait(); err != nil {
- return nil, nil, nil, err
+ return nil, nil, err
+ }
+
+ return filesToDelete, filesWithPartialDeletes, nil
+}
+
+// planningDeletes indexes the live delete files of the planning snapshot the
same way scan planning indexes them.
+type planningDeletes struct {
+ positional *positionalDeleteIndex
+ dvs map[string]iceberg.ManifestEntry
+ equality *equalityDeleteIndex
+}
+
+// readPlanningDeletes reads and indexes every live delete file of the
planning snapshot.
+//
+// Delete manifests are not pruned by the row filter: with filter id=2 the
rewrite keeps id=1,
+// so a delete on id=1 must still be applied.
+//
+// Known limitation (#2160): every live delete manifest of the snapshot is
read on each call,
+// with no partition pruning, so a narrow delete or rewrite on a delete-heavy
table pays for all of them.
+func (t *Transaction) readPlanningDeletes(fs io.IO, builtMeta Metadata)
(planningDeletes, error) {
+ meta, err := t.txnMeta()
+ if err != nil {
+ return planningDeletes{}, err
+ }
+
+ var liveDeletes []iceberg.ManifestEntry
+ if s := t.planningSnapshot(meta); s != nil {
+ // Skip DELETED entries: a replaced DV stays behind as one, and
merging it
+ // into a new DV would bring back deleted rows. See #1372.
+ for entry, err := range s.entries(fs,
iceberg.ManifestContentDeletes, true) {
+ if err != nil {
+ return planningDeletes{}, fmt.Errorf("failed to
read delete manifests: %w", err)
+ }
+ liveDeletes = append(liveDeletes, entry)
+ }
+ }
+
+ classified, err := classifyManifestEntries(liveDeletes)
+ if err != nil {
+ return planningDeletes{}, err
+ }
+ positional, err :=
buildPositionalDeleteIndex(classified.positionalDeleteEntries)
+ if err != nil {
+ return planningDeletes{}, err
+ }
+ dvs, err := buildDVIndex(classified.dvEntries)
+ if err != nil {
+ return planningDeletes{}, err
+ }
+ equality, err :=
buildEqualityDeleteIndex(classified.equalityDeleteEntries, builtMeta,
meta.CurrentSchema())
+ if err != nil {
+ return planningDeletes{}, err
}
- return filesToDelete, filesWithPartialDeletes, fileSeqByPath, nil
+ return planningDeletes{positional: positional, dvs: dvs, equality:
equality}, nil
+}
+
+// removeDataFile removes df and, as the spec requires, the deletion vector
that references it.
+// Removing a deletion vector makes the commit non-replayable (see
snapshotProducer.commitManifests).
+// Position and equality delete files are left alone, since they can still
apply to other files.
+func (d planningDeletes) removeDataFile(updater *snapshotProducer, df
iceberg.DataFile) {
+ updater.deleteDataFile(df)
+ if dv, ok := d.dvs[df.FilePath()]; ok {
+ updater.removeDeletionVector(dv.DataFile())
+ }
}
// rewriteFilesWithFilter rewrites data files by preserving only rows that do
NOT match the filter.
-// fileSeqByPath maps each file's path to its file_sequence_number from the
source manifest entry,
-// used to synthesize _last_updated_sequence_number when the source row's
value is null. The
-// filter binding and substrait conversion are computed once here and reused
across every file.
-func (t *Transaction) rewriteFilesWithFilter(ctx context.Context, fs io.IO,
updater *snapshotProducer, files []iceberg.DataFile, fileSeqByPath
map[string]*int64, filter iceberg.BooleanExpression, caseSensitive bool,
concurrency int) error {
+// Rows already removed by a delete are dropped too, since those deletes stop
applying once the file is replaced.
+// The filter binding and substrait conversion are computed once here and
reused across every file.
+func (t *Transaction) rewriteFilesWithFilter(ctx context.Context, fs io.IO,
updater *snapshotProducer, builtMeta Metadata, deletes planningDeletes, entries
[]iceberg.ManifestEntry, filter iceberg.BooleanExpression, caseSensitive bool,
concurrency int) error {
meta, err := t.txnMeta()
if err != nil {
return err
@@ -2876,13 +2984,18 @@ func (t *Transaction) rewriteFilesWithFilter(ctx
context.Context, fs io.IO, upda
return err
}
- for _, originalFile := range files {
+ for _, entry := range entries {
+ task, err := fileScanTaskForDataEntry(entry,
deletes.positional, deletes.dvs, deletes.equality)
+ if err != nil {
+ return err
+ }
+
// Use a separate UUID for rewrite operations to avoid filename
collisions with new data files
rewriteUUID := uuid.New()
args := rewriteSingleFileArgs{
fs: fs,
- originalFile: originalFile,
- fileSeqNum: fileSeqByPath[originalFile.FilePath()],
+ builtMeta: builtMeta,
+ task: task,
filter: complementFilter,
postFilter: postFilter,
caseSensitive: caseSensitive,
@@ -2891,10 +3004,10 @@ func (t *Transaction) rewriteFilesWithFilter(ctx
context.Context, fs io.IO, upda
}
rewrittenFiles, err := t.rewriteSingleFile(ctx, args)
if err != nil {
- return fmt.Errorf("failed to rewrite file %s: %w",
originalFile.FilePath(), err)
+ return fmt.Errorf("failed to rewrite file %s: %w",
task.File.FilePath(), err)
}
- updater.deleteDataFile(originalFile)
+ deletes.removeDataFile(updater, task.File)
for _, rewrittenFile := range rewrittenFiles {
updater.appendDataFile(rewrittenFile)
}
@@ -2992,12 +3105,13 @@ func (isNotTrueVisitor) VisitBound(pred
iceberg.BoundPredicate) truthExprs {
// — there are several same-typed fields (two filter values, plus a UUID and an
// int) so positional ordering offers no compile-time order protection.
type rewriteSingleFileArgs struct {
- fs io.IO
- originalFile iceberg.DataFile
- // fileSeqNum is the source file's data sequence number from its
- // manifest entry; required to synthesize _last_updated_sequence_number
- // for rows whose value is null in the source file.
- fileSeqNum *int64
+ fs io.IO
+ // builtMeta is the transaction's metadata, built once for the whole
rewrite.
+ builtMeta Metadata
+ // task is the source file as planned by [fileScanTaskForDataEntry].
+ // Its deletes must be applied, and its sequence number seeds
_last_updated_sequence_number
+ // for rows whose value is null in the file.
+ task FileScanTask
// filter is the per-row complement (rows that survive the rewrite).
filter iceberg.BooleanExpression
// postFilter is the pre-bound, pre-substrait-converted version of
@@ -3019,12 +3133,6 @@ func (t *Transaction) rewriteSingleFile(ctx
context.Context, args rewriteSingleF
return nil, err
}
- scanTask := &FileScanTask{
- File: args.originalFile,
- Start: 0,
- Length: args.originalFile.FileSizeBytes(),
- }
-
// Preserve row lineage for v3 tables: include _row_id and
// _last_updated_sequence_number in the scan projection so they are read
// from the source file (or synthesized from file metadata when null)
and
@@ -3042,7 +3150,7 @@ func (t *Transaction) rewriteSingleFile(ctx
context.Context, args rewriteSingleF
// scanner's row filter stays AlwaysTrue. Pruning happens before rows
are
// materialized, so _row_id synthesis still sees the original positions
of
// every surviving row group and the post-synthesis filter remains
correct.
- _, originalFirstRowID, _, _, _ :=
iceberginternal.BorrowedDataFilePointers(args.originalFile)
+ _, originalFirstRowID, _, _, _ :=
iceberginternal.BorrowedDataFilePointers(args.task.File)
preserveRowLineage := meta.formatVersion >= 3 && originalFirstRowID !=
nil
projectedSchema := meta.CurrentSchema()
var factoryOpts []writerFactoryOption
@@ -3055,13 +3163,6 @@ func (t *Transaction) rewriteSingleFile(ctx
context.Context, args rewriteSingleF
projectedSchema = iceberg.SchemaWithRowLineage(projectedSchema)
factoryOpts = append(factoryOpts,
withFactoryFileSchema(projectedSchema))
- scanTask.FirstRowID = args.originalFile.FirstRowID()
- // fileSeqNum drives _last_updated_sequence_number synthesis for
- // source rows whose value is null in the file (or for source
- // files written before row lineage existed). When the column
- // is non-null the existing per-row value wins, preserving the
- // original update sequence across the rewrite.
- scanTask.DataSequenceNumber = args.fileSeqNum
}
// When preserving row lineage, the bound scan-time filter is replaced
@@ -3076,13 +3177,8 @@ func (t *Transaction) rewriteSingleFile(ctx
context.Context, args rewriteSingleF
scanFilter = boundFilter
}
- builtMeta, err := meta.Build()
- if err != nil {
- return nil, fmt.Errorf("failed to build metadata: %w", err)
- }
-
scanner := &arrowScan{
- metadata: builtMeta,
+ metadata: args.builtMeta,
fs: args.fs,
scanSchema: meta.CurrentSchema(),
projectedSchema: projectedSchema,
@@ -3093,7 +3189,7 @@ func (t *Transaction) rewriteSingleFile(ctx
context.Context, args rewriteSingleF
concurrency: args.concurrency,
}
- arrowSchema, recordIter, err := scanner.GetRecords(ctx,
[]FileScanTask{*scanTask})
+ arrowSchema, recordIter, err := scanner.GetRecords(ctx,
[]FileScanTask{args.task})
if err != nil {
return nil, fmt.Errorf("failed to get records from original
file: %w", err)
}
@@ -3104,7 +3200,7 @@ func (t *Transaction) rewriteSingleFile(ctx
context.Context, args rewriteSingleF
if preserveRowLineage {
arrowSchema, err =
SchemaToArrowSchemaWithOptions(projectedSchema, ArrowSchemaOptions{
IncludeFieldIDs: true,
- TableProperties: builtMeta.Properties(),
+ TableProperties: args.builtMeta.Properties(),
})
if err != nil {
return nil, fmt.Errorf("failed to build arrow schema
with field IDs: %w", err)
@@ -3174,7 +3270,7 @@ func (t *Transaction) rewriteSingleFile(ctx
context.Context, args rewriteSingleF
// 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.
-func (t *Transaction) writePositionDeletesForFiles(ctx context.Context, fs
io.IO, updater *snapshotProducer, files []iceberg.DataFile, filter
iceberg.BooleanExpression, caseSensitive bool, concurrency int, commitUUID
uuid.UUID) ([]string, error) {
+func (t *Transaction) writePositionDeletesForFiles(ctx context.Context, fs
io.IO, updater *snapshotProducer, dvs map[string]iceberg.ManifestEntry, files
[]iceberg.DataFile, filter iceberg.BooleanExpression, caseSensitive bool,
concurrency int, commitUUID uuid.UUID) ([]string, error) {
meta, err := t.txnMeta()
if err != nil {
return nil, err
@@ -3200,24 +3296,20 @@ func (t *Transaction) writePositionDeletesForFiles(ctx
context.Context, fs io.IO
// into the new DV (the incremental deletes) and the superseded DVs are
// removed. Without this a second delete on the same file would write a
// second live DV, corrupting the table (buildDVIndex rejects it).
- var (
- existingDVs map[string]iceberg.DataFile
- existingDVBitmaps map[string]*dv.RoaringPositionBitmap
- )
- if meta.formatVersion >= 3 {
- existingDVs, err = t.collectExistingDVs(fs, files)
- if err != nil {
- return nil, err
+ // dvs holds the live DVs keyed by data file path, and is empty on v2.
+ existingDVs := make(map[string]iceberg.DataFile)
+ existingDVBitmaps := make(map[string]*dv.RoaringPositionBitmap)
+ for _, df := range files {
+ entry, ok := dvs[df.FilePath()]
+ if !ok {
+ continue
}
-
- existingDVBitmaps = make(map[string]*dv.RoaringPositionBitmap,
len(existingDVs))
- for ref, dvFile := range existingDVs {
- bitmap, err := dv.ReadDV(fs, dvFile)
- if err != nil {
- return nil, fmt.Errorf("failed to read existing
deletion vector for %s: %w", ref, err)
- }
- existingDVBitmaps[ref] = bitmap
+ bitmap, err := dv.ReadDV(fs, entry.DataFile())
+ if err != nil {
+ return nil, fmt.Errorf("failed to read existing
deletion vector for %s: %w", df.FilePath(), err)
}
+ existingDVs[df.FilePath()] = entry.DataFile()
+ existingDVBitmaps[df.FilePath()] = bitmap
}
posDeleteFiles := positionDeleteRecordsToDataFiles(ctx,
t.tbl.Location(), meta, partitionContextByFilePath, recordWritingArgs{
@@ -3252,49 +3344,6 @@ func (t *Transaction) writePositionDeletesForFiles(ctx
context.Context, fs io.IO
return referenced, nil
}
-// collectExistingDVs returns the live deletion vectors that reference any of
-// files, keyed by the referenced data file path. Used by the v3 merge-on-read
-// delete path to fold prior deletes into a new DV and remove the superseded
-// entries, preserving the spec's one-DV-per-data-file invariant.
-func (t *Transaction) collectExistingDVs(fs io.IO, files []iceberg.DataFile)
(map[string]iceberg.DataFile, error) {
- meta, err := t.txnMeta()
- if err != nil {
- return nil, err
- }
-
- s := t.planningSnapshot(meta)
- if s == nil {
- return nil, nil
- }
-
- wanted := make(map[string]struct{}, len(files))
- for _, df := range files {
- wanted[df.FilePath()] = struct{}{}
- }
-
- result := make(map[string]iceberg.DataFile)
- // Skip DELETED entries: an old, replaced DV stays behind as one.
Reading it
- // would build the new DV from old data and bring back deleted rows.
See #1372.
- for entry, err := range s.entries(fs, iceberg.ManifestContentDeletes,
true) {
- if err != nil {
- return nil, fmt.Errorf("scanning existing deletion
vectors: %w", err)
- }
- df := entry.DataFile()
- if !IsDeletionVector(df) {
- continue
- }
- ref := iceberginternal.BorrowedDataFileReferencedDataFile(df)
- if ref == nil {
- continue
- }
- if _, ok := wanted[*ref]; ok {
- result[*ref] = df
- }
- }
-
- return result, nil
-}
-
func (t *Transaction) makePositionDeleteRecordsForFilter(ctx context.Context,
fs io.IO, files []iceberg.DataFile, filter iceberg.BooleanExpression,
caseSensitive bool, concurrency int) (seq2 iter.Seq2[arrow.RecordBatch, error],
err error) {
meta, err := t.txnMeta()
if err != nil {
