badalprasadsingh commented on code in PR #2434:
URL: https://github.com/apache/iceberg-go/pull/2434#discussion_r4503430888
##########
table/transaction_delete.go:
##########
@@ -200,8 +200,36 @@ func (t *Transaction) performMergeOnReadDeletion(ctx
context.Context, snapshotPr
return nil, err
}
if len(referenced) > 0 {
+ // A first v3 deletion vector is otherwise replayable.
If a
+ // concurrent writer adds a DV for the same file,
replaying ours
+ // would leave two live vectors and make the table
unreadable.
+ dvTargets := make(map[string]struct{}, len(referenced))
+ if meta.formatVersion >= 3 {
+ for _, ref := range referenced {
+ dvTargets[ref] = struct{}{}
+ }
+ }
t.addValidator(func(cc *conflictContext) error {
- return validateDataFilesExist(cc, referenced)
+ if err := validateDataFilesExist(cc,
referenced); err != nil {
+ return err
+ }
+ if len(dvTargets) == 0 {
+ return nil
+ }
+
+ return
cc.forEachAddedEntry(iceberg.ManifestContentDeletes, func(snap Snapshot, entry
iceberg.ManifestEntry) error {
+ file := entry.DataFile()
+ ref := explicitReferencedDataFile(file)
+ if !IsDeletionVector(file) || ref == ""
{
+ return nil
+ }
+ if _, ok := dvTargets[ref]; ok {
+ return fmt.Errorf("cannot add a
deletion vector for data file %s: concurrent snapshot %d committed a live
deletion vector %s for it; reload the table and retry the delete",
+ ref, snap.SnapshotID,
file.FilePath())
Review Comment:
please wrap `ErrConflictingDeleteFiles` like the other conflict checks
##########
table/mor_delete_conflict_test.go:
##########
@@ -117,6 +117,27 @@ func
TestMergeOnReadDeleteConflict_ReferencedDataFileRemoved(t *testing.T) {
require.ErrorIs(t, err, table.ErrDataFilesMissing)
}
+// A first deletion vector is replayable until another writer commits a
+// vector for the same data file. The second writer must reject replay instead
+// of leaving two live vectors, which makes all subsequent scans fail.
+func TestMergeOnReadDeleteConflict_ConcurrentFirstDeletionVector(t *testing.T)
{
+ ctx := t.Context()
+ tbl := appendTenRows(t, newV3MoRConflictTestTable(t))
+
+ pending := tbl.NewTransaction()
+ require.NoError(t, pending.Delete(ctx,
iceberg.EqualTo(iceberg.Reference("id"), int64(7)), nil))
+
+ // Both writers plan against the same file with no existing DV.
+ peer, err := tbl.Delete(ctx, iceberg.EqualTo(iceberg.Reference("id"),
int64(2)), nil)
Review Comment:
here: peer is another delete, no test covers the case this fix came from:
the conversion commits its dv first, in a replace snapshot.
please add that case, so a later filter by snapshot operation can't skip it
##########
table/equality_delete_dv_conversion_test.go:
##########
@@ -0,0 +1,557 @@
+// 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 (
+ "fmt"
+ "os"
+ "path/filepath"
+ "strings"
+ "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/require"
+)
+
+func TestConvertEqualityDeletesToDeletionVectors(t *testing.T) {
+ ctx := t.Context()
+ tbl := appendRows(t, newMergeOnReadTestTableVersion(t, "3"),
+
`[{"id":1,"data":"a"},{"id":2,"data":"b"},{"id":3,"data":"c"},{"id":4,"data":"d"},{"id":5,"data":"e"}]`)
+ beforeDelete := tbl.CurrentSnapshot().SnapshotID
+ tbl = appendEqualityDelete(t, tbl, []int{1}, `[{"id":2},{"id":4}]`)
+ deleteSnapshot := tbl.CurrentSnapshot().SnapshotID
+
+ result, err := tbl.ConvertEqualityDeletesToDeletionVectors(ctx)
+ require.NoError(t, err)
+ require.NotNil(t, result.Table)
+ require.Equal(t, 1, result.ConvertedEqualityDeleteFiles)
+ require.Equal(t, 1, result.AddedDeletionVectors)
+ require.Equal(t, []int64{1, 3, 5}, idsInTable(t, result.Table))
+ require.Equal(t, 1, liveDVCount(t, result.Table))
+ require.Equal(t, 0, liveDeleteFileCount(t, result.Table,
iceberg.EntryContentEqDeletes))
+ require.Equal(t, table.OpReplace,
result.Table.CurrentSnapshot().Summary.Operation)
+
+ // The equality-delete snapshot emits the deletes once. The conversion
+ // snapshot is maintenance and must not emit them again.
+ deleteRows := readChangelog(t, ctx,
+
tbl.NewIncrementalChangelogScan().FromSnapshotExclusive(beforeDelete).ToSnapshot(deleteSnapshot))
+ require.Equal(t, []int64{2, 4}, changelogIDs(deleteRows,
table.ChangelogOpDelete))
+ conversionRows := readChangelog(t, ctx,
+
result.Table.NewIncrementalChangelogScan().FromSnapshotExclusive(deleteSnapshot))
+ require.Empty(t, conversionRows)
+}
+
+// Equality deletes may use different key sets. Each must resolve against the
+// same data file before their positions are merged into a single DV.
+func TestConvertEqualityDeletesWithDifferentEqualityFields(t *testing.T) {
+ ctx := t.Context()
+ tbl := appendRows(t, newMergeOnReadTestTableVersion(t, "3"),
+
`[{"id":1,"data":"a"},{"id":2,"data":"b"},{"id":3,"data":"c"},{"id":4,"data":"d"}]`)
+ tbl = appendEqualityDelete(t, tbl, []int{1}, `[{"id":2}]`)
+ tbl = appendEqualityDelete(t, tbl, []int{2}, `[{"data":"c"}]`)
+
+ require.Equal(t, []int64{1, 4}, idsInTable(t, tbl))
+ result, err := tbl.ConvertEqualityDeletesToDeletionVectors(ctx)
+ require.NoError(t, err)
+ require.Equal(t, 2, result.ConvertedEqualityDeleteFiles)
+ require.Equal(t, 1, result.AddedDeletionVectors)
+ require.Equal(t, 1, liveDVCount(t, result.Table))
+ require.Zero(t, liveDeleteFileCount(t, result.Table,
iceberg.EntryContentEqDeletes))
+ require.Equal(t, []int64{1, 4}, idsInTable(t, result.Table))
+}
+
+// A conversion staged after write.data.path changes must use the transaction's
+// updated output location, not the base table's data directory.
+func TestConvertEqualityDeletesUsesStagedDataPath(t *testing.T) {
+ ctx := t.Context()
+ tbl := appendRows(t, newMergeOnReadTestTableVersion(t, "3"),
+ `[{"id":1,"data":"a"},{"id":2,"data":"b"}]`)
+ tbl = appendEqualityDelete(t, tbl, []int{1}, `[{"id":2}]`)
+
+ dataPath := filepath.ToSlash(filepath.Join(t.TempDir(),
"custom-dv-data"))
+ tx := tbl.NewTransaction()
+ require.NoError(t,
tx.SetProperties(iceberg.Properties{table.WriteDataPathKey: dataPath}))
+
+ result, err := tx.ConvertEqualityDeletesToDeletionVectors(ctx)
+ require.NoError(t, err)
+ require.Equal(t, 1, result.AddedDeletionVectors)
+ committed, err := tx.Commit(ctx)
+ require.NoError(t, err)
+ require.Equal(t, []int64{1}, idsInTable(t, committed))
+
+ fs, err := committed.FS(ctx)
+ require.NoError(t, err)
+ manifests, err := committed.CurrentSnapshot().Manifests(ctx, fs)
+ require.NoError(t, err)
+
+ var dvPaths []string
+ for _, manifest := range manifests {
+ if manifest.ManifestContent() != iceberg.ManifestContentDeletes
{
+ continue
+ }
+ for entry, err := range manifest.Entries(ctx, fs, false) {
+ require.NoError(t, err)
+ if entry.Status() != iceberg.EntryStatusDELETED &&
table.IsDeletionVector(entry.DataFile()) {
+ dvPaths = append(dvPaths,
entry.DataFile().FilePath())
+ }
+ }
+ }
+ require.Len(t, dvPaths, 1)
+ require.True(t, strings.HasPrefix(dvPaths[0], dataPath+"/"), dvPaths[0])
+}
+
+func TestConvertEqualityDeletesMergesExistingPositionDeletes(t *testing.T) {
+ ctx := t.Context()
+ tbl := appendRows(t, newMergeOnReadTestTableVersion(t, "2"),
+
`[{"id":1,"data":"a"},{"id":2,"data":"b"},{"id":3,"data":"c"},{"id":4,"data":"d"},{"id":5,"data":"e"}]`)
+ var err error
+ tbl, err = tbl.Delete(ctx, iceberg.EqualTo(iceberg.Reference("id"),
int64(2)), nil)
+ require.NoError(t, err)
+ require.Equal(t, 1, liveDeleteFileCount(t, tbl,
iceberg.EntryContentPosDeletes))
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.UpgradeFormatVersion(3))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+ tbl = appendEqualityDelete(t, tbl, []int{1}, `[{"id":4}]`)
+
+ result, err := tbl.ConvertEqualityDeletesToDeletionVectors(ctx)
+ require.NoError(t, err)
+ require.Equal(t, 1, result.RemovedPositionDeleteFiles)
+ require.Zero(t, liveDeleteFileCount(t, result.Table,
iceberg.EntryContentPosDeletes))
+ require.Equal(t, 1, liveDVCount(t, result.Table))
+ require.Equal(t, []int64{1, 3, 5}, idsInTable(t, result.Table),
+ "the old position delete and the new equality delete must both
remain applied")
+}
+
+// A legacy position-delete file can cover multiple data files. Replacing it
+// with a DV for only one of those files must not leave the shared file live.
+func TestConvertEqualityDeletesRejectsSharedPositionDeletes(t *testing.T) {
+ ctx := t.Context()
+ tbl := newMergeOnReadTestTableVersion(t, "2")
+ tbl = appendRows(t, tbl,
+ `[{"id":1,"data":"a"},{"id":2,"data":"b"}]`)
+ tbl = appendRows(t, tbl,
+ `[{"id":3,"data":"c"},{"id":4,"data":"d"}]`)
+
+ scan := tbl.Scan()
+ defer scan.Close()
+ tasks, err := scan.PlanFiles(ctx)
+ require.NoError(t, err)
+ require.Len(t, tasks, 2)
+
+ positionPath := tbl.Location() + "/data/shared-position-delete.parquet"
+ writeParquetFile(t, positionPath, table.PositionalDeleteArrowSchema,
fmt.Sprintf(
+ `[{"file_path":%q,"pos":0},{"file_path":%q,"pos":0}]`,
+ tasks[0].File.FilePath(), tasks[1].File.FilePath()))
+ info, err := os.Stat(positionPath)
+ require.NoError(t, err)
+ builder, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec,
+ iceberg.EntryContentPosDeletes, positionPath,
iceberg.ParquetFile, nil, nil, nil, 2, info.Size())
+ require.NoError(t, err)
+ // Deliberately omit ReferencedDataFile: both files share this delete
file.
+ tx := tbl.NewTransaction()
+ require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(builder.Build()).Commit(ctx))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.UpgradeFormatVersion(3))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+ tbl = appendEqualityDelete(t, tbl, []int{1}, `[{"id":2}]`)
+
+ // Enable retries so the final commit verifies replay after the peer
update,
+ // rather than failing immediately when main changes.
+ retryTx := tbl.NewTransaction()
+ require.NoError(t,
retryTx.SetProperties(iceberg.Properties{table.CommitNumRetriesKey: "2"}))
+ tbl, err = retryTx.Commit(ctx)
+ require.NoError(t, err)
+
+ beforeSnapshot := tbl.CurrentSnapshot().SnapshotID
+ require.Equal(t, []int64{4}, idsInTable(t, tbl))
+
+ _, err = tbl.ConvertEqualityDeletesToDeletionVectors(ctx)
+ require.ErrorIs(t, err, table.ErrInvalidOperation)
Review Comment:
no test checks that the new Puffin file is removed after this rejection
if the rejection stays: could you walk the table location here and assert no
`.puffin` file remains?
##########
table/transaction.go:
##########
@@ -2154,6 +2205,9 @@ func (t *Transaction) replaceFiles(ctx context.Context,
dataFilesToDelete, dataF
}
}
}
+ // Every new DV must replace all applicable legacy position-delete
files.
+ // A conversion that leaves an unscoped position-delete file behind is
+ // unsafe, even if its positions were already merged into the new
vector.
Review Comment:
I think its better to say "not supported" than "unsafe" here
##########
table/scanner.go:
##########
@@ -2218,12 +2218,53 @@ func (scan *Scan) ReadTasks(ctx context.Context, tasks
[]FileScanTask) (*arrow.S
// single data file in tasks: rows that survive that task's own deletes are
// kept only when they match addedDeletes. Normal scans pass nil.
func (scan *Scan) readFileTasks(ctx context.Context, tasks []FileScanTask,
addedDeletes *FileScanTask) (*arrow.Schema, iter.Seq2[arrow.RecordBatch,
error], error) {
+ var added []FileScanTask
+ if addedDeletes != nil {
+ added = []FileScanTask{*addedDeletes}
+ }
+
+ return scan.readFileTasksWithAddedDeletes(ctx, tasks, added, nil)
+}
+
+// readFileTasksWithAddedDeletes reads tasks while retaining rows matched by
the
+// corresponding added-delete task for each data file. positionSinks, when set,
+// receives the original file positions of those retained rows before they are
+// returned to the caller.
Review Comment:
nit: with a sink these rows are never returned, because keepAddedDeleteRows
clears them
better: say they are consumed here, not returned
##########
table/convert_equality_deletes_to_dv.go:
##########
@@ -0,0 +1,382 @@
+// 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"
+ "fmt"
+ "slices"
+ "sync"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/table/dv"
+ "github.com/google/uuid"
+)
+
+// equalityDeletePositionBuffer accumulates matched positions for one data
file.
+// Each buffer has its own mutex so parallel scans of different files do not
+// contend on a table-wide lock. The scan finishes before bitmap is read.
+type equalityDeletePositionBuffer struct {
+ mu sync.Mutex
+ bitmap *dv.RoaringPositionBitmap
+ invalidPosition int64
+}
+
+func (b *equalityDeletePositionBuffer) addPositions(positions []int64) {
+ if len(positions) == 0 {
+ return
+ }
+
+ b.mu.Lock()
+ defer b.mu.Unlock()
+
+ if b.invalidPosition < 0 {
+ return
+ }
+ for _, position := range positions {
+ if position < 0 {
+ b.invalidPosition = position
+
+ return
+ }
+ }
+
+ if b.bitmap == nil {
+ b.bitmap = dv.NewRoaringPositionBitmap()
+ }
+ for _, position := range positions {
+ b.bitmap.Set(uint64(position))
+ }
+}
+
+// EqualityDeleteToDVResult summarizes an equality-delete conversion.
+type EqualityDeleteToDVResult struct {
+ // Table is the table returned by the table-level action. It is nil
when the
+ // action is staged on a Transaction and set after that transaction
commits.
Review Comment:
nit: `txn.Commit` never sets this field on a staged result, so "set after
that transaction commits" misleads.
better: say it stays nil for transaction callers
##########
table/equality_delete_dv_conversion_test.go:
##########
@@ -0,0 +1,557 @@
+// 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
Review Comment:
nit: tests here usually match their source file, like
`remove_dangling_delete_files_test.go`
better: please name these after `convert_equality_deletes_to_dv.go`, using
`convert_equality_deletes_to_dv_internal_test.go` for the buffer test in
package table
##########
table/convert_equality_deletes_to_dv.go:
##########
@@ -0,0 +1,382 @@
+// 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"
+ "fmt"
+ "slices"
+ "sync"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/table/dv"
+ "github.com/google/uuid"
+)
+
+// equalityDeletePositionBuffer accumulates matched positions for one data
file.
+// Each buffer has its own mutex so parallel scans of different files do not
+// contend on a table-wide lock. The scan finishes before bitmap is read.
+type equalityDeletePositionBuffer struct {
+ mu sync.Mutex
+ bitmap *dv.RoaringPositionBitmap
+ invalidPosition int64
+}
+
+func (b *equalityDeletePositionBuffer) addPositions(positions []int64) {
+ if len(positions) == 0 {
+ return
+ }
+
+ b.mu.Lock()
+ defer b.mu.Unlock()
+
+ if b.invalidPosition < 0 {
+ return
+ }
+ for _, position := range positions {
+ if position < 0 {
+ b.invalidPosition = position
+
+ return
+ }
+ }
+
+ if b.bitmap == nil {
+ b.bitmap = dv.NewRoaringPositionBitmap()
+ }
+ for _, position := range positions {
+ b.bitmap.Set(uint64(position))
+ }
+}
+
+// EqualityDeleteToDVResult summarizes an equality-delete conversion.
+type EqualityDeleteToDVResult struct {
+ // Table is the table returned by the table-level action. It is nil
when the
+ // action is staged on a Transaction and set after that transaction
commits.
+ Table *Table
+
+ // ConvertedEqualityDeleteFiles is the number of live equality delete
files
+ // removed by the rewrite.
+ ConvertedEqualityDeleteFiles int
+ // AddedDeletionVectors is the number of file-specific deletion vectors
added.
+ AddedDeletionVectors int
+ // RemovedPositionDeleteFiles is the number of legacy position delete
files
+ // scoped to one data file, either by referenced_data_file or equal
+ // file_path bounds, removed with the conversion.
+ RemovedPositionDeleteFiles int
+ // ReplacedDeletionVectors is the number of existing vectors merged
into new
+ // vectors and removed by the rewrite.
+ ReplacedDeletionVectors int
+}
+
+// ConvertEqualityDeletesToDeletionVectors converts live equality deletes on
+// the main branch of a v3 table to file-specific deletion vectors. The rewrite
+// commits as one replace snapshot so changelog readers do not see a second
+// logical deletion event.
+func (t Table) ConvertEqualityDeletesToDeletionVectors(ctx context.Context)
(*EqualityDeleteToDVResult, error) {
+ txn, err := t.NewTransactionOnBranch(MainBranch)
+ if err != nil {
+ return nil, err
+ }
+
+ result, err := txn.ConvertEqualityDeletesToDeletionVectors(ctx)
+ if err != nil {
+ return result, err
+ }
+ if result.ConvertedEqualityDeleteFiles == 0 {
+ result.Table = &t
+
+ return result, nil
+ }
+
+ committed, err := txn.Commit(ctx)
+ if err == nil {
+ result.Table = committed
+ }
+
+ return result, err
+}
+
+// ConvertEqualityDeletesToDeletionVectors stages an equality-delete to
+// deletion-vector conversion on the transaction's branch. Commit the
+// transaction to publish the replace snapshot.
+func (t *Transaction) ConvertEqualityDeletesToDeletionVectors(ctx
context.Context) (*EqualityDeleteToDVResult, error) {
+ meta, err := t.txnMeta()
+ if err != nil {
+ return nil, err
+ }
+ if meta.formatVersion != 3 {
+ return nil, fmt.Errorf("%w: equality-delete to deletion-vector
conversion requires a v3 table, got v%d",
+ iceberg.ErrNotImplemented, meta.formatVersion)
+ }
+
+ snapshot := t.planningSnapshot(meta)
+ if snapshot == nil {
+ return &EqualityDeleteToDVResult{}, nil
+ }
+
+ fs, err := t.tbl.FS(ctx)
+ if err != nil {
+ return nil, err
+ }
+ manifests, err := snapshot.Manifests(ctx, fs)
+ if err != nil {
+ return nil, err
+ }
+ equalityDeleteFiles := make(map[string]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(ctx, fs, true) {
+ if err != nil {
+ return nil, err
+ }
+ file := entry.DataFile()
+ if file.ContentType() != iceberg.EntryContentEqDeletes {
+ continue
+ }
+ if _, exists := equalityDeleteFiles[file.FilePath()];
exists {
+ return nil, fmt.Errorf("%w: duplicate live
equality delete file %s",
+ ErrInvalidMetadata, file.FilePath())
+ }
+ equalityDeleteFiles[file.FilePath()] = file
+ }
+ }
+ if len(equalityDeleteFiles) == 0 {
+ return &EqualityDeleteToDVResult{}, nil
+ }
+ // Only projected equality keys are needed to resolve their matching row
+ // positions. Missing (dropped) keys are handled by the delete loader.
+ keyNames := make(map[string]struct{})
+ for _, file := range equalityDeleteFiles {
+ for _, id := range file.EqualityFieldIDs() {
+ if name, ok := meta.CurrentSchema().FindColumnName(id);
ok {
+ keyNames[name] = struct{}{}
+ }
+ }
+ }
+ selectedFields := make([]string, 0, len(keyNames))
+ for name := range keyNames {
+ selectedFields = append(selectedFields, name)
+ }
+ slices.Sort(selectedFields)
+ scanOpts := []ScanOption{WithScanPlanningMode(ScanPlanningLocal)}
+ // WithSelectedFields with no names selects every column. Keep that
+ // fallback when all equality keys were dropped from the current schema.
+ if len(selectedFields) > 0 {
+ scanOpts = append(scanOpts,
WithSelectedFields(selectedFields...))
+ }
+
+ scan, err := t.Scan(scanOpts...)
+ if err != nil {
+ return nil, err
+ }
+ if t.branch != "" && t.branch != MainBranch &&
scan.metadata.SnapshotByName(t.branch) != nil {
+ branchScan, err := scan.UseRef(t.branch)
+ _ = scan.Close()
+ if err != nil {
+ return nil, err
+ }
+ scan = branchScan
+ }
+ defer scan.Close()
+
+ plannedTasks, err := scan.PlanFiles(ctx)
+ if err != nil {
+ return nil, err
+ }
+ baseTasks := make([]FileScanTask, 0, len(plannedTasks))
+ addedDeleteTasks := make([]FileScanTask, 0, len(plannedTasks))
+ positionsByPath := make(map[string]*equalityDeletePositionBuffer)
+ positionSinks := make(map[string]func([]int64))
+ addedTaskByPath := make(map[string]int)
+ for _, task := range plannedTasks {
+ if len(task.EqualityDeleteFiles) == 0 {
+ continue
+ }
+ if task.File == nil {
+ return nil, fmt.Errorf("%w: scan task with equality
deletes has no data file", ErrInvalidMetadata)
+ }
+ path := task.File.FilePath()
+ if _, exists := positionsByPath[path]; !exists {
+ positions := &equalityDeletePositionBuffer{}
+ positionsByPath[path] = positions
+ positionSinks[path] = positions.addPositions
+ }
+
+ base := task
+ base.EqualityDeleteFiles = nil
+ baseTasks = append(baseTasks, base)
+
+ if _, exists := addedTaskByPath[path]; !exists {
+ added := task
+ added.DeleteFiles = nil
+ added.DeletionVectorFiles = nil
+ addedTaskByPath[path] = len(addedDeleteTasks)
+ addedDeleteTasks = append(addedDeleteTasks, added)
+ }
+ }
+
+ if len(baseTasks) > 0 {
+ _, records, err := scan.readFileTasksWithAddedDeletes(ctx,
baseTasks, addedDeleteTasks, positionSinks)
+ if err != nil {
+ return nil, err
+ }
+ for record, err := range records {
+ if err != nil {
+ return nil, err
+ }
+ record.Release()
+ }
+ }
+
+ // A row-position cursor should never produce negative positions. Unlike
+ // DVWriter.Add, DVWriter.Load accepts an unsigned bitmap, so reject
+ // invalid positions before they can wrap to huge uint64 values.
+ for path, positions := range positionsByPath {
+ if positions.invalidPosition < 0 {
+ return nil, fmt.Errorf("%w: negative equality-delete
row position %d for data file %s",
+ iceberg.ErrInvalidArgument,
positions.invalidPosition, path)
+ }
+ }
+
+ writeFS, err := requireWriteFileIO(fs)
+ if err != nil {
+ return nil, err
+ }
+ metadata, err := meta.Build()
+ if err != nil {
+ return nil, err
+ }
+ positionDeleteLoader := newLazyPositionDeleteLoader(fs, baseTasks)
+ defer positionDeleteLoader.release()
+ deletionVectorLoader, err := newLazyDeletionVectorLoader(fs, baseTasks)
+ if err != nil {
+ return nil, err
+ }
+
+ result := &EqualityDeleteToDVResult{ConvertedEqualityDeleteFiles:
len(equalityDeleteFiles)}
+ deleteFilesToRemove := make([]iceberg.DataFile, 0,
len(equalityDeleteFiles))
+ addRemoval := func(file iceberg.DataFile) {
+ deleteFilesToRemove = append(deleteFilesToRemove, file)
+ if IsDeletionVector(file) {
+ result.ReplacedDeletionVectors++
+ }
+ }
+
+ equalityPaths := make([]string, 0, len(equalityDeleteFiles))
+ for path := range equalityDeleteFiles {
+ equalityPaths = append(equalityPaths, path)
+ }
+ slices.Sort(equalityPaths)
+ for _, path := range equalityPaths {
+ addRemoval(equalityDeleteFiles[path])
+ }
+
+ var writer *dv.DVWriter
+ seenDataPaths := make(map[string]struct{}, len(positionsByPath))
+ for i, task := range baseTasks {
+ path := task.File.FilePath()
+ if _, seen := seenDataPaths[path]; seen {
+ continue
+ }
+ seenDataPaths[path] = struct{}{}
+ positions := positionsByPath[path].bitmap
+ if positions == nil || positions.IsEmpty() {
+ continue
+ }
+ partition := dataFilePartition(task.File)
+ if writer == nil {
+ writer = dv.NewDVWriter(writeFS, func(id int32)
*iceberg.PartitionSpec {
+ return metadata.PartitionSpecByID(int(id))
+ })
+ }
+ writer.Load(path, positions, task.File.SpecID(), partition)
+
+ var positionDeletes *dv.RoaringPositionBitmap
+ if positionDeleteLoader != nil {
+ positionDeletes, err = positionDeleteLoader.load(ctx,
i, path)
+ if err != nil {
+ return nil, fmt.Errorf("read existing position
deletes for %s: %w", path, err)
+ }
+ }
+ if positionDeletes != nil {
+ writer.Load(path, positionDeletes, task.File.SpecID(),
partition)
+ }
+ existingDV, err := deletionVectorLoader.load(ctx, path)
+ if err != nil {
+ return nil, fmt.Errorf("read existing deletion vector
for %s: %w", path, err)
+ }
+ if existingDV != nil {
+ writer.Load(path, existingDV, task.File.SpecID(),
partition)
+ }
+ for _, file := range task.DeletionVectorFiles {
+ addRemoval(file)
+ }
+ for _, file := range task.DeleteFiles {
+ if referencedDataFilePath(file) == path {
+ addRemoval(file)
+ result.RemovedPositionDeleteFiles++
+ }
+ }
+ }
+
+ var additions []rewriteDeleteFileAddition
+ var outputPath string
+ staged := false
+ defer func() {
+ if !staged && outputPath != "" {
+ _ = writeFS.Remove(context.WithoutCancel(ctx),
outputPath)
+ }
+ }()
+ if writer != nil {
+ // Respect staged table location and write.data.path
properties, including
+ // object store location providers. The base table may still
point elsewhere.
+ locationProvider, err := LoadLocationProvider(meta.loc,
meta.props)
+ if err != nil {
+ return nil, fmt.Errorf("resolve deletion vector
location: %w", err)
+ }
+ outputPath = locationProvider.NewDataLocation(
+
"00000-0-"+uuid.New().String()+"-equality-delete-conversion.puffin", "")
+ deletionVectors, err := writer.Flush(ctx, outputPath)
+ if err != nil {
+ return nil, err
+ }
+ sequenceNumber := meta.nextSequenceNumber()
+ for _, file := range deletionVectors {
+ sequence := sequenceNumber
+ additions = append(additions, rewriteDeleteFileAddition{
+ file: file,
+ dataSequenceNumber: &sequence,
Review Comment:
nit: sequence copies a value nothing changes, thus, every addition can share
&sequenceNumber
##########
table/convert_equality_deletes_to_dv.go:
##########
@@ -0,0 +1,382 @@
+// 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"
+ "fmt"
+ "slices"
+ "sync"
+
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/table/dv"
+ "github.com/google/uuid"
+)
+
+// equalityDeletePositionBuffer accumulates matched positions for one data
file.
+// Each buffer has its own mutex so parallel scans of different files do not
+// contend on a table-wide lock. The scan finishes before bitmap is read.
+type equalityDeletePositionBuffer struct {
+ mu sync.Mutex
+ bitmap *dv.RoaringPositionBitmap
+ invalidPosition int64
+}
+
+func (b *equalityDeletePositionBuffer) addPositions(positions []int64) {
+ if len(positions) == 0 {
+ return
+ }
+
+ b.mu.Lock()
+ defer b.mu.Unlock()
+
+ if b.invalidPosition < 0 {
+ return
+ }
+ for _, position := range positions {
+ if position < 0 {
+ b.invalidPosition = position
+
+ return
+ }
+ }
+
+ if b.bitmap == nil {
+ b.bitmap = dv.NewRoaringPositionBitmap()
+ }
+ for _, position := range positions {
+ b.bitmap.Set(uint64(position))
+ }
+}
+
+// EqualityDeleteToDVResult summarizes an equality-delete conversion.
+type EqualityDeleteToDVResult struct {
+ // Table is the table returned by the table-level action. It is nil
when the
+ // action is staged on a Transaction and set after that transaction
commits.
+ Table *Table
+
+ // ConvertedEqualityDeleteFiles is the number of live equality delete
files
+ // removed by the rewrite.
+ ConvertedEqualityDeleteFiles int
+ // AddedDeletionVectors is the number of file-specific deletion vectors
added.
+ AddedDeletionVectors int
+ // RemovedPositionDeleteFiles is the number of legacy position delete
files
+ // scoped to one data file, either by referenced_data_file or equal
+ // file_path bounds, removed with the conversion.
+ RemovedPositionDeleteFiles int
+ // ReplacedDeletionVectors is the number of existing vectors merged
into new
+ // vectors and removed by the rewrite.
+ ReplacedDeletionVectors int
+}
+
+// ConvertEqualityDeletesToDeletionVectors converts live equality deletes on
+// the main branch of a v3 table to file-specific deletion vectors. The rewrite
+// commits as one replace snapshot so changelog readers do not see a second
+// logical deletion event.
+func (t Table) ConvertEqualityDeletesToDeletionVectors(ctx context.Context)
(*EqualityDeleteToDVResult, error) {
+ txn, err := t.NewTransactionOnBranch(MainBranch)
+ if err != nil {
+ return nil, err
+ }
+
+ result, err := txn.ConvertEqualityDeletesToDeletionVectors(ctx)
+ if err != nil {
+ return result, err
+ }
+ if result.ConvertedEqualityDeleteFiles == 0 {
+ result.Table = &t
+
+ return result, nil
+ }
+
+ committed, err := txn.Commit(ctx)
+ if err == nil {
+ result.Table = committed
+ }
+
+ return result, err
+}
+
+// ConvertEqualityDeletesToDeletionVectors stages an equality-delete to
+// deletion-vector conversion on the transaction's branch. Commit the
+// transaction to publish the replace snapshot.
Review Comment:
lets mention the limitations:
- a concurrent commit on the branch fails this with ErrCommitFailed, with no
retry
- unscoped position deletes are rejected
- every affected data file is read
- abandoned transactions leave a Puffin file for orphan cleanup.
##########
table/transaction.go:
##########
@@ -1874,6 +1885,36 @@ func (t *Transaction) replaceFiles(ctx context.Context,
dataFilesToDelete, dataF
if err != nil {
return err
}
+ if cfg.equalityDeleteToDVConversion {
+ if !cfg.rewriteSemantics || len(dataFilesToDelete) > 0 ||
len(dataFilesToAdd) > 0 ||
+ len(autoDeleteFilesToRemove) > 0 ||
len(deleteFilesToRemove) == 0 {
Review Comment:
only one caller sets this option, and it builds these values right above the
call, so none of these checks can fire
better: remove this block, and the IsDeletionVector check in the loop
further down
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
