This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 832bdede3 perf(table): index equality delete metric ranges (#2148)
832bdede3 is described below
commit 832bdede3f3d989a8cfc00e322482017c6aeacbe
Author: Minh Vu <[email protected]>
AuthorDate: Mon Oct 12 20:29:08 2026 +0200
perf(table): index equality delete metric ranges (#2148)
---
table/equality_delete_index.go | 70 ++-
table/equality_delete_range_index.go | 310 ++++++++++++++
.../equality_delete_range_index_properties_test.go | 476 +++++++++++++++++++++
table/equality_delete_range_index_test.go | 249 +++++++++++
4 files changed, 1095 insertions(+), 10 deletions(-)
diff --git a/table/equality_delete_index.go b/table/equality_delete_index.go
index a9716cfe4..1a3f0287e 100644
--- a/table/equality_delete_index.go
+++ b/table/equality_delete_index.go
@@ -34,9 +34,11 @@ import (
// Unpartitioned equality deletes are global and apply across partition specs;
// partitioned deletes only apply to data files in the same spec and partition.
type equalityDeleteIndex struct {
- global []equalityDeleteIndexEntry
- byPartition map[equalityDeletePartitionKey][]equalityDeleteIndexEntry
- schema *iceberg.Schema
+ global []equalityDeleteIndexEntry
+ byPartition
map[equalityDeletePartitionKey][]equalityDeleteIndexEntry
+ globalRangeIndex *equalityDeleteRangeIndex
+ rangeIndexesByPartition
map[equalityDeletePartitionKey]*equalityDeleteRangeIndex
+ schema *iceberg.Schema
}
type equalityDeleteIndexEntry struct {
@@ -497,8 +499,15 @@ func buildEqualityDeleteIndex(
})
}
sortBySequence(idx.global)
- for _, partitionEntries := range idx.byPartition {
+ idx.globalRangeIndex = newEqualityDeleteRangeIndex(idx.global)
+ for key, partitionEntries := range idx.byPartition {
sortBySequence(partitionEntries)
+ if rangeIndex := newEqualityDeleteRangeIndex(partitionEntries);
rangeIndex != nil {
+ if idx.rangeIndexesByPartition == nil {
+ idx.rangeIndexesByPartition =
make(map[equalityDeletePartitionKey]*equalityDeleteRangeIndex)
+ }
+ idx.rangeIndexesByPartition[key] = rangeIndex
+ }
}
return idx, nil
@@ -513,6 +522,7 @@ func (idx *equalityDeleteIndex) forDataFile(dataEntry
iceberg.ManifestEntry) ([]
}
partitionEntries := []equalityDeleteIndexEntry(nil)
+ var partitionRangeIndex *equalityDeleteRangeIndex
if len(idx.byPartition) > 0 {
dataFile := dataEntry.DataFile()
partition := dataFilePartition(dataFile)
@@ -522,6 +532,7 @@ func (idx *equalityDeleteIndex) forDataFile(dataEntry
iceberg.ManifestEntry) ([]
return nil, fmt.Errorf("matching equality
deletes to data file %s: %w", dataFile.FilePath(), err)
}
partitionEntries = idx.byPartition[key]
+ partitionRangeIndex = idx.rangeIndexesByPartition[key]
}
}
@@ -532,21 +543,60 @@ func (idx *equalityDeleteIndex) forDataFile(dataEntry
iceberg.ManifestEntry) ([]
dataStats = &stats
}
- out := appendEqualityDeletesAfter(nil, idx.global, dataSeqNum,
dataStats)
- out = appendEqualityDeletesAfter(out, partitionEntries, dataSeqNum,
dataStats)
+ out := appendEqualityDeletesAfterIndexed(
+ nil, idx.global, dataSeqNum, dataStats, idx.globalRangeIndex)
+ out = appendEqualityDeletesAfterIndexed(
+ out, partitionEntries, dataSeqNum, dataStats,
partitionRangeIndex)
return out, nil
}
-func appendEqualityDeletesAfter(
+func firstEqualityDeleteAfter(
+ entries []equalityDeleteIndexEntry,
+ dataSeqNum int64,
+) int {
+ return sort.Search(len(entries), func(i int) bool {
+ return entries[i].entry.SequenceNum() > dataSeqNum
+ })
+}
+
+func appendEqualityDeletesAfterIndexed(
out []iceberg.DataFile,
entries []equalityDeleteIndexEntry,
dataSeqNum int64,
dataStats *equalityDeleteDataFileStats,
+ rangeIndex *equalityDeleteRangeIndex,
+) []iceberg.DataFile {
+ start := firstEqualityDeleteAfter(entries, dataSeqNum)
+ if rangeIndex == nil || dataStats == nil ||
+ len(dataStats.lowerBounds) == 0 || len(dataStats.upperBounds)
== 0 ||
+ len(entries)-start < equalityDeleteRangeIndexMinLookupEntries {
+ return appendEqualityDeletesFrom(out, entries, start, dataStats)
+ }
+
+ candidates, ok := rangeIndex.candidates(start, dataSeqNum, dataStats)
+ if !ok {
+ return appendEqualityDeletesFrom(out, entries, start, dataStats)
+ }
+
+ out = slices.Grow(out, len(candidates))
+ for _, entryIndex := range candidates {
+ entry := entries[entryIndex]
+ if !equalityDeleteCanContainData(dataStats, entry.fields) {
+ continue
+ }
+ out = append(out, entry.entry.DataFile())
+ }
+
+ return out
+}
+
+func appendEqualityDeletesFrom(
+ out []iceberg.DataFile,
+ entries []equalityDeleteIndexEntry,
+ start int,
+ dataStats *equalityDeleteDataFileStats,
) []iceberg.DataFile {
- start := sort.Search(len(entries), func(i int) bool {
- return entries[i].entry.SequenceNum() > dataSeqNum
- })
for _, entry := range entries[start:] {
if !equalityDeleteCanContainData(dataStats, entry.fields) {
continue
diff --git a/table/equality_delete_range_index.go
b/table/equality_delete_range_index.go
new file mode 100644
index 000000000..d71b53494
--- /dev/null
+++ b/table/equality_delete_range_index.go
@@ -0,0 +1,310 @@
+// 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"
+ "slices"
+ "sort"
+ "sync"
+)
+
+// These floors do not guarantee that the first lookup amortizes construction.
+// BenchmarkEqualityDeleteRangeIndexThreshold measures the cold and repeated
+// lookup tradeoff against linear matching.
+const (
+ // Small buckets rarely amortize tree construction; keep them linear.
+ equalityDeleteRangeIndexMinEntries = 256
+ // Avoid a separate tree for a field represented by only a few deletes.
+ equalityDeleteRangeIndexMinGroupEntries = 64
+ // Even a large bucket should use its sequence-sorted suffix directly
once
+ // fewer than 128 entries remain. At that floor, the half-remaining
limit
+ // below allows fewer than 64 candidates.
+ equalityDeleteRangeIndexMinLookupEntries = 128
+)
+
+type equalityDeleteRangeIndex struct {
+ // Entries and their decoded bounds are immutable after index
construction.
+ // Build the tree only when a lookup has a large enough sequence
suffix, so
+ // unvisited partitions do not pay for sorting or allocating range
nodes.
+ entries []equalityDeleteIndexEntry
+ once sync.Once
+ groups []equalityDeleteRangeGroup
+ fallback []int
+}
+
+type equalityDeleteRangeItem struct {
+ entryIndex int
+ lower equalityDeleteMetricValue
+ upper equalityDeleteMetricValue
+ sequence int64
+}
+
+type equalityDeleteRangeNode struct {
+ entryIndex int
+ left int
+ right int
+ lower equalityDeleteMetricValue
+ upper equalityDeleteMetricValue
+ maxUpper equalityDeleteMetricValue
+ sequence int64
+ maxSequence int64
+}
+
+type equalityDeleteRangeGroup struct {
+ field equalityDeleteFieldMetrics
+ root int
+ nodes []equalityDeleteRangeNode
+}
+
+func newEqualityDeleteRangeIndex(entries []equalityDeleteIndexEntry)
*equalityDeleteRangeIndex {
+ if len(entries) < equalityDeleteRangeIndexMinEntries {
+ return nil
+ }
+
+ return &equalityDeleteRangeIndex{entries: entries}
+}
+
+func (idx *equalityDeleteRangeIndex) build() {
+ entries := idx.entries
+ itemsByField := make(map[int][]equalityDeleteRangeItem)
+ fieldsByID := make(map[int]equalityDeleteFieldMetrics)
+ fallback := make([]int, 0)
+ for entryIndex := range entries {
+ field, ok :=
equalityDeleteRangeIndexField(entries[entryIndex].fields)
+ if !ok {
+ fallback = append(fallback, entryIndex)
+
+ continue
+ }
+
+ itemsByField[field.fieldID] =
append(itemsByField[field.fieldID], equalityDeleteRangeItem{
+ entryIndex: entryIndex,
+ lower: field.lowerValue,
+ upper: field.upperValue,
+ sequence: entries[entryIndex].entry.SequenceNum(),
+ })
+ if _, found := fieldsByID[field.fieldID]; !found {
+ fieldsByID[field.fieldID] = *field
+ }
+ }
+
+ fieldIDs := make([]int, 0, len(itemsByField))
+ for fieldID := range itemsByField {
+ fieldIDs = append(fieldIDs, fieldID)
+ }
+ slices.Sort(fieldIDs)
+
+ groups := make([]equalityDeleteRangeGroup, 0, len(fieldIDs))
+ for _, fieldID := range fieldIDs {
+ items := itemsByField[fieldID]
+ if len(items) < equalityDeleteRangeIndexMinGroupEntries {
+ for _, item := range items {
+ fallback = append(fallback, item.entryIndex)
+ }
+
+ continue
+ }
+
+ groups = append(groups,
newEqualityDeleteRangeGroup(fieldsByID[fieldID], items))
+ }
+
+ slices.Sort(fallback)
+ idx.groups = groups
+ idx.fallback = fallback
+}
+
+// The index may exclude only deletes that equalityDeleteCanContainData would
+// also reject. A ranged field must prove it cannot match null or NaN keys
before
+// the tree can use its bounds. Anything uncertain stays on the linear path.
+func equalityDeleteRangeIndexField(
+ fields []equalityDeleteFieldMetrics,
+) (*equalityDeleteFieldMetrics, bool) {
+ for i := range fields {
+ field := &fields[i]
+ if !field.primitive || !field.canUseRange ||
!field.hasDecodedBounds {
+ continue
+ }
+ if !field.required && (!field.hasNullCount || field.nullCount
!= 0) {
+ continue
+ }
+ if field.floatType && (!field.hasNaNCount || field.nanCount !=
0) {
+ continue
+ }
+
+ // Index one field per delete to keep candidates unique. The
first safe
+ // field may be less selective than a later composite-key
field; the
+ // candidate limit falls back to linear matching in that case.
+ return field, true
+ }
+
+ return nil, false
+}
+
+func newEqualityDeleteRangeGroup(
+ field equalityDeleteFieldMetrics,
+ items []equalityDeleteRangeItem,
+) equalityDeleteRangeGroup {
+ slices.SortFunc(items, func(a, b equalityDeleteRangeItem) int {
+ if order := equalityDeleteMetricValueCompare(&a.lower,
&b.lower); order != 0 {
+ return order
+ }
+
+ return cmp.Compare(a.entryIndex, b.entryIndex)
+ })
+
+ group := equalityDeleteRangeGroup{
+ field: field,
+ root: -1,
+ nodes: make([]equalityDeleteRangeNode, len(items)),
+ }
+ var build func(int, int) int
+ build = func(start, end int) int {
+ if start >= end {
+ return -1
+ }
+
+ middle := start + (end-start)/2
+ item := items[middle]
+ nodeIndex := middle
+
+ left := build(start, middle)
+ right := build(middle+1, end)
+ // Bound values copy scalars or references to bytes already
owned by
+ // entries. Nodes do not clone the string/binary payload again.
+ node := equalityDeleteRangeNode{
+ entryIndex: item.entryIndex,
+ left: left,
+ right: right,
+ lower: item.lower,
+ upper: item.upper,
+ maxUpper: item.upper,
+ sequence: item.sequence,
+ maxSequence: item.sequence,
+ }
+ for _, childIndex := range [...]int{left, right} {
+ if childIndex < 0 {
+ continue
+ }
+ child := &group.nodes[childIndex]
+ if equalityDeleteMetricValueCompare(&child.maxUpper,
&node.maxUpper) > 0 {
+ node.maxUpper = child.maxUpper
+ }
+ if child.maxSequence > node.maxSequence {
+ node.maxSequence = child.maxSequence
+ }
+ }
+ group.nodes[nodeIndex] = node
+
+ return nodeIndex
+ }
+
+ group.root = build(0, len(items))
+
+ return group
+}
+
+func (idx *equalityDeleteRangeIndex) candidates(
+ start int,
+ dataSeqNum int64,
+ dataStats *equalityDeleteDataFileStats,
+) ([]int, bool) {
+ idx.once.Do(idx.build)
+ if len(idx.groups) == 0 {
+ return nil, false
+ }
+
+ remaining := len(idx.entries) - start
+ // Use the linear path once candidates include at least half the
applicable entries.
+ // Stop at that threshold instead of allocating a full candidate list
that
+ // will be discarded before the same entries are scanned again.
+ candidateLimit := remaining / 2
+ fallbackStart := sort.Search(len(idx.fallback), func(i int) bool {
+ return idx.fallback[i] >= start
+ })
+ if len(idx.fallback)-fallbackStart >= candidateLimit {
+ return nil, false
+ }
+
+ candidates := make([]int, 0, min(candidateLimit, 128))
+ candidates = append(candidates, idx.fallback[fallbackStart:]...)
+
+ for i := range idx.groups {
+ var ok bool
+ candidates, ok = idx.groups[i].appendCandidates(candidates,
dataStats, dataSeqNum, candidateLimit)
+ if !ok || len(candidates) >= candidateLimit {
+ return nil, false
+ }
+ }
+
+ slices.Sort(candidates)
+
+ return candidates, true
+}
+
+func (group *equalityDeleteRangeGroup) appendCandidates(
+ out []int,
+ dataStats *equalityDeleteDataFileStats,
+ dataSeqNum int64,
+ candidateLimit int,
+) ([]int, bool) {
+ field := &group.field
+ if field.floatType && !equalityDeleteFloatRangesAreKnown(*dataStats,
field) {
+ return out, false
+ }
+
+ dataBounds := equalityDeleteDataFileBoundsFor(dataStats, field)
+ if !dataBounds.usable {
+ return out, false
+ }
+
+ return group.appendNodeCandidates(out, group.root, dataBounds,
dataSeqNum, candidateLimit), true
+}
+
+func (group *equalityDeleteRangeGroup) appendNodeCandidates(
+ out []int,
+ nodeIndex int,
+ dataBounds *equalityDeleteDataFileBounds,
+ dataSeqNum int64,
+ candidateLimit int,
+) []int {
+ if nodeIndex < 0 || len(out) >= candidateLimit {
+ return out
+ }
+
+ node := &group.nodes[nodeIndex]
+ // These strict disjoint-range checks must remain a subset of the
rejects
+ // in equalityDeleteCanContainData. Equal endpoints still overlap, and
the
+ // strict sequence comparison retains only deletes newer than the data.
+ if node.maxSequence <= dataSeqNum ||
+ equalityDeleteMetricValueCompare(&node.maxUpper,
&dataBounds.lower) < 0 {
+ return out
+ }
+
+ out = group.appendNodeCandidates(out, node.left, dataBounds,
dataSeqNum, candidateLimit)
+ if len(out) >= candidateLimit ||
+ equalityDeleteMetricValueCompare(&node.lower,
&dataBounds.upper) > 0 {
+ return out
+ }
+
+ if node.sequence > dataSeqNum &&
+ equalityDeleteMetricValueCompare(&node.upper,
&dataBounds.lower) >= 0 {
+ out = append(out, node.entryIndex)
+ }
+
+ return group.appendNodeCandidates(out, node.right, dataBounds,
dataSeqNum, candidateLimit)
+}
diff --git a/table/equality_delete_range_index_properties_test.go
b/table/equality_delete_range_index_properties_test.go
new file mode 100644
index 000000000..34b910770
--- /dev/null
+++ b/table/equality_delete_range_index_properties_test.go
@@ -0,0 +1,476 @@
+// 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"
+ "math/rand/v2"
+ "sync"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func equalityDeleteRangeTestSpecs() equalityDeleteIndexTestSpecLookup {
+ specs := equalityDeleteIndexTestSpecs()
+ for _, id := range []int{1, 2} {
+ // Partition on a separate field so equality bounds may vary
within it.
+ specs[id] = iceberg.NewPartitionSpecID(id,
iceberg.PartitionField{
+ SourceIDs: []int{3}, FieldID: 1000, Name: "partition",
+ Transform: iceberg.IdentityTransform{},
+ })
+ }
+
+ return specs
+}
+
+func equalityDeleteRangeTestBucket(t testing.TB, count int, specID int32,
partition map[int]any) []iceberg.ManifestEntry {
+ t.Helper()
+ entries := make([]iceberg.ManifestEntry, count)
+ for i := range entries {
+ lower, upper := equalityDeleteMetricsTestBounds(t, int32(i*10),
int32(i*10+4))
+ entries[i] = newEqualityDeleteMetricsTestEntry(
+ fmt.Sprintf("delete-%d.parquet", i), specID, partition,
iceberg.EntryContentEqDeletes, int64(i+2),
+ []int{1}, map[int]int64{1: 1}, map[int]int64{1: 0},
nil, lower, upper)
+ }
+
+ return entries
+}
+
+func TestEqualityDeleteRangeIndexCandidateLimit(t *testing.T) {
+ entries := equalityDeleteRangeTestBucket(t,
equalityDeleteRangeIndexMinEntries, 0, nil)
+ schema := equalityDeleteMetricsTestSchema(iceberg.PrimitiveTypes.Int32,
true)
+ idx, err := buildEqualityDeleteIndex(entries,
equalityDeleteIndexTestSpecs(), schema)
+ require.NoError(t, err)
+ limit := len(entries) / 2
+ for _, matches := range []int{limit - 1, limit} {
+ lower, upper := equalityDeleteMetricsTestBounds(t, int32(0),
int32((matches-1)*10+4))
+ data := newEqualityDeleteMetricsTestEntry("data.parquet", 0,
nil, iceberg.EntryContentData, 0,
+ nil, nil, nil, nil, lower, upper)
+ stats := equalityDeleteDataFileStatsFor(data.DataFile())
+ candidates, ok := idx.globalRangeIndex.candidates(0, 0, &stats)
+ assert.Equal(t, matches < limit, ok, "candidate count %d, limit
%d", matches, limit)
+ if ok {
+ assert.Len(t, candidates, matches)
+ }
+ got, err := idx.forDataFile(data)
+ require.NoError(t, err)
+ assert.Len(t, got, matches, "falling back must preserve the
matching deletes")
+ }
+}
+
+func TestEqualityDeleteRangeIndexDefersUnusableLookups(t *testing.T) {
+ count := equalityDeleteRangeIndexMinEntries
+ entries := equalityDeleteRangeTestBucket(t, count, 0, nil)
+ schema := equalityDeleteMetricsTestSchema(iceberg.PrimitiveTypes.Int32,
true)
+ for _, remaining := range []int{
+ equalityDeleteRangeIndexMinLookupEntries - 1,
+ equalityDeleteRangeIndexMinLookupEntries,
+ equalityDeleteRangeIndexMinLookupEntries + 1,
+ } {
+ for _, bounds := range []bool{false, true} {
+ t.Run(fmt.Sprintf("remaining=%d/bounds=%t", remaining,
bounds), func(t *testing.T) {
+ idx, err := buildEqualityDeleteIndex(entries,
equalityDeleteIndexTestSpecs(), schema)
+ require.NoError(t, err)
+ require.Empty(t, idx.globalRangeIndex.groups,
"construction must defer the tree")
+ var lower, upper []byte
+ if bounds {
+ lower, upper =
equalityDeleteMetricsTestBounds(t, int32((count-1)*10), int32((count-1)*10+4))
+ }
+ seq := int64(count - remaining + 1)
+ data :=
newEqualityDeleteMetricsTestEntry("data.parquet", 0, nil,
iceberg.EntryContentData, seq,
+ nil, nil, nil, nil, lower, upper)
+ got, err := idx.forDataFile(data)
+ require.NoError(t, err)
+ stats :=
equalityDeleteDataFileStatsFor(data.DataFile())
+ want := appendEqualityDeletesAfterLinear(nil,
idx.global, seq, &stats)
+ assert.Equal(t,
equalityDeleteMetricPaths(want), equalityDeleteMetricPaths(got))
+ assert.Equal(t, bounds && remaining >=
equalityDeleteRangeIndexMinLookupEntries,
+ len(idx.globalRangeIndex.groups) > 0,
"only an eligible lookup should build the tree")
+ })
+ }
+ }
+}
+
+func TestEqualityDeleteRangeIndexMultipleFieldsAndFallback(t *testing.T) {
+ schema := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "first", Type:
iceberg.PrimitiveTypes.Int32},
+ iceberg.NestedField{ID: 2, Name: "second", Type:
iceberg.PrimitiveTypes.Int32},
+ iceberg.NestedField{ID: 3, Name: "small_group", Type:
iceberg.PrimitiveTypes.Int32})
+ count := equalityDeleteRangeIndexMinEntries
+ groupSize := equalityDeleteRangeIndexMinGroupEntries
+ entries := equalityDeleteRangeTestBucket(t, count, 0, nil)
+ for i, entry := range entries {
+ file := entry.DataFile().(*equalityDeleteMetricsTestFile)
+ switch {
+ case i < count-2*groupSize:
+ case i < count-groupSize:
+ file.equalityFieldIDs = []int{1, 2}
+ file.lowerBounds[2], file.upperBounds[2] =
file.lowerBounds[1], file.upperBounds[1]
+ delete(file.lowerBounds, 1)
+ file.nullCounts[2] = 0
+ case i < count-1:
+ file.equalityFieldIDs = []int{3}
+ file.lowerBounds[3], file.upperBounds[3] =
file.lowerBounds[1], file.upperBounds[1]
+ file.nullCounts[3] = 0
+ default:
+ file.equalityFieldIDs = []int{99}
+ }
+ }
+ idx, err := buildEqualityDeleteIndex(entries,
equalityDeleteIndexTestSpecs(), schema)
+ require.NoError(t, err)
+ data := entries[0].DataFile().(*equalityDeleteMetricsTestFile)
+ stats := equalityDeleteDataFileStatsFor(data)
+ stats.lowerBounds = map[int][]byte{1: data.lowerBounds[1], 2:
data.lowerBounds[1]}
+ stats.upperBounds = map[int][]byte{1: data.upperBounds[1], 2:
data.upperBounds[1]}
+ candidates, ok := idx.globalRangeIndex.candidates(0, 0, &stats)
+ require.True(t, ok)
+ require.Len(t, idx.globalRangeIndex.groups, 2)
+ assert.Equal(t, 1, idx.globalRangeIndex.groups[0].field.fieldID)
+ assert.Equal(t, 2, idx.globalRangeIndex.groups[1].field.fieldID,
+ "a composite key must skip an unusable first field")
+ assert.Len(t, idx.globalRangeIndex.fallback, groupSize,
+ "a sub-threshold group and an unknown field must stay in the
fallback list")
+ assert.Len(t, candidates, groupSize+1)
+ got := appendEqualityDeletesAfterIndexed(nil, idx.global, 0, &stats,
idx.globalRangeIndex)
+ want := appendEqualityDeletesAfterLinear(nil, idx.global, 0, &stats)
+ assert.Equal(t, equalityDeleteMetricPaths(want),
equalityDeleteMetricPaths(got))
+}
+
+func equalityDeleteRandomRangeFile(t testing.TB, rng *rand.Rand, fields
[]iceberg.NestedField, safe bool) *equalityDeleteMetricsTestFile {
+ t.Helper()
+ file := &equalityDeleteMetricsTestFile{mockDataFile: &mockDataFile{
+ format: iceberg.ParquetFile, count: 100,
+ valueCounts: make(map[int]int64), nullCounts:
make(map[int]int64), nanCounts: make(map[int]int64),
+ lowerBounds: make(map[int][]byte), upperBounds:
make(map[int][]byte),
+ }}
+ for _, field := range fields {
+ id := field.ID
+ lo := rng.IntN(4096)
+ hi := lo + rng.IntN(8)
+ var lower, upper []byte
+ switch field.Type.(type) {
+ case iceberg.Int32Type:
+ lower, upper = equalityDeleteMetricsTestBounds(t,
int32(lo), int32(hi))
+ case iceberg.Int64Type:
+ lower, upper = equalityDeleteMetricsTestBounds(t,
int64(lo), int64(hi))
+ case iceberg.Float32Type:
+ lower, upper = equalityDeleteMetricsTestBounds(t,
float32(lo), float32(hi))
+ case iceberg.Float64Type:
+ lower, upper = equalityDeleteMetricsTestBounds(t,
float64(lo), float64(hi))
+ case iceberg.StringType:
+ lower, upper = equalityDeleteMetricsTestBounds(t,
fmt.Sprintf("%04d", lo), fmt.Sprintf("%04d", hi))
+ case iceberg.BinaryType:
+ lower, upper = equalityDeleteMetricsTestBounds(t,
[]byte(fmt.Sprintf("%04d", lo)), []byte(fmt.Sprintf("%04d", hi)))
+ default:
+ t.Fatalf("unsupported test type %s", field.Type)
+ }
+ file.valueCounts[id], file.nullCounts[id], file.nanCounts[id] =
100, 0, 0
+ file.lowerBounds[id], file.upperBounds[id] = lower, upper
+ if safe {
+ continue
+ }
+ switch rng.IntN(40) {
+ case 0:
+ delete(file.lowerBounds, id)
+ case 1:
+ delete(file.upperBounds, id)
+ case 2:
+ file.lowerBounds[id], file.upperBounds[id] = upper,
lower
+ case 3:
+ delete(file.nullCounts, id)
+ case 4:
+ file.nullCounts[id] = 1
+ case 5:
+ delete(file.nanCounts, id)
+ case 6:
+ file.nanCounts[id] = 1
+ case 7:
+ delete(file.valueCounts, id)
+ case 8:
+ file.nullCounts[id] = 100
+ }
+ }
+
+ return file
+}
+
+func TestEqualityDeleteRangeIndexRandomizedMatchesLinearReference(t
*testing.T) {
+ for _, types := range [][2]iceberg.Type{
+ {iceberg.PrimitiveTypes.Int32, iceberg.PrimitiveTypes.Int64},
+ {iceberg.PrimitiveTypes.Float32,
iceberg.PrimitiveTypes.Float64},
+ {iceberg.PrimitiveTypes.String, iceberg.PrimitiveTypes.Binary},
+ } {
+ t.Run(fmt.Sprintf("%s-%s", types[0], types[1]), func(t
*testing.T) {
+ fields := []iceberg.NestedField{
+ {ID: 1, Name: "first", Type: types[0],
Required: true},
+ {ID: 2, Name: "second", Type: types[1]},
+ }
+ schema := iceberg.NewSchema(1, fields[0], fields[1],
iceberg.NestedField{
+ ID: 3, Name: "partition", Type:
iceberg.PrimitiveTypes.Int32, Required: true,
+ })
+ fastLookups := 0
+ for _, count := range []int{63, 64, 65, 127, 128, 129,
255, 256, 257, 511, 512, 513} {
+ rng := rand.New(rand.NewPCG(uint64(count),
2148))
+ var entries []iceberg.ManifestEntry
+ for bucket := range 3 {
+ for i := range count {
+ file :=
equalityDeleteRandomRangeFile(t, rng, fields, false)
+ file.path =
fmt.Sprintf("bucket-%d/delete-%d.parquet", bucket, i)
+ file.contentType =
iceberg.EntryContentEqDeletes
+ if bucket > 0 {
+ file.specid = 1
+ file.partition =
map[int]any{1000: int32(bucket - 1)}
+ }
+ file.equalityFieldIDs = []int{1
+ rng.IntN(2)}
+ if i%4 == 0 {
+ file.equalityFieldIDs =
[]int{1, 2}
+ if i%16 == 0 {
+
delete(file.lowerBounds, 1)
+ }
+ }
+ if i%37 == 0 {
+ file.equalityFieldIDs =
[]int{99}
+ }
+ seq := int64(1 + rng.IntN(8))
+ entries = append(entries,
iceberg.NewManifestEntry(iceberg.EntryStatusADDED, nil, &seq, nil, file))
+ }
+ }
+ // Input order must not supply hidden sequence
or range ordering.
+ rng.Shuffle(len(entries), func(i, j int) {
entries[i], entries[j] = entries[j], entries[i] })
+ idx, err := buildEqualityDeleteIndex(entries,
equalityDeleteRangeTestSpecs(), schema)
+ require.NoError(t, err)
+ for query := range 64 {
+ file :=
equalityDeleteRandomRangeFile(t, rng, fields, query%8 == 0)
+ file.path, file.contentType =
"data.parquet", iceberg.EntryContentData
+ file.specid = int32(query % 3)
+ if file.specid != 0 {
+ file.partition =
map[int]any{1000: int32(rng.IntN(3))}
+ }
+ seq := int64(rng.IntN(9))
+ if query%8 == 0 {
+ seq = 0
+ }
+ data :=
iceberg.NewManifestEntry(iceberg.EntryStatusADDED, nil, &seq, nil, file)
+ got, err := idx.forDataFile(data)
+ require.NoError(t, err)
+ key, err :=
newEqualityDeletePartitionKey(file.specid, file.partition)
+ require.NoError(t, err)
+ stats :=
equalityDeleteDataFileStatsFor(file)
+ want :=
appendEqualityDeletesAfterLinear(nil, idx.global, seq, &stats)
+ want =
appendEqualityDeletesAfterLinear(want, idx.byPartition[key], seq, &stats)
+ require.Equal(t,
equalityDeleteMetricPaths(want), equalityDeleteMetricPaths(got),
+ "seed=%d/2148 query=%d
sequence=%d", count, query, seq)
+ for _, index := range
[]*equalityDeleteRangeIndex{idx.globalRangeIndex,
idx.rangeIndexesByPartition[key]} {
+ if index == nil {
+ continue
+ }
+ start :=
firstEqualityDeleteAfter(index.entries, seq)
+ if len(index.entries)-start <
equalityDeleteRangeIndexMinLookupEntries {
+ continue
+ }
+ if candidates, ok :=
index.candidates(start, seq, &stats); ok {
+ fastLookups++
+ assert.Less(t,
len(candidates), len(index.entries)-start)
+ }
+ }
+ }
+ }
+ assert.Positive(t, fastLookups, "differential coverage
must exercise the range index")
+ })
+ }
+}
+
+func TestEqualityDeleteRangeIndexConcurrentFirstLookups(t *testing.T) {
+ partition := map[int]any{1000: int32(0)}
+ otherPartition := map[int]any{1000: int32(1)}
+ entries := equalityDeleteRangeTestBucket(t, 512, 0, nil)
+ entries = append(entries, equalityDeleteRangeTestBucket(t, 512, 1,
partition)...)
+ entries = append(entries, equalityDeleteRangeTestBucket(t, 512, 1,
otherPartition)...)
+ for i, entry := range entries {
+ entry.DataFile().(*equalityDeleteMetricsTestFile).path =
fmt.Sprintf("delete-%d.parquet", i)
+ }
+ schema := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 3, Name: "partition", Type:
iceberg.PrimitiveTypes.Int32, Required: true})
+ idx, err := buildEqualityDeleteIndex(entries,
equalityDeleteRangeTestSpecs(), schema)
+ require.NoError(t, err)
+ lower, upper := equalityDeleteMetricsTestBounds(t, int32(2000),
int32(2004))
+ data := newEqualityDeleteMetricsTestEntry("data.parquet", 1, partition,
iceberg.EntryContentData, 0,
+ nil, nil, nil, nil, lower, upper)
+ start := make(chan struct{})
+ var wg sync.WaitGroup
+ for range 8 {
+ wg.Go(func() {
+ <-start
+ got, err := idx.forDataFile(data)
+ if assert.NoError(t, err) {
+ assert.Equal(t, []string{"delete-200.parquet",
"delete-712.parquet"}, equalityDeleteMetricPaths(got))
+ }
+ })
+ }
+ close(start)
+ wg.Wait()
+ otherKey, err := newEqualityDeletePartitionKey(1, otherPartition)
+ require.NoError(t, err)
+ assert.Empty(t, idx.rangeIndexesByPartition[otherKey].groups, "unprobed
partitions must not allocate a tree")
+ assert.Len(t, idx.globalRangeIndex.groups[0].nodes, 512)
+}
+
+func TestEqualityDeleteRangeIndexFloatNaNGuards(t *testing.T) {
+ for _, typ := range []iceberg.Type{iceberg.PrimitiveTypes.Float32,
iceberg.PrimitiveTypes.Float64} {
+ for _, deleteNaNs := range []int64{-1, 0, 1} {
+ for _, dataNaNs := range []int64{-1, 0, 1} {
+
t.Run(fmt.Sprintf("%s/deleteNaNs=%d/dataNaNs=%d", typ, deleteNaNs, dataNaNs),
func(t *testing.T) {
+ fields := []iceberg.NestedField{{ID: 1,
Name: "id", Type: typ, Required: true}}
+ rng := rand.New(rand.NewPCG(2148, 1))
+ file :=
equalityDeleteRandomRangeFile(t, rng, fields, true)
+ file.equalityFieldIDs = []int{1}
+ file.nanCounts[1] = deleteNaNs
+ if deleteNaNs < 0 {
+ delete(file.nanCounts, 1)
+ }
+ var dataLower, dataUpper []byte
+ if
typ.Equals(iceberg.PrimitiveTypes.Float32) {
+ file.lowerBounds[1],
file.upperBounds[1] = equalityDeleteMetricsTestBounds(t, float32(20),
float32(30))
+ dataLower, dataUpper =
equalityDeleteMetricsTestBounds(t, float32(0), float32(10))
+ } else {
+ file.lowerBounds[1],
file.upperBounds[1] = equalityDeleteMetricsTestBounds(t, float64(20),
float64(30))
+ dataLower, dataUpper =
equalityDeleteMetricsTestBounds(t, float64(0), float64(10))
+ }
+ seq := int64(2)
+ entries :=
make([]iceberg.ManifestEntry, equalityDeleteRangeIndexMinEntries)
+ for i := range entries {
+ deleteFile := *file
+ deleteFile.path =
fmt.Sprintf("delete-%d.parquet", i)
+ deleteFile.contentType =
iceberg.EntryContentEqDeletes
+ entries[i] =
iceberg.NewManifestEntry(iceberg.EntryStatusADDED, nil, &seq, nil, &deleteFile)
+ }
+ idx, err :=
buildEqualityDeleteIndex(entries, equalityDeleteIndexTestSpecs(),
iceberg.NewSchema(1, fields...))
+ require.NoError(t, err)
+ nanCounts := map[int]int64{1: dataNaNs}
+ if dataNaNs < 0 {
+ nanCounts = nil
+ }
+ data :=
newEqualityDeleteMetricsTestEntry("data.parquet", 0, nil,
iceberg.EntryContentData, 0,
+ nil, nil, nil, nanCounts,
dataLower, dataUpper)
+ stats :=
equalityDeleteDataFileStatsFor(data.DataFile())
+ candidates, indexed :=
idx.globalRangeIndex.candidates(0, 0, &stats)
+ assert.Equal(t, deleteNaNs == 0 &&
dataNaNs == 0, indexed)
+ got, err := idx.forDataFile(data)
+ require.NoError(t, err)
+ if indexed {
+ assert.Empty(t, candidates)
+ assert.Empty(t, got)
+ } else {
+ assert.Len(t, got,
len(entries), "unknown NaNs cannot prove disjointness")
+ }
+ })
+ }
+ }
+ }
+}
+
+func TestEqualityDeleteRangeIndexOptionalParent(t *testing.T) {
+ schema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 2, Name: "parent", Type: &iceberg.StructType{FieldList:
[]iceberg.NestedField{
+ {ID: 1, Name: "child", Type:
iceberg.PrimitiveTypes.Int32, Required: true},
+ }},
+ })
+ entries := equalityDeleteRangeTestBucket(t,
equalityDeleteRangeIndexMinEntries, 0, nil)
+ for _, entry := range entries {
+
delete(entry.DataFile().(*equalityDeleteMetricsTestFile).nullCounts, 1)
+ }
+ idx, err := buildEqualityDeleteIndex(entries,
equalityDeleteIndexTestSpecs(), schema)
+ require.NoError(t, err)
+ lower, upper := equalityDeleteMetricsTestBounds(t, int32(20_000),
int32(20_004))
+ data := newEqualityDeleteMetricsTestEntry("data.parquet", 0, nil,
iceberg.EntryContentData, 0,
+ nil, map[int]int64{1: 1}, map[int]int64{1: 1}, nil, lower,
upper)
+ got, err := idx.forDataFile(data)
+ require.NoError(t, err)
+ assert.Len(t, got, len(entries), "a required child may be null under an
optional parent")
+ assert.Empty(t, idx.globalRangeIndex.groups, "unknown null keys must
stay off the range tree")
+}
+
+// BenchmarkEqualityDeleteRangeIndexThreshold includes range-tree construction
+// and repeated lookups over already-decoded delete metrics. A single lookup
+// exposes cold-start cost; repeated lookups show when that cost is amortized.
+func BenchmarkEqualityDeleteRangeIndexThreshold(b *testing.B) {
+ schema := equalityDeleteMetricsTestSchema(iceberg.PrimitiveTypes.Int32,
true)
+ for _, count := range []int{255, 256, 257, 511, 512} {
+ manifestEntries := equalityDeleteRangeTestBucket(b, count, 0,
nil)
+ entries := make([]equalityDeleteIndexEntry, count)
+ for i, entry := range manifestEntries {
+ entries[i] = newEqualityDeleteIndexEntry(entry, schema)
+ }
+ for _, lookups := range []int{1, 16, 128} {
+ for _, broad := range []bool{false, true} {
+ lower, upper :=
equalityDeleteMetricsTestBounds(b, int32(100), int32(104))
+ if broad {
+ lower, upper =
equalityDeleteMetricsTestBounds(b, int32(0), int32(count*10))
+ }
+ data :=
newEqualityDeleteMetricsTestEntry("data.parquet", 0, nil,
iceberg.EntryContentData, 0,
+ nil, nil, nil, nil, lower, upper)
+ for _, indexed := range []bool{false, true} {
+
b.Run(fmt.Sprintf("entries=%d/lookups=%d/broad=%t/indexed=%t", count, lookups,
broad, indexed), func(b *testing.B) {
+ b.ReportAllocs()
+ for b.Loop() {
+ var rangeIndex
*equalityDeleteRangeIndex
+ if indexed {
+ rangeIndex =
newEqualityDeleteRangeIndex(entries)
+ }
+ for range lookups {
+ stats :=
equalityDeleteDataFileStatsFor(data.DataFile())
+ files :=
appendEqualityDeletesAfterIndexed(nil, entries, 0, &stats, rangeIndex)
+
equalityDeleteBenchmarkSink = len(files)
+ }
+ want := 1
+ if broad {
+ want = count
+ }
+ if
equalityDeleteBenchmarkSink != want {
+
b.Fatalf("expected %d deletes, got %d", want, equalityDeleteBenchmarkSink)
+ }
+ }
+ })
+ }
+ }
+ }
+ }
+}
+
+var equalityDeleteRangeUnprobedSink *equalityDeleteRangeIndex
+
+func BenchmarkEqualityDeleteRangeIndexUnprobed(b *testing.B) {
+ schema := equalityDeleteMetricsTestSchema(iceberg.PrimitiveTypes.Int32,
true)
+ for _, count := range []int{256, 512, 4096} {
+ manifestEntries := equalityDeleteRangeTestBucket(b, count, 0,
nil)
+ entries := make([]equalityDeleteIndexEntry, count)
+ for i, entry := range manifestEntries {
+ entries[i] = newEqualityDeleteIndexEntry(entry, schema)
+ }
+ b.Run(fmt.Sprintf("entries=%d", count), func(b *testing.B) {
+ b.ReportAllocs()
+ for b.Loop() {
+ equalityDeleteRangeUnprobedSink =
newEqualityDeleteRangeIndex(entries)
+ }
+ })
+ }
+}
diff --git a/table/equality_delete_range_index_test.go
b/table/equality_delete_range_index_test.go
new file mode 100644
index 000000000..61f0fa42b
--- /dev/null
+++ b/table/equality_delete_range_index_test.go
@@ -0,0 +1,249 @@
+// 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"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// appendEqualityDeletesAfterLinear is an independent reference matcher. It
+// scans every entry so neither the sequence search nor range tree can hide a
+// false negative in the production path.
+func appendEqualityDeletesAfterLinear(
+ out []iceberg.DataFile,
+ entries []equalityDeleteIndexEntry,
+ dataSeqNum int64,
+ dataStats *equalityDeleteDataFileStats,
+) []iceberg.DataFile {
+ for _, entry := range entries {
+ if entry.entry.SequenceNum() > dataSeqNum &&
equalityDeleteCanContainData(dataStats, entry.fields) {
+ out = append(out, entry.entry.DataFile())
+ }
+ }
+
+ return out
+}
+
+func TestEqualityDeleteRangeIndexMatchesLinearReference(t *testing.T) {
+ const deleteFileCount = 512
+
+ schema := equalityDeleteMetricsTestSchema(iceberg.PrimitiveTypes.Int32,
true)
+ partition := map[int]any{1000: int32(0)}
+ deleteEntries := make([]iceberg.ManifestEntry, deleteFileCount)
+ for i := range deleteEntries {
+ var lower, upper []byte
+ if i%31 != 0 {
+ lower, upper = equalityDeleteMetricsTestBounds(
+ t, int32(i*10), int32(i*10+4))
+ }
+ deleteEntries[i] = newEqualityDeleteMetricsTestEntry(
+ fmt.Sprintf("delete-%d.parquet", i),
+ 1,
+ partition,
+ iceberg.EntryContentEqDeletes,
+ int64(i+2),
+ []int{1},
+ map[int]int64{1: 1},
+ map[int]int64{1: 0},
+ map[int]int64{1: 0},
+ lower,
+ upper,
+ )
+ }
+
+ idx, err := buildEqualityDeleteIndex(
+ deleteEntries, equalityDeleteIndexTestSpecs(), schema)
+ require.NoError(t, err)
+
+ key, err := newEqualityDeletePartitionKey(1, partition)
+ require.NoError(t, err)
+ require.NotNil(t, idx.rangeIndexesByPartition[key])
+ entries := idx.byPartition[key]
+
+ tests := []struct {
+ name string
+ sequence int64
+ lower int32
+ upper int32
+ indexed bool
+ }{
+ {name: "selective middle", sequence: 0, lower: 2000, upper:
2099, indexed: true},
+ {name: "selective with sequence pruning", sequence: 200, lower:
4000, upper: 4099, indexed: true},
+ {name: "no ranged overlap keeps fallback", sequence: 0, lower:
9000, upper: 9100, indexed: true},
+ {name: "unselective range uses linear fallback", sequence: 0,
lower: 0, upper: 5200},
+ {name: "unselective sequence suffix", sequence: 200, lower: 0,
upper: 5200},
+ {name: "small sequence suffix", sequence: 450, lower: 0, upper:
5200},
+ {name: "no applicable sequences", sequence: 513, lower: 0,
upper: 5200},
+ {name: "equal lower endpoint overlaps", sequence: 0, lower:
2000, upper: 2000, indexed: true},
+ {name: "equal upper endpoint overlaps", sequence: 0, lower:
2004, upper: 2004, indexed: true},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ lower, upper := equalityDeleteMetricsTestBounds(t,
tt.lower, tt.upper)
+ dataEntry := newEqualityDeleteMetricsTestEntry(
+ "data.parquet",
+ 1,
+ partition,
+ iceberg.EntryContentData,
+ tt.sequence,
+ nil,
+ map[int]int64{1: 1},
+ map[int]int64{1: 0},
+ map[int]int64{1: 0},
+ lower,
+ upper,
+ )
+
+ got, err := idx.forDataFile(dataEntry)
+ require.NoError(t, err)
+
+ stats :=
equalityDeleteDataFileStatsFor(dataEntry.DataFile())
+ want := appendEqualityDeletesAfterLinear(
+ nil, entries, dataEntry.SequenceNum(), &stats)
+ assert.Equal(t, equalityDeleteMetricPaths(want),
equalityDeleteMetricPaths(got))
+ start := firstEqualityDeleteAfter(entries,
dataEntry.SequenceNum())
+ if len(entries)-start >=
equalityDeleteRangeIndexMinLookupEntries {
+ candidates, ok :=
idx.rangeIndexesByPartition[key].candidates(start, dataEntry.SequenceNum(),
&stats)
+ assert.Equal(t, tt.indexed, ok, "the expected
fast path must actually run")
+ if ok {
+ assert.Less(t, len(candidates),
len(entries)-start)
+ }
+ }
+ })
+ }
+}
+
+func BenchmarkEqualityDeleteRangeIndexLookup(b *testing.B) {
+ const (
+ dataFileCount = 10_000
+ deleteFileCount = 10_000
+ groupCount = 100
+ filesPerGroup = deleteFileCount / groupCount
+ )
+
+ schema := equalityDeleteMetricsTestSchema(iceberg.PrimitiveTypes.Int32,
true)
+ partition := map[int]any{1000: int32(0)}
+ lowerBounds := make([][]byte, groupCount)
+ upperBounds := make([][]byte, groupCount)
+ for group := range groupCount {
+ lowerBounds[group], upperBounds[group] =
equalityDeleteMetricsTestBounds(
+ b, int32(group*filesPerGroup),
int32((group+1)*filesPerGroup-1))
+ }
+
+ deleteEntries := make([]iceberg.ManifestEntry, deleteFileCount)
+ for i := range deleteEntries {
+ group := i / filesPerGroup
+ deleteEntries[i] = newEqualityDeleteMetricsTestEntry(
+ fmt.Sprintf("delete-%d.parquet", i),
+ 1,
+ partition,
+ iceberg.EntryContentEqDeletes,
+ int64(i+1),
+ []int{1},
+ map[int]int64{1: 1},
+ map[int]int64{1: 0},
+ map[int]int64{1: 0},
+ lowerBounds[group],
+ upperBounds[group],
+ )
+ }
+
+ dataEntries := make([]iceberg.ManifestEntry, dataFileCount)
+ for i := range dataEntries {
+ group := i / filesPerGroup
+ dataEntries[i] = newEqualityDeleteMetricsTestEntry(
+ fmt.Sprintf("data-%d.parquet", i),
+ 1,
+ partition,
+ iceberg.EntryContentData,
+ 0,
+ nil,
+ map[int]int64{1: 1},
+ map[int]int64{1: 0},
+ map[int]int64{1: 0},
+ lowerBounds[group],
+ upperBounds[group],
+ )
+ }
+
+ idx, err := buildEqualityDeleteIndex(
+ deleteEntries, equalityDeleteIndexTestSpecs(), schema)
+ if err != nil {
+ b.Fatal(err)
+ }
+ key, err := newEqualityDeletePartitionKey(1, partition)
+ if err != nil {
+ b.Fatal(err)
+ }
+ entries := idx.byPartition[key]
+ rangeIndex := idx.rangeIndexesByPartition[key]
+ if rangeIndex == nil {
+ b.Fatal("range index was not built")
+ }
+
+ // This benchmark isolates warm lookups; construction is measured
separately.
+ rangeIndex.once.Do(rangeIndex.build)
+ benchmark := func(b *testing.B, indexed bool, dataEntries
[]iceberg.ManifestEntry) {
+ b.ReportAllocs()
+ matched := 0
+ b.ResetTimer()
+ for range b.N {
+ matched = 0
+ for _, dataEntry := range dataEntries {
+ stats :=
equalityDeleteDataFileStatsFor(dataEntry.DataFile())
+ var files []iceberg.DataFile
+ if indexed {
+ files =
appendEqualityDeletesAfterIndexed(
+ nil, entries,
dataEntry.SequenceNum(), &stats, rangeIndex)
+ } else {
+ files =
appendEqualityDeletesAfterIndexed(
+ nil, entries,
dataEntry.SequenceNum(), &stats, nil)
+ }
+ matched += len(files)
+ }
+ equalityDeleteBenchmarkSink = matched
+ }
+ b.StopTimer()
+ b.ReportMetric(float64(matched)/float64(len(dataEntries)),
"attached_deletes_per_data_file")
+ }
+
+ b.Run("linear", func(b *testing.B) {
+ benchmark(b, false, dataEntries)
+ })
+ b.Run("range_index", func(b *testing.B) {
+ benchmark(b, true, dataEntries)
+ })
+
+ // A broad data range exercises the cost of abandoning an unselective
index.
+ lower, upper := equalityDeleteMetricsTestBounds(b, int32(0),
int32(deleteFileCount))
+ unselectiveEntries :=
[]iceberg.ManifestEntry{newEqualityDeleteMetricsTestEntry(
+ "unselective.parquet", 1, partition, iceberg.EntryContentData,
0, nil,
+ map[int]int64{1: 1}, map[int]int64{1: 0}, map[int]int64{1: 0},
lower, upper,
+ )}
+ b.Run("unselective_linear", func(b *testing.B) {
+ benchmark(b, false, unselectiveEntries)
+ })
+ b.Run("unselective_range_index", func(b *testing.B) {
+ benchmark(b, true, unselectiveEntries)
+ })
+}
