badalprasadsingh commented on code in PR #2434: URL: https://github.com/apache/iceberg-go/pull/2434#discussion_r4480833414
########## table/convert_equality_deletes_to_dv.go: ########## @@ -0,0 +1,346 @@ +// 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" +) + +// 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 explicitly file-scoped legacy + // position delete files 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(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(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 + } + if err := t.AssertRefSnapshotID(""); err != nil { + return nil, err + } + + scan, err := t.Scan(WithScanPlanningMode(ScanPlanningLocal)) + 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]*dv.RoaringPositionBitmap) + positionSinks := make(map[string]func([]int64)) + addedTaskByPath := make(map[string]int) + var positionsMu sync.Mutex + 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 { + filePath := path + positionsByPath[path] = nil + positionSinks[path] = func(positions []int64) { + positionsMu.Lock() + bitmap := positionsByPath[filePath] + if bitmap == nil { + bitmap = dv.NewRoaringPositionBitmap() + positionsByPath[filePath] = bitmap + } + for _, position := range positions { + bitmap.Set(uint64(position)) + } + positionsMu.Unlock() + } + } + + 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() + } + } + + 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)) + seenDeletePaths := make(map[string]struct{}, len(equalityDeleteFiles)) + seenDVRefs := make(map[pathRefKey]struct{}) + addRemoval := func(file iceberg.DataFile) { + if IsDeletionVector(file) { + key := pathRefKey{path: file.FilePath(), ref: explicitReferencedDataFile(file)} + if _, exists := seenDVRefs[key]; exists { + return + } + seenDVRefs[key] = struct{}{} + deleteFilesToRemove = append(deleteFilesToRemove, file) + result.ReplacedDeletionVectors++ + + return + } + if _, exists := seenDeletePaths[file.FilePath()]; exists { + return + } + seenDeletePaths[file.FilePath()] = struct{}{} + deleteFilesToRemove = append(deleteFilesToRemove, file) + } + + 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] + 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 explicitReferencedDataFile(file) == path { + before := len(deleteFilesToRemove) + addRemoval(file) + if len(deleteFilesToRemove) > before { + 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, + }) + } + result.AddedDeletionVectors = len(deletionVectors) + } + + if err := t.replaceFiles(ctx, nil, nil, deleteFilesToRemove, nil, additions, nil, Review Comment: P1: this can add a file's first dv a `txn.Delete` staged earlier then replays on retry without checking for it, so the file gets two live deletion vectos and every scan fails can we add `RowDelta`'s deletion vector check to merge-on-read delete first? ########## table/convert_equality_deletes_to_dv.go: ########## @@ -0,0 +1,346 @@ +// 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" +) + +// 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 explicitly file-scoped legacy + // position delete files 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(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(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 + } + if err := t.AssertRefSnapshotID(""); err != nil { + return nil, err + } Review Comment: what happens here is: if converrsion fails later, for eg, on the pos-delete rejection, this pin stays on the txn, and later operations in it can no longer retry. removing delete files already blocks replay, can we drop it? or move it after replaceFiles succeeds? ########## table/convert_equality_deletes_to_dv.go: ########## @@ -0,0 +1,346 @@ +// 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" +) + +// 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 explicitly file-scoped legacy + // position delete files 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(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(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 + } + if err := t.AssertRefSnapshotID(""); err != nil { + return nil, err + } + + scan, err := t.Scan(WithScanPlanningMode(ScanPlanningLocal)) + 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 + } Review Comment: currently, no test reaches this block, without it, a conversion on a branch would read main's files please add a test with an eq-delete only on an existing branch, and check the rows on both the branch and main ########## table/convert_equality_deletes_to_dv.go: ########## @@ -0,0 +1,346 @@ +// 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" +) + +// 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 explicitly file-scoped legacy + // position delete files 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(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(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 + } + if err := t.AssertRefSnapshotID(""); err != nil { + return nil, err + } + + scan, err := t.Scan(WithScanPlanningMode(ScanPlanningLocal)) Review Comment: you are passing every col of each affected file, but only the equality cols are needed, please pass their names to `WithSelectedFields` keys dropped from the schema still load through the delete loader, and an empty list falls back to all cols ########## table/convert_equality_deletes_to_dv.go: ########## @@ -0,0 +1,346 @@ +// 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" +) + +// 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 explicitly file-scoped legacy + // position delete files 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(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(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 + } + if err := t.AssertRefSnapshotID(""); err != nil { + return nil, err + } + + scan, err := t.Scan(WithScanPlanningMode(ScanPlanningLocal)) + 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]*dv.RoaringPositionBitmap) + positionSinks := make(map[string]func([]int64)) + addedTaskByPath := make(map[string]int) + var positionsMu sync.Mutex + 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 { + filePath := path + positionsByPath[path] = nil + positionSinks[path] = func(positions []int64) { + positionsMu.Lock() + bitmap := positionsByPath[filePath] + if bitmap == nil { + bitmap = dv.NewRoaringPositionBitmap() + positionsByPath[filePath] = bitmap + } + for _, position := range positions { + bitmap.Set(uint64(position)) + } + positionsMu.Unlock() + } + } + + 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() + } + } + + 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)) + seenDeletePaths := make(map[string]struct{}, len(equalityDeleteFiles)) + seenDVRefs := make(map[pathRefKey]struct{}) + addRemoval := func(file iceberg.DataFile) { Review Comment: these maps never catch anything, each data file is handled once, and each dv or file-scoped delete belongs to one data file `replaceFiles` already rejects real duplicates, could you append directly and drop `seenDVRefs`, `seenDeletePaths` and the `before` count? correct me if I am mistaken ########## table/equality_delete_dv_conversion_test.go: ########## @@ -0,0 +1,407 @@ +// 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 := appendEqualityConversionRows(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) +} + +// 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 := appendEqualityConversionRows(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(fs) + require.NoError(t, err) + + var dvPaths []string + for _, manifest := range manifests { + if manifest.ManifestContent() != iceberg.ManifestContentDeletes { + continue + } + for entry, err := range manifest.Entries(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 := appendEqualityConversionRows(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 = appendEqualityConversionRows(t, tbl, + `[{"id":1,"data":"a"},{"id":2,"data":"b"}]`) + tbl = appendEqualityConversionRows(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}]`) + beforeSnapshot := tbl.CurrentSnapshot().SnapshotID + require.Equal(t, []int64{4}, idsInTable(t, tbl)) + + _, err = tbl.ConvertEqualityDeletesToDeletionVectors(ctx) + require.ErrorIs(t, err, table.ErrInvalidOperation) + require.Equal(t, beforeSnapshot, tbl.CurrentSnapshot().SnapshotID) + require.Equal(t, 1, liveDeleteFileCount(t, tbl, iceberg.EntryContentEqDeletes)) + require.Equal(t, 1, liveDeleteFileCount(t, tbl, iceberg.EntryContentPosDeletes)) + require.Zero(t, liveDVCount(t, tbl)) + require.Equal(t, []int64{4}, idsInTable(t, tbl)) +} + +func TestConvertEqualityDeletesMergesExistingDeletionVector(t *testing.T) { + ctx := t.Context() + tbl := appendEqualityConversionRows(t, newMergeOnReadTestTableVersion(t, "3"), + `[{"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) + tbl = appendEqualityDelete(t, tbl, []int{1}, `[{"id":4}]`) + + result, err := tbl.ConvertEqualityDeletesToDeletionVectors(ctx) + require.NoError(t, err) + require.Equal(t, 1, result.ReplacedDeletionVectors) + require.Equal(t, 1, liveDVCount(t, result.Table)) + require.Equal(t, []int64{1, 3, 5}, idsInTable(t, result.Table), + "merging a new equality delete must not resurrect rows from the old vector") +} + +func TestConvertEqualityDeletesWritesOneVectorPerPartitionFile(t *testing.T) { Review Comment: please add the test for a global equality delete also as you are touching it, you can add the test for an unpartitioned spec as well ########## table/equality_delete_dv_conversion_test.go: ########## @@ -0,0 +1,407 @@ +// 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 := appendEqualityConversionRows(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) +} + +// 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 := appendEqualityConversionRows(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(fs) + require.NoError(t, err) + + var dvPaths []string + for _, manifest := range manifests { + if manifest.ManifestContent() != iceberg.ManifestContentDeletes { + continue + } + for entry, err := range manifest.Entries(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 := appendEqualityConversionRows(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 = appendEqualityConversionRows(t, tbl, + `[{"id":1,"data":"a"},{"id":2,"data":"b"}]`) + tbl = appendEqualityConversionRows(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}]`) + beforeSnapshot := tbl.CurrentSnapshot().SnapshotID + require.Equal(t, []int64{4}, idsInTable(t, tbl)) + + _, err = tbl.ConvertEqualityDeletesToDeletionVectors(ctx) + require.ErrorIs(t, err, table.ErrInvalidOperation) + require.Equal(t, beforeSnapshot, tbl.CurrentSnapshot().SnapshotID) + require.Equal(t, 1, liveDeleteFileCount(t, tbl, iceberg.EntryContentEqDeletes)) + require.Equal(t, 1, liveDeleteFileCount(t, tbl, iceberg.EntryContentPosDeletes)) + require.Zero(t, liveDVCount(t, tbl)) + require.Equal(t, []int64{4}, idsInTable(t, tbl)) +} + +func TestConvertEqualityDeletesMergesExistingDeletionVector(t *testing.T) { + ctx := t.Context() + tbl := appendEqualityConversionRows(t, newMergeOnReadTestTableVersion(t, "3"), + `[{"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) + tbl = appendEqualityDelete(t, tbl, []int{1}, `[{"id":4}]`) + + result, err := tbl.ConvertEqualityDeletesToDeletionVectors(ctx) + require.NoError(t, err) + require.Equal(t, 1, result.ReplacedDeletionVectors) + require.Equal(t, 1, liveDVCount(t, result.Table)) + require.Equal(t, []int64{1, 3, 5}, idsInTable(t, result.Table), + "merging a new equality delete must not resurrect rows from the old vector") +} + +func TestConvertEqualityDeletesWritesOneVectorPerPartitionFile(t *testing.T) { + ctx := t.Context() + tbl := newPartitionedMergeOnReadTestTable(t) + arrowSchema, err := table.SchemaToArrowSchema(tbl.Metadata().CurrentSchema(), nil, false, false) + require.NoError(t, err) + data, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema, []string{ + `[{"id":1,"category":"x"},{"id":2,"category":"x"},{"id":1,"category":"y"},{"id":3,"category":"y"}]`, + }) + require.NoError(t, err) + defer data.Release() + tbl, err = tbl.Append(ctx, array.NewTableReader(data, -1), nil) + require.NoError(t, err) + tbl = appendEqualityDelete(t, tbl, []int{1, 2}, + `[{"id":1,"category":"x"},{"id":1,"category":"y"}]`) + + result, err := tbl.ConvertEqualityDeletesToDeletionVectors(ctx) + require.NoError(t, err) + require.Equal(t, 2, result.ConvertedEqualityDeleteFiles) + require.Equal(t, 2, result.AddedDeletionVectors) + require.Equal(t, 2, liveDVCount(t, result.Table)) + require.Equal(t, []int64{2, 3}, idsInTable(t, result.Table)) +} + +func TestConvertEqualityDeletesAcrossDataFileSplits(t *testing.T) { + ctx := t.Context() + tbl := appendEqualityConversionRows(t, newMergeOnReadTestTableVersion(t, "3"), + `[{"id":1,"data":"a"},{"id":2,"data":"b"},{"id":3,"data":"c"},{"id":4,"data":"d"},{"id":5,"data":"e"},{"id":6,"data":"f"},{"id":7,"data":"g"},{"id":8,"data":"h"},{"id":9,"data":"i"},{"id":10,"data":"j"}]`) + tx := tbl.NewTransaction() + require.NoError(t, tx.SetProperties(iceberg.Properties{table.ReadSplitTargetSizeKey: "1"})) + tbl, err := tx.Commit(ctx) + require.NoError(t, err) + tbl = appendEqualityDelete(t, tbl, []int{1}, `[{"id":2},{"id":7}]`) + + tasks, err := tbl.Scan().PlanFiles(ctx) + require.NoError(t, err) + require.Len(t, tasks, 2, "the data file has two row groups and each must be planned as a split") + require.Equal(t, tasks[0].File.FilePath(), tasks[1].File.FilePath()) + + result, err := tbl.ConvertEqualityDeletesToDeletionVectors(ctx) + require.NoError(t, err) + require.Equal(t, 1, result.AddedDeletionVectors, "all split positions must form one vector for the data file") + require.Equal(t, 1, liveDVCount(t, result.Table)) + require.Equal(t, []int64{1, 3, 4, 5, 6, 8, 9, 10}, idsInTable(t, result.Table)) +} + +func TestConvertEqualityDeletesRemovesDeletesWithNoMatches(t *testing.T) { + tbl := appendEqualityConversionRows(t, newMergeOnReadTestTableVersion(t, "3"), + `[{"id":1,"data":"a"},{"id":2,"data":"b"}]`) + tbl = appendEqualityDelete(t, tbl, []int{1}, `[{"id":99}]`) + + result, err := tbl.ConvertEqualityDeletesToDeletionVectors(t.Context()) + require.NoError(t, err) + require.Equal(t, 1, result.ConvertedEqualityDeleteFiles) + require.Zero(t, result.AddedDeletionVectors) + require.Zero(t, liveDeleteFileCount(t, result.Table, iceberg.EntryContentEqDeletes)) + require.Empty(t, liveDVCount(t, result.Table)) + require.Equal(t, []int64{1, 2}, idsInTable(t, result.Table)) +} + +func TestConvertEqualityDeletesPreservesSameSnapshotSequenceSemantics(t *testing.T) { + ctx := t.Context() + tbl := newMergeOnReadTestTableVersion(t, "3") + dataPath := filepath.ToSlash(tbl.Location()) + "/data/same-snapshot-data.parquet" + equalityPath := filepath.ToSlash(tbl.Location()) + "/data/same-snapshot-equality.parquet" + dataSchema, err := table.SchemaToArrowSchema(tbl.Metadata().CurrentSchema(), nil, true, false) + require.NoError(t, err) + writeParquetFile(t, dataPath, dataSchema, `[{"id":7,"data":"same snapshot"}]`) + equalitySchema, err := table.SchemaToArrowSchema( + iceberg.NewSchema(0, iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}), + nil, true, false) + require.NoError(t, err) + writeParquetFile(t, equalityPath, equalitySchema, `[{"id":7}]`) + info, err := os.Stat(equalityPath) + require.NoError(t, err) + equalityBuilder, err := iceberg.NewDataFileBuilder( + *iceberg.UnpartitionedSpec, iceberg.EntryContentEqDeletes, + equalityPath, iceberg.ParquetFile, nil, nil, nil, 1, info.Size()) + require.NoError(t, err) + equalityBuilder.EqualityFieldIDs([]int{1}) + + tx := tbl.NewTransaction() + require.NoError(t, tx.NewRowDelta(nil). + AddRows(changelogDataFile(t, dataPath, 1)). + AddDeletes(equalityBuilder.Build()). + Commit(ctx)) + tbl, err = tx.Commit(ctx) + require.NoError(t, err) + require.Equal(t, []int64{7}, idsInTable(t, tbl), "same-snapshot equality deletes do not apply to same-snapshot data") + + result, err := tbl.ConvertEqualityDeletesToDeletionVectors(ctx) + require.NoError(t, err) + require.Equal(t, 1, result.ConvertedEqualityDeleteFiles) + require.Zero(t, result.AddedDeletionVectors) + require.Equal(t, []int64{7}, idsInTable(t, result.Table)) +} + +// Replacing equality deletes with a new DV must not change the applicability +// of older delete files to newer data files that contain the same key. +func TestConvertEqualityDeletesPreservesMultipleDeleteSequences(t *testing.T) { + ctx := t.Context() + tbl := appendEqualityConversionRows(t, newMergeOnReadTestTableVersion(t, "3"), + `[{"id":1,"data":"old"},{"id":2,"data":"b"},{"id":3,"data":"c"}]`) + tbl = appendEqualityDelete(t, tbl, []int{1}, `[{"id":1}]`) + tbl = appendEqualityConversionRows(t, tbl, + `[{"id":1,"data":"new"},{"id":4,"data":"d"}]`) + tbl = appendEqualityDelete(t, tbl, []int{1}, `[{"id":2}]`) + require.Equal(t, []int64{1, 3, 4}, idsInTable(t, tbl)) + + result, err := tbl.ConvertEqualityDeletesToDeletionVectors(ctx) + require.NoError(t, err) + require.Equal(t, 2, result.ConvertedEqualityDeleteFiles) + require.Equal(t, []int64{1, 3, 4}, idsInTable(t, result.Table), + "an older equality delete must not delete a newer row after conversion") + require.Zero(t, liveDeleteFileCount(t, result.Table, iceberg.EntryContentEqDeletes)) + + // Re-running the maintenance action must not create another snapshot. + convertedSnapshotID := result.Table.CurrentSnapshot().SnapshotID + again, err := result.Table.ConvertEqualityDeletesToDeletionVectors(ctx) + require.NoError(t, err) + require.Zero(t, again.ConvertedEqualityDeleteFiles) + require.Zero(t, again.AddedDeletionVectors) + require.Equal(t, convertedSnapshotID, again.Table.CurrentSnapshot().SnapshotID) +} + +func TestConvertEqualityDeletesRequiresV3(t *testing.T) { + tbl := appendEqualityConversionRows(t, newMergeOnReadTestTableVersion(t, "2"), + `[{"id":1,"data":"a"}]`) + tbl = appendEqualityDelete(t, tbl, []int{1}, `[{"id":1}]`) + + _, err := tbl.ConvertEqualityDeletesToDeletionVectors(t.Context()) + require.ErrorIs(t, err, iceberg.ErrNotImplemented) +} + +func appendEqualityConversionRows(t *testing.T, tbl *table.Table, rowsJSON string) *table.Table { Review Comment: this does same as `appendRows` in cow_rewrite_deletes_test.go, please use that ########## table/convert_equality_deletes_to_dv.go: ########## @@ -0,0 +1,346 @@ +// 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" +) + +// 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 explicitly file-scoped legacy + // position delete files 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(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(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 + } + if err := t.AssertRefSnapshotID(""); err != nil { + return nil, err + } + + scan, err := t.Scan(WithScanPlanningMode(ScanPlanningLocal)) + 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]*dv.RoaringPositionBitmap) + positionSinks := make(map[string]func([]int64)) + addedTaskByPath := make(map[string]int) + var positionsMu sync.Mutex + 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 { + filePath := path + positionsByPath[path] = nil + positionSinks[path] = func(positions []int64) { + positionsMu.Lock() + bitmap := positionsByPath[filePath] + if bitmap == nil { + bitmap = dv.NewRoaringPositionBitmap() + positionsByPath[filePath] = bitmap + } + for _, position := range positions { + bitmap.Set(uint64(position)) + } + positionsMu.Unlock() + } + } + + 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() + } + } + + 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)) + seenDeletePaths := make(map[string]struct{}, len(equalityDeleteFiles)) + seenDVRefs := make(map[pathRefKey]struct{}) + addRemoval := func(file iceberg.DataFile) { + if IsDeletionVector(file) { + key := pathRefKey{path: file.FilePath(), ref: explicitReferencedDataFile(file)} + if _, exists := seenDVRefs[key]; exists { + return + } + seenDVRefs[key] = struct{}{} + deleteFilesToRemove = append(deleteFilesToRemove, file) + result.ReplacedDeletionVectors++ + + return + } + if _, exists := seenDeletePaths[file.FilePath()]; exists { + return + } + seenDeletePaths[file.FilePath()] = struct{}{} + deleteFilesToRemove = append(deleteFilesToRemove, file) + } + + 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] + 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 explicitReferencedDataFile(file) == path { Review Comment: missed the file-scoped deletes with only equal `file_path` bounds, the validator still counts them so the conversion fails can we addd `referencedDataFilePath(file)` here, and update the `RemovedPositionDeleteFiles` doc? would be great. ########## table/convert_equality_deletes_to_dv.go: ########## @@ -0,0 +1,346 @@ +// 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" +) + +// 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 explicitly file-scoped legacy + // position delete files 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(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(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 + } + if err := t.AssertRefSnapshotID(""); err != nil { + return nil, err + } + + scan, err := t.Scan(WithScanPlanningMode(ScanPlanningLocal)) + 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]*dv.RoaringPositionBitmap) + positionSinks := make(map[string]func([]int64)) + addedTaskByPath := make(map[string]int) + var positionsMu sync.Mutex + 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 { + filePath := path + positionsByPath[path] = nil + positionSinks[path] = func(positions []int64) { + positionsMu.Lock() + bitmap := positionsByPath[filePath] + if bitmap == nil { + bitmap = dv.NewRoaringPositionBitmap() + positionsByPath[filePath] = bitmap + } + for _, position := range positions { + bitmap.Set(uint64(position)) + } + positionsMu.Unlock() + } + } + + 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() + } + } + + 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)) + seenDeletePaths := make(map[string]struct{}, len(equalityDeleteFiles)) + seenDVRefs := make(map[pathRefKey]struct{}) + addRemoval := func(file iceberg.DataFile) { + if IsDeletionVector(file) { + key := pathRefKey{path: file.FilePath(), ref: explicitReferencedDataFile(file)} + if _, exists := seenDVRefs[key]; exists { + return + } + seenDVRefs[key] = struct{}{} + deleteFilesToRemove = append(deleteFilesToRemove, file) + result.ReplacedDeletionVectors++ + + return + } + if _, exists := seenDeletePaths[file.FilePath()]; exists { + return + } + seenDeletePaths[file.FilePath()] = struct{}{} + deleteFilesToRemove = append(deleteFilesToRemove, file) + } + + 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] + 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 explicitReferencedDataFile(file) == path { + before := len(deleteFilesToRemove) + addRemoval(file) + if len(deleteFilesToRemove) > before { + 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, + }) + } + result.AddedDeletionVectors = len(deletionVectors) + } + + if err := t.replaceFiles(ctx, nil, nil, deleteFilesToRemove, nil, additions, nil, + withRewriteSemantics(), withEqualityDeleteToDVConversion()); err != nil { Review Comment: also writing `schema.name-mapping.default` to tbl properties, though no data file is added catalogs that reject that property can't run this conversion, and callers can't opt out, can we pass `WithoutAutoNameMapping()` here? -- 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]
