laskoviymishka commented on code in PR #2148:
URL: https://github.com/apache/iceberg-go/pull/2148#discussion_r4232280600
##########
table/equality_delete_index.go:
##########
@@ -547,6 +560,48 @@ func appendEqualityDeletesAfter(
start := sort.Search(len(entries), func(i int) bool {
return entries[i].entry.SequenceNum() > dataSeqNum
})
+
+ return appendEqualityDeletesFrom(out, entries, start, dataStats)
+}
+
+func appendEqualityDeletesAfterIndexed(
Review Comment:
With the three-way split, `appendEqualityDeletesAfter` is now only reached
from the test and benchmark; production goes through `...Indexed`. A reference
implementation that lives in production code but is only exercised by tests
tends to drift from the real path over time, so I'd either move it next to the
test as an explicit linear-reference helper or rename it so its role is
obvious. The `sort.Search` for `start` is also copy-pasted across `...After`
and `...Indexed`, worth pulling into one `firstEntryAfter(entries, seq)`.
##########
table/equality_delete_index.go:
##########
@@ -497,8 +499,15 @@ func buildEqualityDeleteIndex(
})
}
sortBySequence(idx.global)
- for _, partitionEntries := range idx.byPartition {
+ idx.globalRangeIndex = newEqualityDeleteRangeIndex(idx.global)
Review Comment:
We build the global index and one per partition eagerly at construction
time, so every scan pays the O(n log n) build plus the per-node copies of
`lower`/`upper`/`maxUpper` even for partitions no data file ever probes. For a
table with many partitions over the 256 threshold that's a lot of allocation
that may never earn its keep. I'd either build lazily (`sync.Once` on the first
qualifying lookup) or, if eager keeps the code simpler, drop a comment owning
the tradeoff.
While we're here, can we confirm `equalityDeleteMetricValue` is cheap to
copy and isn't holding onto byte slices? Three per node adds up if it is.
##########
table/equality_delete_range_index.go:
##########
@@ -0,0 +1,284 @@
+// 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"
+)
+
+const (
+ equalityDeleteRangeIndexMinEntries = 256
+ equalityDeleteRangeIndexMinGroupEntries = 64
+ equalityDeleteRangeIndexMinLookupEntries = 128
+)
+
+type equalityDeleteRangeIndex struct {
+ 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
+ }
+
+ 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))
+ }
+
+ if len(groups) == 0 {
+ return nil
+ }
+
+ slices.Sort(fallback)
+
+ return &equalityDeleteRangeIndex{
+ groups: groups,
+ fallback: fallback,
+ }
+}
+
+func equalityDeleteRangeIndexField(
Review Comment:
The whole correctness story rests on this gate staying a strict subset of
`equalityDeleteCanContainData` (the index may only prune entries the linear
filter would also prune), and that invariant is completely implicit right now.
I'd add a comment here (and on the overlap checks in `appendNodeCandidates`)
stating it, so that a future loosening of `equalityDeleteCanContainData` gets
mirrored here or breaks loudly instead of silently resurrecting deleted rows.
The randomized differential test below would enforce it mechanically, but the
comment is the cheap floor.
##########
table/equality_delete_range_index.go:
##########
@@ -0,0 +1,284 @@
+// 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"
+)
+
+const (
+ equalityDeleteRangeIndexMinEntries = 256
Review Comment:
These three thresholds plus the `remaining/2` bail-out in `candidates` are
doing real work and they interact in a way the names don't show: 256 with a 128
lookup floor means a partition is only ever indexed when more than half its
entries are sequence-applicable, and `remaining/2` then lands the candidate
limit at 64, exactly `MinGroupEntries`. Could we add a sentence per constant on
where the number comes from, and one benchmark point around the 256 to 512
boundary showing it doesn't regress just over the threshold? wdyt
##########
table/equality_delete_range_index.go:
##########
@@ -0,0 +1,284 @@
+// 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"
+)
+
+const (
+ equalityDeleteRangeIndexMinEntries = 256
+ equalityDeleteRangeIndexMinGroupEntries = 64
+ equalityDeleteRangeIndexMinLookupEntries = 128
+)
+
+type equalityDeleteRangeIndex struct {
+ 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
+ }
+
+ 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))
+ }
+
+ if len(groups) == 0 {
+ return nil
+ }
+
+ slices.Sort(fallback)
+
+ return &equalityDeleteRangeIndex{
+ groups: groups,
+ fallback: fallback,
+ }
+}
+
+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
+ }
+
+ return field, true
+ }
+
+ return nil, false
+}
+
+func newEqualityDeleteRangeGroup(
+ field equalityDeleteFieldMetrics,
+ items []equalityDeleteRangeItem,
+) equalityDeleteRangeGroup {
+ slices.SortStableFunc(items, func(a, b equalityDeleteRangeItem) int {
Review Comment:
The comparator already breaks ties on `entryIndex`, which is unique, so the
order is total and stability buys nothing. `slices.SortFunc` is the right call
here.
##########
table/equality_delete_range_index_test.go:
##########
@@ -0,0 +1,222 @@
+// 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"
+)
+
+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
+ }{
+ {name: "selective middle", sequence: 0, lower: 2000, upper:
2099},
+ {name: "selective with sequence pruning", sequence: 200, lower:
4000, upper: 4099},
+ {name: "no ranged overlap keeps fallback", sequence: 0, lower:
9000, upper: 9100},
+ {name: "just below fallback threshold", sequence: 0, lower: 0,
upper: 2454},
Review Comment:
The 2454/2464 bounds encode the half-of-remaining fallback threshold
implicitly, via arithmetic on `i*10`, the 31-stride nil bounds, and
`remaining/2`. If any constant or the dataset layout shifts, these two cases
quietly stop testing the boundary they're named for and nobody notices. I'd
derive the bounds from the computed limit, or assert directly on `candidates()`
`ok` at limit-1 vs limit.
##########
table/equality_delete_range_index.go:
##########
@@ -0,0 +1,284 @@
+// 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"
+)
+
+const (
+ equalityDeleteRangeIndexMinEntries = 256
+ equalityDeleteRangeIndexMinGroupEntries = 64
+ equalityDeleteRangeIndexMinLookupEntries = 128
+)
+
+type equalityDeleteRangeIndex struct {
+ 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
+ }
+
+ 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))
+ }
+
+ if len(groups) == 0 {
+ return nil
+ }
+
+ slices.Sort(fallback)
+
+ return &equalityDeleteRangeIndex{
+ groups: groups,
+ fallback: fallback,
+ }
+}
+
+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
+ }
+
+ return field, true
+ }
+
+ return nil, false
+}
+
+func newEqualityDeleteRangeGroup(
+ field equalityDeleteFieldMetrics,
+ items []equalityDeleteRangeItem,
+) equalityDeleteRangeGroup {
+ slices.SortStableFunc(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, 0, len(items)),
Review Comment:
The `&group.nodes[childIndex]` reads in `build` are only safe because this
slice is preallocated to exactly `len(items)`, so the appends never reallocate
and move the backing array out from under the pointers. That's correct today
but invisible: a future tweak to the preallocation turns it into a
use-after-realloc. Worth a one-line comment pinning the invariant, or
preallocate with length and assign by index instead of appending.
##########
table/equality_delete_range_index_test.go:
##########
@@ -0,0 +1,222 @@
+// 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"
+)
+
+func TestEqualityDeleteRangeIndexMatchesLinearReference(t *testing.T) {
Review Comment:
This is where I'd want more before we merge. The index is a
correctness-critical fast path (a wrong prune drops a delete and resurrects
deleted rows), and right now the only thing between that and a release is
eleven hand-picked cases on a single shape: one nullable int32 field, one
partition, all-zero null counts, every 31st entry unbounded.
The branches that actually decide whether a field is safe to index are the
ones not exercised: float/double with NaN counts (missing, nonzero, and the
"unknown so abort" path), optional fields with nonzero null counts, multi-field
keys where the first field is unusable and the second carries the index, more
than one group plus the fallback merge, sub-`MinGroupEntries` demotion, and the
global index (`forDataFile` only hits the partitioned path here). A bug in any
of those is green on CI today.
I'd add a seeded randomized differential test: generate N deletes with
random fields/bounds/sequences/null+NaN counts/nil bounds and M data files,
assert indexed output equals the linear reference on every query with sizes
straddling each threshold, and include at least one two-field case and one that
goes through both the partition and the global index. That one test covers more
than the eleven cases combined and is the thing that lets this land with
confidence.
##########
table/equality_delete_range_index_test.go:
##########
@@ -0,0 +1,222 @@
+// 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"
+)
+
+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
+ }{
+ {name: "selective middle", sequence: 0, lower: 2000, upper:
2099},
+ {name: "selective with sequence pruning", sequence: 200, lower:
4000, upper: 4099},
+ {name: "no ranged overlap keeps fallback", sequence: 0, lower:
9000, upper: 9100},
+ {name: "just below fallback threshold", sequence: 0, lower: 0,
upper: 2454},
+ {name: "at fallback threshold", sequence: 0, lower: 0, upper:
2464},
+ {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},
+ {name: "equal upper endpoint overlaps", sequence: 0, lower:
2004, upper: 2004},
+ }
+
+ 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 := appendEqualityDeletesAfter(
+ nil, entries, dataEntry.SequenceNum(), &stats)
+ assert.Equal(t, equalityDeleteMetricPaths(want),
equalityDeleteMetricPaths(got))
Review Comment:
This asserts indexed output equals linear output, which is necessary but
can't tell us the index actually engaged: a regression that made `candidates`
always return `ok=false` (everything falls back to linear) would pass all
eleven cases. For the selective cases I'd call `rangeIndex.candidates` directly
and assert `ok` with `len(candidates) < remaining`, and assert `!ok` on the
unselective ones, so the test fails if the fast path silently stops firing.
##########
table/equality_delete_range_index.go:
##########
@@ -0,0 +1,284 @@
+// 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"
+)
+
+const (
+ equalityDeleteRangeIndexMinEntries = 256
+ equalityDeleteRangeIndexMinGroupEntries = 64
+ equalityDeleteRangeIndexMinLookupEntries = 128
+)
+
+type equalityDeleteRangeIndex struct {
+ 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
+ }
+
+ 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))
+ }
+
+ if len(groups) == 0 {
+ return nil
+ }
+
+ slices.Sort(fallback)
+
+ return &equalityDeleteRangeIndex{
+ groups: groups,
+ fallback: fallback,
+ }
+}
+
+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
+ }
+
+ return field, true
Review Comment:
We index on the first usable field only. For a composite equality key where
the first field is low-cardinality and the second is the selective one, every
lookup collects most entries as candidates, trips the `remaining/2` bail-out,
and falls back to linear with the index build cost already paid. Not a blocker,
but worth a comment noting the limitation, or picking the field by bound spread
/ selectivity if that's cheap to do.
--
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]
