You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/08/17 16:51:33 UTC
(iceberg-go) branch main updated: perf(table): index positional deletes by path and partition (#1753)
This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new baf7d97f8 perf(table): index positional deletes by path and partition (#1753)
baf7d97f8 is described below
commit baf7d97f88abce8f9da72cece05309f346226202
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Mon Aug 17 18:51:27 2026 +0200
perf(table): index positional deletes by path and partition (#1753)
* perf(table): index positional deletes by path and partition
* fix(table): preserve positional delete metrics pruning
* test(table): benchmark selective delete partitions
* test(table): compare positional delete index with reference
* fix(table): use canonical partition keys in equality index
* fix(table): resolve positional delete rebase
---
table/conflict_validation.go | 39 ++-
table/conflict_validation_test.go | 28 +--
table/equality_delete_index.go | 4 +-
table/positional_delete_index.go | 154 ++++++++++++
table/positional_delete_index_bench_test.go | 363 ++++++++++++++++++++++++++++
table/positional_delete_index_test.go | 343 ++++++++++++++++++++++++++
table/scanner.go | 63 +++--
7 files changed, 934 insertions(+), 60 deletions(-)
diff --git a/table/conflict_validation.go b/table/conflict_validation.go
index f7955fa78..efc28f4d4 100644
--- a/table/conflict_validation.go
+++ b/table/conflict_validation.go
@@ -653,7 +653,7 @@ func validateNoNewDeletesForRewrittenFiles(ctx *conflictContext, rewrittenFiles
rewrittenPartitions := make(map[string]struct{}, len(rewrittenFiles))
for _, df := range rewrittenFiles {
rewrittenPaths[df.FilePath()] = struct{}{}
- key, err := partitionConflictKey(df.SpecID(), dataFilePartition(df))
+ key, err := canonicalPartitionKey(df.SpecID(), dataFilePartition(df))
if err != nil {
return fmt.Errorf("building partition conflict key for rewritten file %s (spec %d): %w",
df.FilePath(), df.SpecID(), err)
@@ -674,7 +674,7 @@ func validateNoNewDeletesForRewrittenFiles(ctx *conflictContext, rewrittenFiles
return nil
}
- key, err := partitionConflictKey(df.SpecID(), dataFilePartition(df))
+ key, err := canonicalPartitionKey(df.SpecID(), dataFilePartition(df))
if err != nil {
return fmt.Errorf("building partition conflict key for concurrent pos-delete %s (spec %d): %w",
df.FilePath(), df.SpecID(), err)
@@ -733,24 +733,23 @@ var filePathFieldID = func() int {
return f.ID
}()
-// partitionConflictKey returns a deterministic, injective string key
-// for a (specID, partition) tuple, used to detect overlap between a
-// partition-scoped delete and a rewritten data file. Partition values
-// are compared post-transform (direct equality), matching Java's
-// DeleteFileIndex — no transform-aware logic is needed because both
-// sides store the same post-transform representation.
+// canonicalPartitionKey returns a deterministic, injective string key for a
+// (specID, partition) tuple. Partition values are compared post-transform
+// (direct equality), matching Java's DeleteFileIndex — no transform-aware
+// logic is needed because both sides store the same post-transform
+// representation.
//
// Values are encoded type-normalized across the known Avro-decode and
-// write-side representations (see appendPartitionConflictValue):
+// write-side representations (see appendCanonicalPartitionValue):
// integer-backed types are unified per field, so e.g. an int32 and an
-// iceberg.Date carrying the same day match. An unknown value type
-// fails the key build — and thereby the validation — rather than
-// falling back to a lossy formatting that could silently mismatch.
+// iceberg.Date carrying the same day match. An unknown value type fails the
+// key build rather than falling back to lossy formatting that could silently
+// mismatch.
//
// Different spec IDs never match, even with identical-looking tuples:
// Java's DeleteFileIndex keys its PartitionMap by specId, and field IDs
// are only meaningful within a single spec.
-func partitionConflictKey(specID int32, partition map[int]any) (string, error) {
+func canonicalPartitionKey(specID int32, partition map[int]any) (string, error) {
ids := make([]int, 0, len(partition))
for id := range partition {
ids = append(ids, id)
@@ -763,7 +762,7 @@ func partitionConflictKey(specID int32, partition map[int]any) (string, error) {
buf = strconv.AppendInt(buf, int64(id), 10)
buf = append(buf, ':')
var err error
- buf, err = appendPartitionConflictValue(buf, partition[id])
+ buf, err = appendCanonicalPartitionValue(buf, partition[id])
if err != nil {
return "", fmt.Errorf("partition field %d: %w", id, err)
}
@@ -773,7 +772,7 @@ func partitionConflictKey(specID int32, partition map[int]any) (string, error) {
return string(buf), nil
}
-// appendPartitionConflictValue encodes one partition value into dst by
+// appendCanonicalPartitionValue encodes one partition value into dst by
// semantic class, so different Go representations of the same logical
// value produce the same bytes:
//
@@ -787,7 +786,7 @@ func partitionConflictKey(specID int32, partition map[int]any) (string, error) {
//
// String values are length-prefixed so keys stay injective for inputs
// containing the field separators. Unknown types return an error.
-func appendPartitionConflictValue(dst []byte, v any) ([]byte, error) {
+func appendCanonicalPartitionValue(dst []byte, v any) ([]byte, error) {
switch v := v.(type) {
case nil:
return append(dst, 'z'), nil
@@ -820,15 +819,15 @@ func appendPartitionConflictValue(dst []byte, v any) ([]byte, error) {
case uuid.UUID:
return hex.AppendEncode(append(dst, 'B', ':'), v[:]), nil
case iceberg.Decimal:
- return appendDecimalConflictValue(dst, v), nil
+ return appendCanonicalPartitionDecimal(dst, v), nil
case iceberg.DecimalLiteral:
- return appendDecimalConflictValue(dst, iceberg.Decimal(v)), nil
+ return appendCanonicalPartitionDecimal(dst, iceberg.Decimal(v)), nil
default:
- return nil, fmt.Errorf("unsupported partition value type %T in conflict key", v)
+ return nil, fmt.Errorf("unsupported partition value type %T in canonical key", v)
}
}
-func appendDecimalConflictValue(dst []byte, v iceberg.Decimal) []byte {
+func appendCanonicalPartitionDecimal(dst []byte, v iceberg.Decimal) []byte {
dst = strconv.AppendInt(append(dst, 'd', ':'), int64(v.Scale), 10)
return append(append(dst, ':'), v.Val.BigInt().String()...)
diff --git a/table/conflict_validation_test.go b/table/conflict_validation_test.go
index d6a39504d..ae064485d 100644
--- a/table/conflict_validation_test.go
+++ b/table/conflict_validation_test.go
@@ -492,17 +492,17 @@ func TestReferencedDataFilePath(t *testing.T) {
})
}
-// mustPartitionConflictKey unwraps partitionConflictKey for test
+// mustCanonicalPartitionKey unwraps canonicalPartitionKey for test
// tuples that only contain supported value types.
-func mustPartitionConflictKey(t *testing.T, specID int32, partition map[int]any) string {
+func mustCanonicalPartitionKey(t *testing.T, specID int32, partition map[int]any) string {
t.Helper()
- key, err := partitionConflictKey(specID, partition)
+ key, err := canonicalPartitionKey(specID, partition)
require.NoError(t, err)
return key
}
-func TestPartitionConflictKey(t *testing.T) {
+func TestCanonicalPartitionKey(t *testing.T) {
sampleUUID := uuid.MustParse("12345678-9abc-def0-1234-56789abcdef0")
dec := iceberg.Decimal{Val: decimal128.FromI64(12345), Scale: 2}
decOther := iceberg.Decimal{Val: decimal128.FromI64(12346), Scale: 2}
@@ -532,8 +532,8 @@ func TestPartitionConflictKey(t *testing.T) {
for _, tt := range equal {
t.Run("equal/"+tt.name, func(t *testing.T) {
assert.Equal(t,
- mustPartitionConflictKey(t, 0, tt.a),
- mustPartitionConflictKey(t, 0, tt.b))
+ mustCanonicalPartitionKey(t, 0, tt.a),
+ mustCanonicalPartitionKey(t, 0, tt.b))
})
}
@@ -563,24 +563,24 @@ func TestPartitionConflictKey(t *testing.T) {
for _, tt := range distinct {
t.Run("distinct/"+tt.name, func(t *testing.T) {
assert.NotEqual(t,
- mustPartitionConflictKey(t, 0, tt.a),
- mustPartitionConflictKey(t, 0, tt.b))
+ mustCanonicalPartitionKey(t, 0, tt.a),
+ mustCanonicalPartitionKey(t, 0, tt.b))
})
}
t.Run("different spec ids never collide", func(t *testing.T) {
tuple := map[int]any{1: int32(1), 2: "x"}
assert.NotEqual(t,
- mustPartitionConflictKey(t, 0, tuple),
- mustPartitionConflictKey(t, 1, tuple))
+ mustCanonicalPartitionKey(t, 0, tuple),
+ mustCanonicalPartitionKey(t, 1, tuple))
assert.NotEqual(t,
- mustPartitionConflictKey(t, 0, nil),
- mustPartitionConflictKey(t, 1, nil))
+ mustCanonicalPartitionKey(t, 0, nil),
+ mustCanonicalPartitionKey(t, 1, nil))
})
t.Run("unsupported types fail closed", func(t *testing.T) {
for _, v := range []any{struct{}{}, time.Now()} {
- _, err := partitionConflictKey(0, map[int]any{1: v})
+ _, err := canonicalPartitionKey(0, map[int]any{1: v})
require.Error(t, err)
assert.Contains(t, err.Error(), "unsupported partition value type")
}
@@ -699,7 +699,7 @@ func TestValidateNoNewDeletesForRewrittenFiles_EqDeleteAlwaysConflicts(t *testin
// TestValidateNoNewDeletesForRewrittenFiles_UnsupportedPartitionType
// pins the fail-closed contract: a rewritten file whose partition
-// carries a value type outside partitionConflictKey's closed set must
+// carries a value type outside canonicalPartitionKey's closed set must
// fail the validation loudly instead of silently skipping the file.
// Only the rewritten side is testable — the delete side is decoded
// from real Avro manifests whose value domain is exactly the closed
diff --git a/table/equality_delete_index.go b/table/equality_delete_index.go
index da5440e1f..e049e37cd 100644
--- a/table/equality_delete_index.go
+++ b/table/equality_delete_index.go
@@ -78,7 +78,7 @@ func newEqualityDeletePartitionKey(
}
}
- tuple, err := partitionConflictKey(specID, partition)
+ tuple, err := canonicalPartitionKey(specID, partition)
if err != nil {
return equalityDeletePartitionKey{}, err
}
@@ -126,7 +126,7 @@ func comparableEqualityDeletePartitionValue(value any) (any, error) {
case uuid.UUID:
return equalityDeleteBinaryPartitionValue(value[:]), nil
default:
- encoded, err := appendPartitionConflictValue(nil, value)
+ encoded, err := appendCanonicalPartitionValue(nil, value)
if err != nil {
return nil, err
}
diff --git a/table/positional_delete_index.go b/table/positional_delete_index.go
new file mode 100644
index 000000000..084594c9f
--- /dev/null
+++ b/table/positional_delete_index.go
@@ -0,0 +1,154 @@
+// 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 (
+ "cmp"
+ "fmt"
+ "slices"
+ "sort"
+
+ "github.com/apache/iceberg-go"
+)
+
+// positionalDeleteIndex groups file-scoped deletes by referenced data path
+// and all remaining deletes by partition. Each bucket is sequence-sorted so a
+// data-file lookup only visits deletes that can apply to that file.
+type positionalDeleteIndex struct {
+ byPath map[string][]iceberg.ManifestEntry
+ byPartition map[string][]iceberg.ManifestEntry
+}
+
+func buildPositionalDeleteIndex(entries []iceberg.ManifestEntry) (*positionalDeleteIndex, error) {
+ idx := &positionalDeleteIndex{}
+ for _, entry := range entries {
+ deleteFile := entry.DataFile()
+ if path := referencedDataFilePath(deleteFile); path != "" {
+ if idx.byPath == nil {
+ idx.byPath = make(map[string][]iceberg.ManifestEntry)
+ }
+ idx.byPath[path] = append(idx.byPath[path], entry)
+
+ continue
+ }
+
+ partitionKey, err := canonicalPartitionKey(deleteFile.SpecID(), deleteFile.Partition())
+ if err != nil {
+ return nil, fmt.Errorf("indexing positional delete file %s: %w", deleteFile.FilePath(), err)
+ }
+ if idx.byPartition == nil {
+ idx.byPartition = make(map[string][]iceberg.ManifestEntry)
+ }
+ idx.byPartition[partitionKey] = append(idx.byPartition[partitionKey], entry)
+ }
+
+ sortBySequence := func(entries []iceberg.ManifestEntry) {
+ slices.SortStableFunc(entries, func(a, b iceberg.ManifestEntry) int {
+ return cmp.Compare(a.SequenceNum(), b.SequenceNum())
+ })
+ }
+ for _, pathEntries := range idx.byPath {
+ sortBySequence(pathEntries)
+ }
+ for _, partitionEntries := range idx.byPartition {
+ sortBySequence(partitionEntries)
+ }
+
+ return idx, nil
+}
+
+// forDataFile returns positional deletes with a greater than or equal sequence
+// number. Partition-scoped candidates are pruned using file_path metrics and
+// returned before path-scoped deletes, matching Java's ordering.
+func (idx *positionalDeleteIndex) forDataFile(dataEntry iceberg.ManifestEntry) ([]iceberg.DataFile, error) {
+ if len(idx.byPath) == 0 && len(idx.byPartition) == 0 {
+ return nil, nil
+ }
+
+ dataFile := dataEntry.DataFile()
+ var partitionEntries []iceberg.ManifestEntry
+ if len(idx.byPartition) > 0 {
+ partitionKey, err := canonicalPartitionKey(dataFile.SpecID(), dataFile.Partition())
+ if err != nil {
+ return nil, fmt.Errorf("matching positional deletes to data file %s: %w", dataFile.FilePath(), err)
+ }
+ partitionEntries = idx.byPartition[partitionKey]
+ }
+
+ dataSeqNum := dataEntry.SequenceNum()
+ out, err := appendPartitionDeletesFromSequence(
+ nil, partitionEntries, dataSeqNum, dataFile.FilePath())
+ if err != nil {
+ return nil, err
+ }
+ out = appendPositionalDeletesFromSequence(
+ out, idx.byPath[dataFile.FilePath()], dataSeqNum)
+
+ return out, nil
+}
+
+func appendPartitionDeletesFromSequence(
+ out []iceberg.DataFile,
+ entries []iceberg.ManifestEntry,
+ dataSeqNum int64,
+ dataFilePath string,
+) ([]iceberg.DataFile, error) {
+ start := sort.Search(len(entries), func(i int) bool {
+ return entries[i].SequenceNum() >= dataSeqNum
+ })
+ if start == len(entries) {
+ return out, nil
+ }
+
+ evaluator, err := newInclusiveMetricsEvaluator(
+ iceberg.PositionalDeleteSchema,
+ iceberg.EqualTo(iceberg.Reference("file_path"), dataFilePath),
+ true,
+ false,
+ )
+ if err != nil {
+ return nil, err
+ }
+
+ for _, entry := range entries[start:] {
+ matches, err := evaluator(entry.DataFile())
+ if err != nil {
+ return nil, err
+ }
+ if matches {
+ out = append(out, entry.DataFile())
+ }
+ }
+
+ return out, nil
+}
+
+func appendPositionalDeletesFromSequence(
+ out []iceberg.DataFile,
+ entries []iceberg.ManifestEntry,
+ dataSeqNum int64,
+) []iceberg.DataFile {
+ start := sort.Search(len(entries), func(i int) bool {
+ return entries[i].SequenceNum() >= dataSeqNum
+ })
+ for _, entry := range entries[start:] {
+ out = append(out, entry.DataFile())
+ }
+
+ return out
+}
diff --git a/table/positional_delete_index_bench_test.go b/table/positional_delete_index_bench_test.go
new file mode 100644
index 000000000..f082a977b
--- /dev/null
+++ b/table/positional_delete_index_bench_test.go
@@ -0,0 +1,363 @@
+// 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 (
+ "fmt"
+ "sort"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+)
+
+var positionalDeleteBenchmarkSink int
+
+func BenchmarkPositionalDeleteIndexSparsePaths20KBy5K(b *testing.B) {
+ const (
+ dataFileCount = 20_000
+ deleteFileCount = 5_000
+ )
+ dataEntries, deleteEntries := positionalDeleteBenchmarkEntries(dataFileCount, deleteFileCount)
+
+ b.ReportAllocs()
+ b.ReportMetric(dataFileCount, "data_files")
+ b.ReportMetric(deleteFileCount, "delete_files")
+ b.ResetTimer()
+ for range b.N {
+ positionalDeleteBenchmarkSink = benchmarkPositionalDeleteIndex(b, dataEntries, deleteEntries)
+ }
+ if positionalDeleteBenchmarkSink != deleteFileCount {
+ b.Fatalf("expected %d matches, got %d", deleteFileCount, positionalDeleteBenchmarkSink)
+ }
+}
+
+func BenchmarkPositionalDeletePlanningSparsePaths(b *testing.B) {
+ const (
+ dataFileCount = 2_000
+ deleteFileCount = 500
+ )
+ dataEntries, deleteEntries := positionalDeleteBenchmarkEntries(dataFileCount, deleteFileCount)
+ benchmarkPositionalDeletePlanningVariants(b, dataEntries, deleteEntries, deleteFileCount)
+}
+
+func BenchmarkPositionalDeletePlanningPartitionScopedMetrics(b *testing.B) {
+ const (
+ dataFileCount = 2_000
+ deleteFileCount = 500
+ )
+ dataEntries, deleteEntries := positionalDeletePartitionBenchmarkEntries(
+ dataFileCount, deleteFileCount)
+ benchmarkPositionalDeletePlanningVariants(b, dataEntries, deleteEntries, dataFileCount)
+}
+
+func BenchmarkPositionalDeletePlanningSelectivePartition(b *testing.B) {
+ const (
+ dataFileCount = 2_000
+ deleteFileCount = 500
+ partitionCount = 100
+ )
+ dataEntries, deleteEntries := positionalDeleteSelectivePartitionBenchmarkEntries(
+ dataFileCount, deleteFileCount, partitionCount)
+ expectedMatches := dataFileCount * (deleteFileCount / partitionCount)
+ benchmarkPositionalDeletePlanningPartitionVariants(
+ b, dataEntries, deleteEntries, expectedMatches)
+}
+
+func BenchmarkPositionalDeletePlanningMixed(b *testing.B) {
+ const (
+ dataFileCount = 2_000
+ deleteFileCount = 500
+ )
+ dataEntries, deleteEntries, expectedMatches := positionalDeleteMixedBenchmarkEntries(
+ dataFileCount, deleteFileCount)
+ benchmarkPositionalDeletePlanningVariants(b, dataEntries, deleteEntries, expectedMatches)
+}
+
+func benchmarkPositionalDeletePlanningVariants(
+ b *testing.B,
+ dataEntries, deleteEntries []iceberg.ManifestEntry,
+ expectedMatches int,
+) {
+ b.Run("indexed", func(b *testing.B) {
+ b.ReportAllocs()
+ for range b.N {
+ positionalDeleteBenchmarkSink = benchmarkPositionalDeleteIndex(b, dataEntries, deleteEntries)
+ }
+ if positionalDeleteBenchmarkSink != expectedMatches {
+ b.Fatalf("expected %d matches, got %d", expectedMatches, positionalDeleteBenchmarkSink)
+ }
+ })
+
+ b.Run("metrics_suffix", func(b *testing.B) {
+ b.ReportAllocs()
+ for range b.N {
+ matched := 0
+ for _, dataEntry := range dataEntries {
+ files, err := matchPositionalDeletesByMetrics(dataEntry, deleteEntries)
+ if err != nil {
+ b.Fatal(err)
+ }
+ matched += len(files)
+ }
+ positionalDeleteBenchmarkSink = matched
+ }
+ if positionalDeleteBenchmarkSink != expectedMatches {
+ b.Fatalf("expected %d matches, got %d", expectedMatches, positionalDeleteBenchmarkSink)
+ }
+ })
+}
+
+func benchmarkPositionalDeletePlanningPartitionVariants(
+ b *testing.B,
+ dataEntries, deleteEntries []iceberg.ManifestEntry,
+ expectedMatches int,
+) {
+ b.Run("indexed", func(b *testing.B) {
+ b.ReportAllocs()
+ for range b.N {
+ positionalDeleteBenchmarkSink = benchmarkPositionalDeleteIndex(b, dataEntries, deleteEntries)
+ }
+ if positionalDeleteBenchmarkSink != expectedMatches {
+ b.Fatalf("expected %d matches, got %d", expectedMatches, positionalDeleteBenchmarkSink)
+ }
+ })
+
+ b.Run("partition_suffix", func(b *testing.B) {
+ b.ReportAllocs()
+ for range b.N {
+ matched := 0
+ for _, dataEntry := range dataEntries {
+ files, err := matchPositionalDeletesByPartition(dataEntry, deleteEntries)
+ if err != nil {
+ b.Fatal(err)
+ }
+ matched += len(files)
+ }
+ positionalDeleteBenchmarkSink = matched
+ }
+ if positionalDeleteBenchmarkSink != expectedMatches {
+ b.Fatalf("expected %d matches, got %d", expectedMatches, positionalDeleteBenchmarkSink)
+ }
+ })
+}
+
+func positionalDeleteBenchmarkEntries(
+ dataFileCount, deleteFileCount int,
+) ([]iceberg.ManifestEntry, []iceberg.ManifestEntry) {
+ dataEntries := positionalDeleteBenchmarkDataEntries(dataFileCount)
+
+ deleteEntries := make([]iceberg.ManifestEntry, deleteFileCount)
+ for i := range deleteEntries {
+ path := positionalDeleteBenchmarkPath(i * (dataFileCount / deleteFileCount))
+ deleteEntries[i] = newPositionalDeleteIndexTestEntry(
+ "delete-"+path, 0, nil, 2, &path, path)
+ }
+
+ return dataEntries, deleteEntries
+}
+
+func positionalDeletePartitionBenchmarkEntries(
+ dataFileCount, deleteFileCount int,
+) ([]iceberg.ManifestEntry, []iceberg.ManifestEntry) {
+ dataEntries := positionalDeleteBenchmarkDataEntries(dataFileCount)
+ deleteEntries := make([]iceberg.ManifestEntry, deleteFileCount)
+ filesPerDelete := dataFileCount / deleteFileCount
+ for i := range deleteEntries {
+ first := i * filesPerDelete
+ last := first + filesPerDelete - 1
+ deleteEntries[i] = newPositionalDeleteIndexTestEntryWithBounds(
+ fmt.Sprintf("partition-delete-%06d.parquet", i),
+ 0,
+ nil,
+ 2,
+ nil,
+ positionalDeleteBenchmarkPath(first),
+ positionalDeleteBenchmarkPath(last),
+ )
+ }
+
+ return dataEntries, deleteEntries
+}
+
+func positionalDeleteSelectivePartitionBenchmarkEntries(
+ dataFileCount, deleteFileCount, partitionCount int,
+) ([]iceberg.ManifestEntry, []iceberg.ManifestEntry) {
+ dataEntries := make([]iceberg.ManifestEntry, dataFileCount)
+ for i := range dataEntries {
+ partition := map[int]any{1000: int32(i % partitionCount)}
+ dataEntries[i] = newPositionalDeleteIndexDataEntry(
+ positionalDeleteBenchmarkPath(i), 0, partition, 1)
+ }
+
+ deleteEntries := make([]iceberg.ManifestEntry, deleteFileCount)
+ for i := range deleteEntries {
+ partition := map[int]any{1000: int32(i % partitionCount)}
+ deleteEntries[i] = newPositionalDeleteIndexTestEntry(
+ fmt.Sprintf("partition-delete-%06d.parquet", i),
+ 0,
+ partition,
+ 2,
+ nil,
+ "",
+ )
+ }
+
+ return dataEntries, deleteEntries
+}
+
+func positionalDeleteMixedBenchmarkEntries(
+ dataFileCount, deleteFileCount int,
+) ([]iceberg.ManifestEntry, []iceberg.ManifestEntry, int) {
+ dataEntries := positionalDeleteBenchmarkDataEntries(dataFileCount)
+ fileScopedCount := deleteFileCount / 2
+ partitionScopedCount := deleteFileCount - fileScopedCount
+ deleteEntries := make([]iceberg.ManifestEntry, 0, deleteFileCount)
+
+ fileScopedDataCount := dataFileCount / 2
+ fileStride := fileScopedDataCount / fileScopedCount
+ for i := range fileScopedCount {
+ path := positionalDeleteBenchmarkPath(i * fileStride)
+ deleteEntries = append(deleteEntries, newPositionalDeleteIndexTestEntry(
+ "file-delete-"+path, 0, nil, 2, &path, path))
+ }
+
+ partitionDataCount := dataFileCount - fileScopedDataCount
+ partitionStride := partitionDataCount / partitionScopedCount
+ for i := range partitionScopedCount {
+ first := fileScopedDataCount + i*partitionStride
+ last := first + partitionStride - 1
+ deleteEntries = append(deleteEntries, newPositionalDeleteIndexTestEntryWithBounds(
+ fmt.Sprintf("partition-delete-%06d.parquet", i),
+ 0,
+ nil,
+ 2,
+ nil,
+ positionalDeleteBenchmarkPath(first),
+ positionalDeleteBenchmarkPath(last),
+ ))
+ }
+
+ return dataEntries, deleteEntries, fileScopedCount + partitionDataCount
+}
+
+func positionalDeleteBenchmarkDataEntries(dataFileCount int) []iceberg.ManifestEntry {
+ dataEntries := make([]iceberg.ManifestEntry, dataFileCount)
+ for i := range dataEntries {
+ path := positionalDeleteBenchmarkPath(i)
+ dataEntries[i] = newPositionalDeleteIndexDataEntry(path, 0, nil, 1)
+ }
+
+ return dataEntries
+}
+
+func benchmarkPositionalDeleteIndex(
+ b *testing.B,
+ dataEntries, deleteEntries []iceberg.ManifestEntry,
+) int {
+ idx, err := buildPositionalDeleteIndex(deleteEntries)
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ matched := 0
+ for _, dataEntry := range dataEntries {
+ files, err := idx.forDataFile(dataEntry)
+ if err != nil {
+ b.Fatal(err)
+ }
+ matched += len(files)
+ }
+
+ return matched
+}
+
+func matchPositionalDeletesByMetrics(
+ dataEntry iceberg.ManifestEntry,
+ deleteEntries []iceberg.ManifestEntry,
+) ([]iceberg.DataFile, error) {
+ start := sort.Search(len(deleteEntries), func(i int) bool {
+ return deleteEntries[i].SequenceNum() >= dataEntry.SequenceNum()
+ })
+ evaluator, err := newInclusiveMetricsEvaluator(iceberg.PositionalDeleteSchema,
+ iceberg.EqualTo(iceberg.Reference("file_path"), dataEntry.DataFile().FilePath()),
+ true, false)
+ if err != nil {
+ return nil, err
+ }
+
+ var out []iceberg.DataFile
+ for _, deleteEntry := range deleteEntries[start:] {
+ matches, err := evaluator(deleteEntry.DataFile())
+ if err != nil {
+ return nil, err
+ }
+ if matches {
+ out = append(out, deleteEntry.DataFile())
+ }
+ }
+
+ return out, nil
+}
+
+func matchPositionalDeletesByPartition(
+ dataEntry iceberg.ManifestEntry,
+ deleteEntries []iceberg.ManifestEntry,
+) ([]iceberg.DataFile, error) {
+ dataFile := dataEntry.DataFile()
+ dataPartitionKey, err := canonicalPartitionKey(dataFile.SpecID(), dataFile.Partition())
+ if err != nil {
+ return nil, err
+ }
+
+ start := sort.Search(len(deleteEntries), func(i int) bool {
+ return deleteEntries[i].SequenceNum() >= dataEntry.SequenceNum()
+ })
+ evaluator, err := newInclusiveMetricsEvaluator(iceberg.PositionalDeleteSchema,
+ iceberg.EqualTo(iceberg.Reference("file_path"), dataFile.FilePath()),
+ true, false)
+ if err != nil {
+ return nil, err
+ }
+
+ var out []iceberg.DataFile
+ for _, deleteEntry := range deleteEntries[start:] {
+ deleteFile := deleteEntry.DataFile()
+ deletePartitionKey, err := canonicalPartitionKey(
+ deleteFile.SpecID(), deleteFile.Partition())
+ if err != nil {
+ return nil, err
+ }
+ if deletePartitionKey != dataPartitionKey {
+ continue
+ }
+
+ matches, err := evaluator(deleteFile)
+ if err != nil {
+ return nil, err
+ }
+ if matches {
+ out = append(out, deleteFile)
+ }
+ }
+
+ return out, nil
+}
+
+func positionalDeleteBenchmarkPath(i int) string {
+ return fmt.Sprintf("data-%06d.parquet", i)
+}
diff --git a/table/positional_delete_index_test.go b/table/positional_delete_index_test.go
new file mode 100644
index 000000000..d5dab03a6
--- /dev/null
+++ b/table/positional_delete_index_test.go
@@ -0,0 +1,343 @@
+// 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 (
+ "bytes"
+ "fmt"
+ "reflect"
+ "sort"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+type positionalDeleteIndexTestFile struct {
+ *mockDataFile
+ referencedDataFile *string
+}
+
+func (f *positionalDeleteIndexTestFile) ReferencedDataFile() *string {
+ return f.referencedDataFile
+}
+
+func newPositionalDeleteIndexTestEntry(
+ path string,
+ specID int32,
+ partition map[int]any,
+ sequenceNumber int64,
+ referencedDataFile *string,
+ boundPath string,
+) iceberg.ManifestEntry {
+ return newPositionalDeleteIndexTestEntryWithBounds(
+ path, specID, partition, sequenceNumber, referencedDataFile, boundPath, boundPath)
+}
+
+func newPositionalDeleteIndexTestEntryWithBounds(
+ path string,
+ specID int32,
+ partition map[int]any,
+ sequenceNumber int64,
+ referencedDataFile *string,
+ lowerPath string,
+ upperPath string,
+) iceberg.ManifestEntry {
+ var lowerBounds, upperBounds map[int][]byte
+ if lowerPath != "" {
+ lowerBounds = map[int][]byte{filePathFieldID: []byte(lowerPath)}
+ }
+ if upperPath != "" {
+ upperBounds = map[int][]byte{filePathFieldID: []byte(upperPath)}
+ }
+
+ file := &positionalDeleteIndexTestFile{
+ mockDataFile: &mockDataFile{
+ path: path, specid: specID, partition: partition,
+ contentType: iceberg.EntryContentPosDeletes,
+ lowerBounds: lowerBounds, upperBounds: upperBounds,
+ count: 1,
+ },
+ referencedDataFile: referencedDataFile,
+ }
+
+ return iceberg.NewManifestEntry(
+ iceberg.EntryStatusADDED, nil, &sequenceNumber, nil, file)
+}
+
+func newPositionalDeleteIndexDataEntry(
+ path string,
+ specID int32,
+ partition map[int]any,
+ sequenceNumber int64,
+) iceberg.ManifestEntry {
+ file := &mockDataFile{
+ path: path, specid: specID, partition: partition,
+ contentType: iceberg.EntryContentData,
+ }
+
+ return iceberg.NewManifestEntry(
+ iceberg.EntryStatusADDED, nil, &sequenceNumber, nil, file)
+}
+
+func TestPositionalDeleteIndexMatchesPathPartitionAndSequence(t *testing.T) {
+ partition := map[int]any{1000: []byte{0xde, 0xad}}
+ dataPath := "data.parquet"
+ otherPath := "other.parquet"
+ deleteEntries := []iceberg.ManifestEntry{
+ newPositionalDeleteIndexTestEntry(
+ "path-same-sequence.parquet", 1, partition, 5, &dataPath, ""),
+ newPositionalDeleteIndexTestEntry(
+ "path-newer-from-bounds.parquet", 1, partition, 7, nil, dataPath),
+ newPositionalDeleteIndexTestEntry(
+ "other-path.parquet", 1, partition, 8, &otherPath, ""),
+ newPositionalDeleteIndexTestEntry(
+ "partition-same-sequence.parquet", 1, partition, 5, nil, ""),
+ newPositionalDeleteIndexTestEntry(
+ "partition-newer.parquet", 1,
+ map[int]any{1000: append([]byte(nil), 0xde, 0xad)}, 6, nil, ""),
+ newPositionalDeleteIndexTestEntry(
+ "partition-older.parquet", 1, partition, 4, nil, ""),
+ newPositionalDeleteIndexTestEntry(
+ "different-partition.parquet", 1, map[int]any{1000: []byte{0xbe, 0xef}}, 8, nil, ""),
+ newPositionalDeleteIndexTestEntry(
+ "different-spec.parquet", 2, partition, 8, nil, ""),
+ }
+
+ idx, err := buildPositionalDeleteIndex(deleteEntries)
+ require.NoError(t, err)
+
+ dataEntry := newPositionalDeleteIndexDataEntry(dataPath, 1, partition, 5)
+ matched, err := idx.forDataFile(dataEntry)
+ require.NoError(t, err)
+ require.Len(t, matched, 4)
+ assert.Equal(t, []string{
+ "partition-same-sequence.parquet",
+ "partition-newer.parquet",
+ "path-same-sequence.parquet",
+ "path-newer-from-bounds.parquet",
+ }, positionalDeletePaths(matched))
+}
+
+func TestPositionalDeleteIndexMatchesReferenceApplicability(t *testing.T) {
+ dataA := "data-a.parquet"
+ dataB := "data-b.parquet"
+ otherData := "other-data.parquet"
+ usPartition := map[int]any{1000: "us", 1001: int32(7)}
+ euPartition := map[int]any{1000: "eu", 1001: int32(7)}
+
+ // Keep the input deliberately out of sequence and bucket order. The
+ // reference below applies the Iceberg rules directly rather than calling
+ // the previous scanner matcher or any index classification helper.
+ deleteEntries := []iceberg.ManifestEntry{
+ newPositionalDeleteIndexTestEntryWithBounds(
+ "partition-us-after.parquet", 1, usPartition, 8, nil,
+ "data-x.parquet", "data-z.parquet"),
+ newPositionalDeleteIndexTestEntry(
+ "path-explicit-a-same-sequence.parquet", 99, euPartition, 5, &dataA, ""),
+ newPositionalDeleteIndexTestEntry(
+ "partition-eu-newer.parquet", 1, euPartition, 6, nil, ""),
+ newPositionalDeleteIndexTestEntry(
+ "path-inferred-b-newer.parquet", 2, euPartition, 7, nil, dataB),
+ newPositionalDeleteIndexTestEntry(
+ "partition-us-older.parquet", 1, usPartition, 4, nil, ""),
+ newPositionalDeleteIndexTestEntryWithBounds(
+ "partition-us-range.parquet", 1, usPartition, 7, nil, dataA, dataB),
+ newPositionalDeleteIndexTestEntry(
+ "path-explicit-other.parquet", 1, usPartition, 9, &otherData, ""),
+ newPositionalDeleteIndexTestEntry(
+ "partition-other-spec.parquet", 2, usPartition, 6, nil, ""),
+ newPositionalDeleteIndexTestEntry(
+ "path-explicit-a-older.parquet", 1, usPartition, 4, &dataA, ""),
+ newPositionalDeleteIndexTestEntryWithBounds(
+ "partition-us-before.parquet", 1, usPartition, 8, nil,
+ "data-0.parquet", "data-9.parquet"),
+ newPositionalDeleteIndexTestEntry(
+ "path-inferred-a-newer.parquet", 2, euPartition, 7, nil, dataA),
+ newPositionalDeleteIndexTestEntry(
+ "partition-us-same-sequence.parquet", 1, usPartition, 5, nil, ""),
+ newPositionalDeleteIndexTestEntry(
+ "path-explicit-a-unknown-sequence.parquet", 1, usPartition, -1, &dataA, ""),
+ }
+
+ dataEntries := []iceberg.ManifestEntry{
+ newPositionalDeleteIndexDataEntry(dataA, 1, usPartition, 5),
+ newPositionalDeleteIndexDataEntry(dataB, 1, usPartition, 6),
+ newPositionalDeleteIndexDataEntry("data-c.parquet", 1, euPartition, 5),
+ newPositionalDeleteIndexDataEntry("data-d.parquet", 2, usPartition, 5),
+ newPositionalDeleteIndexDataEntry(dataA, 1, usPartition, -1),
+ newPositionalDeleteIndexDataEntry(dataA, 1, usPartition, 8),
+ }
+
+ pathField, ok := iceberg.PositionalDeleteSchema.FindFieldByName("file_path")
+ require.True(t, ok)
+ idx, err := buildPositionalDeleteIndex(deleteEntries)
+ require.NoError(t, err)
+
+ for _, dataEntry := range dataEntries {
+ dataEntry := dataEntry
+ t.Run(fmt.Sprintf("%s/spec=%d/sequence=%d",
+ dataEntry.DataFile().FilePath(), dataEntry.DataFile().SpecID(), dataEntry.SequenceNum()), func(t *testing.T) {
+ matched, err := idx.forDataFile(dataEntry)
+ require.NoError(t, err)
+
+ got := positionalDeletePaths(matched)
+ want := make([]string, 0)
+ for _, deleteEntry := range deleteEntries {
+ if referencePositionalDeleteApplies(dataEntry, deleteEntry, pathField.ID) {
+ want = append(want, deleteEntry.DataFile().FilePath())
+ }
+ }
+ sort.Strings(got)
+ sort.Strings(want)
+ assert.Equal(t, want, got)
+ })
+ }
+}
+
+func referencePositionalDeleteApplies(
+ dataEntry iceberg.ManifestEntry,
+ deleteEntry iceberg.ManifestEntry,
+ pathFieldID int,
+) bool {
+ if deleteEntry.SequenceNum() < dataEntry.SequenceNum() {
+ return false
+ }
+
+ dataFile := dataEntry.DataFile()
+ deleteFile := deleteEntry.DataFile()
+ if referencedPath := deleteFile.ReferencedDataFile(); referencedPath != nil {
+ return *referencedPath == dataFile.FilePath()
+ }
+
+ lower, hasLower := deleteFile.LowerBoundValues()[pathFieldID]
+ upper, hasUpper := deleteFile.UpperBoundValues()[pathFieldID]
+ if hasLower && len(lower) > 0 && bytes.Equal(lower, upper) {
+ return string(lower) == dataFile.FilePath()
+ }
+
+ if deleteFile.SpecID() != dataFile.SpecID() ||
+ !referencePartitionsEqual(deleteFile.Partition(), dataFile.Partition()) {
+ return false
+ }
+
+ path := []byte(dataFile.FilePath())
+ if hasLower && bytes.Compare(path, lower) < 0 {
+ return false
+ }
+ if hasUpper && bytes.Compare(path, upper) > 0 {
+ return false
+ }
+
+ return true
+}
+
+func referencePartitionsEqual(left, right map[int]any) bool {
+ if len(left) != len(right) {
+ return false
+ }
+ for fieldID, leftValue := range left {
+ rightValue, ok := right[fieldID]
+ if !ok || !reflect.DeepEqual(leftValue, rightValue) {
+ return false
+ }
+ }
+
+ return true
+}
+
+func TestPositionalDeleteIndexPrunesPartitionEntriesUsingPathMetrics(t *testing.T) {
+ partition := map[int]any{1000: "partition"}
+ deleteEntries := []iceberg.ManifestEntry{
+ newPositionalDeleteIndexTestEntryWithBounds(
+ "matching-range.parquet", 1, partition, 5, nil, "data-a.parquet", "data-z.parquet"),
+ newPositionalDeleteIndexTestEntryWithBounds(
+ "range-before.parquet", 1, partition, 5, nil, "data-a.parquet", "data-b.parquet"),
+ newPositionalDeleteIndexTestEntryWithBounds(
+ "range-after.parquet", 1, partition, 5, nil, "data-x.parquet", "data-z.parquet"),
+ newPositionalDeleteIndexTestEntry(
+ "no-metrics.parquet", 1, partition, 5, nil, ""),
+ }
+
+ idx, err := buildPositionalDeleteIndex(deleteEntries)
+ require.NoError(t, err)
+
+ dataEntry := newPositionalDeleteIndexDataEntry("data-m.parquet", 1, partition, 5)
+ matched, err := idx.forDataFile(dataEntry)
+ require.NoError(t, err)
+ assert.Equal(t, []string{"matching-range.parquet", "no-metrics.parquet"},
+ positionalDeletePaths(matched))
+}
+
+func TestPositionalDeleteIndexHandlesUnknownSequenceNumbers(t *testing.T) {
+ dataPath := "data.parquet"
+ deleteEntries := []iceberg.ManifestEntry{
+ newPositionalDeleteIndexTestEntry("unknown.parquet", 0, nil, -1, &dataPath, ""),
+ newPositionalDeleteIndexTestEntry("sequence-1.parquet", 0, nil, 1, &dataPath, ""),
+ newPositionalDeleteIndexTestEntry("sequence-3.parquet", 0, nil, 3, &dataPath, ""),
+ }
+
+ idx, err := buildPositionalDeleteIndex(deleteEntries)
+ require.NoError(t, err)
+
+ knownData := newPositionalDeleteIndexDataEntry(dataPath, 0, nil, 2)
+ matched, err := idx.forDataFile(knownData)
+ require.NoError(t, err)
+ assert.Equal(t, []string{"sequence-3.parquet"}, positionalDeletePaths(matched))
+
+ unknownData := newPositionalDeleteIndexDataEntry(dataPath, 0, nil, -1)
+ matched, err = idx.forDataFile(unknownData)
+ require.NoError(t, err)
+ assert.Equal(t, []string{"unknown.parquet", "sequence-1.parquet", "sequence-3.parquet"},
+ positionalDeletePaths(matched))
+}
+
+func TestPositionalDeleteIndexSkipsPartitionKeyForPathOnlyDeletes(t *testing.T) {
+ dataPath := "data.parquet"
+ deleteEntry := newPositionalDeleteIndexTestEntry(
+ "delete.parquet", 1, nil, 2, &dataPath, "")
+
+ idx, err := buildPositionalDeleteIndex([]iceberg.ManifestEntry{deleteEntry})
+ require.NoError(t, err)
+
+ dataEntry := newPositionalDeleteIndexDataEntry(
+ dataPath, 1, map[int]any{1000: struct{}{}}, 1)
+ matched, err := idx.forDataFile(dataEntry)
+ require.NoError(t, err)
+ assert.Equal(t, []string{"delete.parquet"}, positionalDeletePaths(matched))
+}
+
+func TestPositionalDeleteIndexRejectsUnsupportedPartitionValue(t *testing.T) {
+ deleteEntry := newPositionalDeleteIndexTestEntry(
+ "delete.parquet", 1, map[int]any{1000: struct{}{}}, 2, nil, "")
+
+ _, err := buildPositionalDeleteIndex([]iceberg.ManifestEntry{deleteEntry})
+ require.Error(t, err)
+ assert.ErrorContains(t, err, "indexing positional delete file delete.parquet")
+ assert.ErrorContains(t, err, "partition field 1000")
+}
+
+func positionalDeletePaths(files []iceberg.DataFile) []string {
+ paths := make([]string, len(files))
+ for i, file := range files {
+ paths[i] = file.FilePath()
+ }
+
+ return paths
+}
diff --git a/table/scanner.go b/table/scanner.go
index 9201da281..ef7f4d82e 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -18,10 +18,10 @@
package table
import (
- "cmp"
"context"
"fmt"
"iter"
+ "maps"
"math"
"reflect"
"slices"
@@ -572,30 +572,44 @@ func minSequenceNum(manifests []iceberg.ManifestFile) int64 {
return n
}
-func matchDeletesToData(entry iceberg.ManifestEntry, positionalDeletes []iceberg.ManifestEntry) ([]iceberg.DataFile, error) {
- idx, _ := slices.BinarySearchFunc(positionalDeletes, entry, func(me1, me2 iceberg.ManifestEntry) int {
- return cmp.Compare(me1.SequenceNum(), me2.SequenceNum())
- })
-
- evaluator, err := newInclusiveMetricsEvaluator(iceberg.PositionalDeleteSchema,
- iceberg.EqualTo(iceberg.Reference("file_path"), entry.DataFile().FilePath()), true, false)
- if err != nil {
- return nil, err
- }
+// matchEqualityDeletesToData returns the equality delete files that apply to
+// the given data entry. An equality delete applies when:
+// - it has a strictly greater sequence number than the data file
+// - it shares the same partition (for partitioned tables)
+//
+// The "strictly greater" rule ensures that data files committed in the same
+// snapshot as the equality deletes are not affected — this is how RowDelta
+// atomically adds new rows alongside deletes for old rows.
+func matchEqualityDeletesToData(dataEntry iceberg.ManifestEntry, eqDeleteEntries []iceberg.ManifestEntry) []iceberg.DataFile {
+ dataSeqNum := dataEntry.SequenceNum()
+ dataPartition := dataEntry.DataFile().Partition()
out := make([]iceberg.DataFile, 0)
- for _, relevant := range positionalDeletes[idx:] {
- df := relevant.DataFile()
- ok, err := evaluator(df)
- if err != nil {
- return nil, err
+ for _, del := range eqDeleteEntries {
+ // Equality deletes only apply to data files with a strictly lower
+ // sequence number.
+ if del.SequenceNum() <= dataSeqNum {
+ continue
}
- if ok {
- out = append(out, df)
+
+ // For partitioned tables, equality deletes must share the same
+ // partition as the data file. Unpartitioned deletes (nil/empty
+ // partition) apply globally.
+ delPartition := del.DataFile().Partition()
+ if len(delPartition) > 0 && len(dataPartition) > 0 {
+ if !partitionsMatch(dataPartition, delPartition) {
+ continue
+ }
}
+
+ out = append(out, del.DataFile())
}
- return out, nil
+ return out
+}
+
+func partitionsMatch(a, b map[int]any) bool {
+ return maps.EqualFunc(a, b, reflect.DeepEqual)
}
// buildDVIndex indexes deletion vectors by the data file path they reference.
@@ -898,10 +912,11 @@ func (scan *Scan) planFilesLocal(ctx context.Context, acc *scanMetricsAccumulato
return nil, err
}
- // Step 3: Sort positional deletes and match them to data files.
- slices.SortFunc(entries.positionalDeleteEntries, func(a, b iceberg.ManifestEntry) int {
- return cmp.Compare(a.SequenceNum(), b.SequenceNum())
- })
+ // Step 3: Index positional deletes and match them to data files.
+ posDeleteIndex, err := buildPositionalDeleteIndex(entries.positionalDeleteEntries)
+ if err != nil {
+ return nil, err
+ }
dvIndex, err := buildDVIndex(entries.dvEntries)
if err != nil {
@@ -923,7 +938,7 @@ func (scan *Scan) planFilesLocal(ctx context.Context, acc *scanMetricsAccumulato
dvFiles := matchDVToData(e, dvIndex)
var deleteFiles []iceberg.DataFile
if len(dvFiles) == 0 {
- deleteFiles, err = matchDeletesToData(e, entries.positionalDeleteEntries)
+ deleteFiles, err = posDeleteIndex.forDataFile(e)
if err != nil {
return nil, err
}