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 9fb8cb632 perf(table): fast-path IN metrics pruning with extrema
(#2117)
9fb8cb632 is described below
commit 9fb8cb632964921df5d6ed1c1bd66d4087aca3d8
Author: Minh Vu <[email protected]>
AuthorDate: Tue Oct 13 13:30:13 2026 +0200
perf(table): fast-path IN metrics pruning with extrema (#2117)
* perf(table): fast-path IN metrics pruning with extrema
* test(table): verify IN benchmark pruning results
* perf(table): reuse literal comparator for IN extrema
* perf(table): resolve IN extrema comparator once
* perf(table): straighten IN extrema pruning
* docs(expr): update extrema visitor contract
* test(table): cover IN extrema dispatch and types
* bench(table): label IN extrema baselines
* test(expr): keep geo IN off ordering extrema path
* test(table): gofmt IN extrema coverage
* perf(table): reuse IN extrema comparator
* perf(table): reuse manifest IN extrema comparator
* test(expr): prove IN extrema borrowed dispatch
* perf(table): avoid comparator closure allocation for IN extrema
* test(table): cover IN extrema dispatch and overlap
* test: drop redundant bound extrema dispatch coverage
* refactor(table): straighten IN bound evaluation
* style: gofmt bound extrema test
* style: gofmt IN extrema tests
* perf(table): prune oversized IN sets by extrema
* test(table): pin IN extrema evaluator dispatch
* test: drop duplicate geo extrema coverage
* fix(table): preserve IN slow-path pruning and cover extrema dispatch
* style(table): gofmt IN extrema regression tests
* test(table): verify IN extrema fast paths with valid bounds
Check member access with real bound extrema across five literal types and
remove a condition dominated by the oversized-set return.
Signed-off-by: Minh Vu <[email protected]>
* test(table): pin IN extrema dispatch and comparator coverage
Preserve member-scan behavior, cover supplied-extrema sentinels,
verify comparator types, and document oversized and geo fallbacks.
Signed-off-by: Minh Vu <[email protected]>
* style(table): gofmt IN extrema comparator coverage
* test(table): benchmark IN pruning across literal types
Signed-off-by: Codex <[email protected]>
Signed-off-by: Minh Vu <[email protected]>
* test(table): guard oversized IN overlap from member scans
---------
Signed-off-by: Minh Vu <[email protected]>
Signed-off-by: Codex <[email protected]>
---
bound_set_extrema_internal_test.go | 19 +
table/evaluators.go | 125 ++++++-
table/in_predicate_extrema_bench_test.go | 116 ++++++
table/in_predicate_extrema_test.go | 586 +++++++++++++++++++++++++++++++
visitors.go | 4 +-
5 files changed, 836 insertions(+), 14 deletions(-)
diff --git a/bound_set_extrema_internal_test.go
b/bound_set_extrema_internal_test.go
index 12908dfc6..6bdfb3355 100644
--- a/bound_set_extrema_internal_test.go
+++ b/bound_set_extrema_internal_test.go
@@ -88,3 +88,22 @@ func TestVisitBoundPredicateRefPassesInExtrema(t *testing.T)
{
assert.Nil(t, visitor.minLit)
assert.Nil(t, visitor.maxLit)
}
+
+func TestVisitBoundPredicateRefGeoSkipsExtrema(t *testing.T) {
+ geometry := GeometryType{}
+ first, err := LiteralFromBytes(geometry, []byte("1234567890abcdef"))
+ require.NoError(t, err)
+ second, err := LiteralFromBytes(geometry, []byte("fedcba9876543210"))
+ require.NoError(t, err)
+
+ schema := NewSchema(1, NestedField{ID: 1, Name: "value", Type:
geometry})
+ bound, err := BindExpr(schema, SetPredicate(OpIn, Reference("value"),
[]Literal{first, second}), true)
+ require.NoError(t, err)
+ pred, ok := bound.(BoundPredicate)
+ require.True(t, ok)
+
+ visitor := &boundSetExtremaVisitVisitor{boundSetVisitVisitor:
boundSetVisitVisitor{needle: first}}
+ assert.True(t, VisitBoundPredicateRef(pred, visitor,
internal.BoundPredicateRef{}))
+ assert.Nil(t, visitor.minLit)
+ assert.Nil(t, visitor.maxLit)
+}
diff --git a/table/evaluators.go b/table/evaluators.go
index 0e3e61967..1cf077d38 100644
--- a/table/evaluators.go
+++ b/table/evaluators.go
@@ -178,6 +178,9 @@ func (m *manifestEvalVisitor) VisitIn(term
iceberg.BoundTerm, literals iceberg.S
return m.visitIn(term, literals, nil, nil)
}
+// VisitInWithExtrema expects minLit and maxLit to either both be the actual
+// extrema of literals or both be nil. A partial pair is treated as unavailable
+// and follows the same path as VisitIn.
func (m *manifestEvalVisitor) VisitInWithExtrema(term iceberg.BoundTerm,
literals iceberg.Set[iceberg.Literal], minLit, maxLit iceberg.Literal) bool {
return m.visitIn(term, literals, minLit, maxLit)
}
@@ -185,6 +188,7 @@ func (m *manifestEvalVisitor) VisitInWithExtrema(term
iceberg.BoundTerm, literal
func (m *manifestEvalVisitor) visitIn(term iceberg.BoundTerm, literals
iceberg.Set[iceberg.Literal], minLit, maxLit iceberg.Literal) bool {
pos := term.Ref().Pos()
field := m.partitionFields[pos]
+ hasExtrema := minLit != nil && maxLit != nil
if field.LowerBound == nil {
return rowsCannotMatch
@@ -195,8 +199,10 @@ func (m *manifestEvalVisitor) visitIn(term
iceberg.BoundTerm, literals iceberg.S
panic(err)
}
- if maxLit != nil {
- if getCmpLiteral(lower)(lower, maxLit) > 0 {
+ if hasExtrema {
+ // Like data-file metrics pruning, extrema can reject oversized
IN
+ // sets before we apply the per-member scan limit.
+ if compareBoundLiterals(lower, maxLit) > 0 {
return rowsCannotMatch
}
} else {
@@ -214,8 +220,8 @@ func (m *manifestEvalVisitor) visitIn(term
iceberg.BoundTerm, literals iceberg.S
panic(err)
}
- if minLit != nil {
- if getCmpLiteral(upper)(upper, minLit) < 0 {
+ if hasExtrema {
+ if compareBoundLiterals(upper, minLit) < 0 {
return rowsCannotMatch
}
} else if allBoundCheck(upper, literals, -1) {
@@ -332,6 +338,54 @@ func getCmpLiteral(boundary iceberg.Literal)
func(iceberg.Literal, iceberg.Liter
panic(iceberg.ErrType)
}
+// compareLiteralValues requires the same concrete literal type on both sides,
+// as guaranteed for bound predicates. A mismatched type assertion will panic.
+func compareLiteralValues[T iceberg.LiteralType](left, right iceberg.Literal)
int {
+ leftValue := left.(iceberg.TypedLiteral[T])
+ rightValue := right.(iceberg.TypedLiteral[T])
+
+ return leftValue.Comparator()(leftValue.Value(), rightValue.Value())
+}
+
+// Keep this dispatch aligned with getCmpLiteral, removeBoundCheck, and
+// allBoundCheck when adding a new literal type.
+func compareBoundLiterals(left, right iceberg.Literal) int {
+ switch left.(type) {
+ case iceberg.TypedLiteral[bool]:
+ return compareLiteralValues[bool](left, right)
+ case iceberg.TypedLiteral[int32]:
+ return compareLiteralValues[int32](left, right)
+ case iceberg.TypedLiteral[int64]:
+ return compareLiteralValues[int64](left, right)
+ case iceberg.TypedLiteral[float32]:
+ return compareLiteralValues[float32](left, right)
+ case iceberg.TypedLiteral[float64]:
+ return compareLiteralValues[float64](left, right)
+ case iceberg.TypedLiteral[iceberg.Date]:
+ return compareLiteralValues[iceberg.Date](left, right)
+ case iceberg.TypedLiteral[iceberg.Time]:
+ return compareLiteralValues[iceberg.Time](left, right)
+ case iceberg.TypedLiteral[iceberg.Timestamp]:
+ return compareLiteralValues[iceberg.Timestamp](left, right)
+ case iceberg.TypedLiteral[iceberg.TimestampNano]:
+ return compareLiteralValues[iceberg.TimestampNano](left, right)
+ case iceberg.TypedLiteral[[]byte]:
+ return compareLiteralValues[[]byte](left, right)
+ case iceberg.TypedLiteral[string]:
+ return compareLiteralValues[string](left, right)
+ case iceberg.TypedLiteral[uuid.UUID]:
+ return compareLiteralValues[uuid.UUID](left, right)
+ case iceberg.TypedLiteral[iceberg.Decimal]:
+ return compareLiteralValues[iceberg.Decimal](left, right)
+ case iceberg.GeoLiteral:
+ // Bound IN dispatch never supplies geo extrema because geo has
no
+ // ordering. Keep this guard for direct or incorrectly routed
calls.
+ panic(fmt.Errorf("%w: geometry/geography has no ordering,
cannot compare %s bounds",
+ iceberg.ErrType, left.Type()))
+ }
+ panic(iceberg.ErrType)
+}
+
func (m *manifestEvalVisitor) VisitEqual(term iceberg.BoundTerm, lit
iceberg.Literal) bool {
pos := term.Ref().Pos()
field := m.partitionFields[pos]
@@ -1160,34 +1214,81 @@ func (m *inclusiveMetricsEval)
VisitNotEqual(iceberg.BoundTerm, iceberg.Literal)
}
func (m *inclusiveMetricsEval) VisitIn(t iceberg.BoundTerm, s
iceberg.Set[iceberg.Literal]) bool {
+ return m.visitIn(t, s, nil, nil)
+}
+
+// VisitInWithExtrema expects minLit and maxLit to either both be the actual
+// extrema of s or both be nil. A partial pair is treated as unavailable and
+// follows the same path as VisitIn.
+func (m *inclusiveMetricsEval) VisitInWithExtrema(
+ t iceberg.BoundTerm, s iceberg.Set[iceberg.Literal], minLit, maxLit
iceberg.Literal,
+) bool {
+ return m.visitIn(t, s, minLit, maxLit)
+}
+
+func (m *inclusiveMetricsEval) visitIn(
+ t iceberg.BoundTerm, s iceberg.Set[iceberg.Literal], minLit, maxLit
iceberg.Literal,
+) bool {
fieldID := t.Ref().Field().ID
if m.containsNullsOnly(fieldID) || m.containsNansOnly(fieldID) {
return rowsCannotMatch
}
- if s.Len() > inPredicateLimit {
- // skip evaluating the predicate if the number of values is too
big
+ hasExtrema := minLit != nil && maxLit != nil
+ oversized := s.Len() > inPredicateLimit
+ if oversized && !hasExtrema {
+ // Skip evaluating the predicate if the number of values is too
big.
return rowsMightMatch
}
- values := s.Members()
- if lowerBound, ok := m.boundFor(t, m.lowerBounds[fieldID]); ok {
+ var values []iceberg.Literal
+ lowerBound, hasLowerBound := m.boundFor(t, m.lowerBounds[fieldID])
+ if hasLowerBound {
if m.isNan(lowerBound) {
return rowsMightMatch
}
-
- values = removeBoundCheck(lowerBound, values, 1)
- if len(values) == 0 {
+ if hasExtrema && compareBoundLiterals(lowerBound, maxLit) > 0 {
+ // Short-circuit before decoding an unnecessary
(possibly invalid) upper bound.
return rowsCannotMatch
}
+ if !hasExtrema {
+ // The slow path must finish the lower scan before
reading the upper bound.
+ values = removeBoundCheck(lowerBound, s.Members(), 1)
+ if len(values) == 0 {
+ return rowsCannotMatch
+ }
+ }
}
- if upperBound, ok := m.boundFor(t, m.upperBounds[fieldID]); ok {
+ upperBound, hasUpperBound := m.boundFor(t, m.upperBounds[fieldID])
+ if hasUpperBound {
if m.isNan(upperBound) {
return rowsMightMatch
}
+ if hasExtrema && compareBoundLiterals(upperBound, minLit) < 0 {
+ return rowsCannotMatch
+ }
+ }
+
+ if oversized {
+ // Both metrics evaluators prune disjoint oversized IN sets
with O(1)
+ // extrema comparisons before limiting the member scan. Unlike
Java's
+ // inclusive evaluator, this can reject files past the IN size
limit.
+ return rowsMightMatch
+ }
+
+ if hasExtrema || !hasLowerBound {
+ values = s.Members()
+ }
+ if hasExtrema && hasLowerBound {
+ values = removeBoundCheck(lowerBound, values, 1)
+ if len(values) == 0 {
+ return rowsCannotMatch
+ }
+ }
+ if hasUpperBound {
values = removeBoundCheck(upperBound, values, -1)
if len(values) == 0 {
return rowsCannotMatch
diff --git a/table/in_predicate_extrema_bench_test.go
b/table/in_predicate_extrema_bench_test.go
index 03ce9ab9b..81a3c6a4f 100644
--- a/table/in_predicate_extrema_bench_test.go
+++ b/table/in_predicate_extrema_bench_test.go
@@ -18,6 +18,7 @@
package table
import (
+ "fmt"
"strconv"
"testing"
@@ -116,3 +117,118 @@ func benchmarkManifestIn[T iceberg.LiteralType](b
*testing.B, values []T, lower,
manifestInPredicateBenchmarkSink = matched
}
}
+
+func BenchmarkInclusiveMetricsEvalInPredicate(b *testing.B) {
+ for _, literalCount := range []int{10, 50, 200} {
+ for _, typ := range []struct {
+ name string
+ typ iceberg.Type
+ literal func(int) iceberg.Literal
+ }{
+ {"int32", iceberg.PrimitiveTypes.Int32, func(value int)
iceberg.Literal {
+ return iceberg.NewLiteral(int32(value))
+ }},
+ {"string", iceberg.PrimitiveTypes.String, func(value
int) iceberg.Literal {
+ // Keep lexical and numeric order identical,
including the below-set bound.
+ return
iceberg.NewLiteral(fmt.Sprintf("value-%06d", value+1000))
+ }},
+ {"decimal", iceberg.DecimalTypeOf(12, 2), func(value
int) iceberg.Literal {
+ return iceberg.NewLiteral(iceberg.Decimal{Val:
decimal128.FromI64(int64(value)), Scale: 2})
+ }},
+ } {
+ b.Run(typ.name+"/in="+strconv.Itoa(literalCount),
func(b *testing.B) {
+ values := make([]iceberg.Literal, literalCount)
+ for i := range values {
+ values[i] = typ.literal(i * 2)
+ }
+
+ below := typ.literal(-2)
+ above := typ.literal(literalCount * 2)
+ gap := typ.literal((literalCount/2)*2 + 1)
+ // Disjoint bounds exercise the extrema
short-circuit. Overlap,
+ // sparse gaps and missing bounds catch
regressions on the fallback path.
+ for _, bounds := range []struct {
+ name string
+ lower, upper iceberg.Literal
+ wantMatch bool
+ }{
+ {name: "below-set", lower: below,
upper: below},
+ {name: "above-set", lower: above,
upper: above},
+ {name: "overlap", lower: values[0],
upper: values[len(values)-1], wantMatch: true},
+ {name: "sparse-gap", lower: gap, upper:
gap},
+ {name: "missing-bounds", wantMatch:
true},
+ } {
+ b.Run(bounds.name, func(b *testing.B) {
+ benchmarkInclusiveMetricsIn(b,
typ.typ, values, bounds.lower, bounds.upper, bounds.wantMatch)
+ })
+ }
+ })
+ }
+ }
+}
+
+func benchmarkInclusiveMetricsIn(b *testing.B, typ iceberg.Type, values
[]iceberg.Literal, lower, upper iceberg.Literal, wantMatch bool) {
+ b.Helper()
+
+ schema := iceberg.NewSchema(1, iceberg.NestedField{ID: 1, Name:
"value", Type: typ})
+ eval, err := newInclusiveMetricsEvaluator(
+ schema, iceberg.SetPredicate(iceberg.OpIn,
iceberg.Reference("value"), values), true, true,
+ )
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ builder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentData,
"file.parquet", iceberg.ParquetFile,
+ nil, nil, nil, 10, 100,
+ )
+ if err != nil {
+ b.Fatal(err)
+ }
+ builder.
+ ValueCounts(map[int]int64{1: 10}).
+ NullValueCounts(map[int]int64{1: 0}).
+ NaNValueCounts(map[int]int64{1: 0})
+ if lower != nil {
+ encoded, err := lower.MarshalBinary()
+ if err != nil {
+ b.Fatal(err)
+ }
+ builder.LowerBoundValues(map[int][]byte{1: encoded})
+ }
+ if upper != nil {
+ encoded, err := upper.MarshalBinary()
+ if err != nil {
+ b.Fatal(err)
+ }
+ builder.UpperBoundValues(map[int][]byte{1: encoded})
+ }
+ file := builder.Build()
+ files := make([]iceberg.DataFile, 10_000)
+ for i := range files {
+ files[i] = file
+ }
+
+ b.ReportAllocs()
+ for b.Loop() {
+ matchedFiles := 0
+ for _, f := range files {
+ matched, err := eval(f)
+ if err != nil {
+ b.Fatal(err)
+ }
+ if matched {
+ matchedFiles++
+ }
+ }
+ wantFiles := 0
+ if wantMatch {
+ wantFiles = len(files)
+ }
+ if matchedFiles != wantFiles {
+ b.Fatalf("matched %d files, want %d", matchedFiles,
wantFiles)
+ }
+ }
+ b.ReportMetric(float64(len(files)), "files/op")
+ b.ReportMetric(float64(len(values)), "literals/file")
+}
diff --git a/table/in_predicate_extrema_test.go
b/table/in_predicate_extrema_test.go
index 81f6cb821..663293eda 100644
--- a/table/in_predicate_extrema_test.go
+++ b/table/in_predicate_extrema_test.go
@@ -18,14 +18,56 @@
package table
import (
+ "math"
"testing"
"github.com/apache/arrow-go/v18/arrow/decimal128"
+ "github.com/apache/arrow-go/v18/parquet/variant"
"github.com/apache/iceberg-go"
+ "github.com/google/uuid"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
+func TestCompareBoundLiteralsMatchesGetCmpLiteral(t *testing.T) {
+ decimal := func(value int64) iceberg.Decimal {
+ return iceberg.Decimal{Val: decimal128.FromI64(value), Scale: 2}
+ }
+ // Both dispatchers must recognize every orderable concrete literal
type.
+ tests := []struct {
+ name string
+ lower, upper iceberg.Literal
+ }{
+ {"bool", iceberg.NewLiteral(false), iceberg.NewLiteral(true)},
+ {"int32", iceberg.NewLiteral(int32(1)),
iceberg.NewLiteral(int32(2))},
+ {"int64", iceberg.NewLiteral(int64(1)),
iceberg.NewLiteral(int64(2))},
+ {"float32", iceberg.NewLiteral(float32(1)),
iceberg.NewLiteral(float32(2))},
+ {"float64", iceberg.NewLiteral(float64(1)),
iceberg.NewLiteral(float64(2))},
+ {"date", iceberg.NewLiteral(iceberg.Date(1)),
iceberg.NewLiteral(iceberg.Date(2))},
+ {"time", iceberg.NewLiteral(iceberg.Time(1)),
iceberg.NewLiteral(iceberg.Time(2))},
+ {"timestamp", iceberg.NewLiteral(iceberg.Timestamp(1)),
iceberg.NewLiteral(iceberg.Timestamp(2))},
+ {"timestamp nanos",
iceberg.NewLiteral(iceberg.TimestampNano(1)),
iceberg.NewLiteral(iceberg.TimestampNano(2))},
+ {"binary", iceberg.NewLiteral([]byte("a")),
iceberg.NewLiteral([]byte("b"))},
+ {"fixed", iceberg.FixedLiteral("a"), iceberg.FixedLiteral("b")},
+ {"string", iceberg.NewLiteral("a"), iceberg.NewLiteral("b")},
+ {"uuid", iceberg.NewLiteral(uuid.Nil),
iceberg.NewLiteral(uuid.MustParse("00000000-0000-0000-0000-000000000001"))},
+ {"decimal", iceberg.NewLiteral(decimal(100)),
iceberg.NewLiteral(decimal(200))},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ compare := getCmpLiteral(tt.lower)
+ for _, pair := range [][2]iceberg.Literal{
+ {tt.lower, tt.upper},
+ {tt.upper, tt.lower},
+ {tt.lower, tt.lower},
+ } {
+ assert.Equal(t, compare(pair[0], pair[1]),
compareBoundLiterals(pair[0], pair[1]))
+ }
+ })
+ }
+}
+
func TestManifestEvaluatorInPredicateExtrema(t *testing.T) {
decimal := func(value int64) iceberg.Decimal {
return iceberg.Decimal{Val: decimal128.FromI64(value), Scale: 2}
@@ -139,6 +181,550 @@ func TestManifestEvaluatorInPredicateExtrema(t
*testing.T) {
}
}
+func TestInclusiveMetricsEvaluatorInPredicateExtrema(t *testing.T) {
+ encode := func(value int32) []byte {
+ encoded, err := iceberg.NewLiteral(value).MarshalBinary()
+ require.NoError(t, err)
+
+ return encoded
+ }
+
+ schema := iceberg.NewSchema(1, iceberg.NestedField{ID: 1, Name:
"value", Type: iceberg.PrimitiveTypes.Int32})
+ expr := iceberg.IsIn(iceberg.Reference("value"), int32(1), int32(100))
+ eval, err := newInclusiveMetricsEvaluator(schema, expr, true, true)
+ require.NoError(t, err)
+
+ tests := []struct {
+ name string
+ lowerBound []byte
+ upperBound []byte
+ want bool
+ }{
+ {name: "lower bound above set", lowerBound: encode(101),
upperBound: encode(200), want: false},
+ {name: "upper bound below set", lowerBound: encode(-10),
upperBound: encode(0), want: false},
+ {name: "sparse set has no member in range", lowerBound:
encode(40), upperBound: encode(60), want: false},
+ {name: "overlapping range contains member", lowerBound:
encode(50), upperBound: encode(100), want: true},
+ {name: "upper bound equals minimum literal", upperBound:
encode(1), want: true},
+ {name: "lower bound equals maximum literal", lowerBound:
encode(100), want: true},
+ {name: "missing bounds", want: true},
+ {name: "upper bound below set without lower bound", upperBound:
encode(0), want: false},
+ {name: "lower bound above set without upper bound", lowerBound:
encode(101), want: false},
+ {name: "disjoint lower bound avoids malformed upper bound",
lowerBound: encode(101), upperBound: []byte{1}, want: false},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ file := inclusiveMetricsInTestFile(t, tt.lowerBound,
tt.upperBound)
+ got, err := eval(file)
+ require.NoError(t, err)
+ assert.Equal(t, tt.want, got)
+ })
+ }
+
+ t.Run("NaN bounds fail open", func(t *testing.T) {
+ encodeFloat := func(value float64) []byte {
+ encoded, err :=
iceberg.NewLiteral(value).MarshalBinary()
+ require.NoError(t, err)
+
+ return encoded
+ }
+ nan := encodeFloat(math.NaN())
+ finiteLower := encodeFloat(0)
+ finiteUpper := encodeFloat(3)
+ floatSchema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 1, Name: "value", Type:
iceberg.PrimitiveTypes.Float64,
+ })
+ floatEval, err := newInclusiveMetricsEvaluator(
+ floatSchema, iceberg.IsIn(iceberg.Reference("value"),
float64(1), float64(2)), true, true,
+ )
+ require.NoError(t, err)
+
+ for _, bounds := range []struct {
+ name string
+ lower, upper []byte
+ }{
+ {name: "both NaN", lower: nan, upper: nan},
+ {name: "NaN lower finite upper", lower: nan, upper:
finiteUpper},
+ {name: "finite lower NaN upper", lower: finiteLower,
upper: nan},
+ } {
+ t.Run(bounds.name, func(t *testing.T) {
+ got, err :=
floatEval(inclusiveMetricsInTestFile(t, bounds.lower, bounds.upper))
+ require.NoError(t, err)
+ assert.True(t, got)
+ })
+ }
+ })
+
+ t.Run("oversized set still uses extrema before member scan limit",
func(t *testing.T) {
+ values := make([]int32, inPredicateLimit+1)
+ for i := range values {
+ values[i] = int32(i)
+ }
+ largeEval, err := newInclusiveMetricsEvaluator(
+ schema, iceberg.IsIn(iceberg.Reference("value"),
values...), true, true,
+ )
+ require.NoError(t, err)
+
+ disjointLower := int32(inPredicateLimit + 1000)
+ got, err := largeEval(inclusiveMetricsInTestFile(
+ t, encode(disjointLower), encode(disjointLower+1),
+ ))
+ require.NoError(t, err)
+ assert.False(t, got)
+
+ got, err = largeEval(inclusiveMetricsInTestFile(t, encode(50),
encode(60)))
+ require.NoError(t, err)
+ assert.True(t, got)
+
+ // An overlapping oversized set must return before scanning its
members.
+ // This guards the size limit even when extrema cannot prune
the file.
+ bound, err := iceberg.BindExpr(schema,
iceberg.IsIn(iceberg.Reference("value"), values...), true)
+ require.NoError(t, err)
+ pred, ok := bound.(iceberg.BoundSetPredicate)
+ require.True(t, ok)
+
+ tracked := &inExtremaTrackingSet{Set: pred.Literals()}
+ visitor := &inclusiveMetricsEval{metricsEvaluator:
metricsEvaluator{
+ valueCounts: map[int]int64{1: 10},
+ nullCounts: map[int]int64{1: 0},
+ nanCounts: map[int]int64{1: 0},
+ lowerBounds: map[int][]byte{1: encode(50)},
+ upperBounds: map[int][]byte{1: encode(60)},
+ }}
+ require.True(t, visitor.VisitInWithExtrema(
+ pred.Term(), tracked, iceberg.NewLiteral(int32(0)),
iceberg.NewLiteral(int32(inPredicateLimit)),
+ ))
+ require.Zero(t, tracked.memberCalls)
+ })
+}
+
+func TestInclusiveMetricsEvaluatorInPredicateExtremaFastPathTypes(t
*testing.T) {
+ decimal := func(value int64) iceberg.Decimal {
+ return iceberg.Decimal{Val: decimal128.FromI64(value), Scale: 2}
+ }
+
+ tests := []struct {
+ name string
+ typ iceberg.Type
+ expr iceberg.BooleanExpression
+ minLit, maxLit iceberg.Literal
+ lower, upper iceberg.Literal
+ want bool
+ }{
+ {
+ name: "string lower disjoint",
+ typ: iceberg.PrimitiveTypes.String,
+ expr: iceberg.IsIn(iceberg.Reference("value"), "a",
"b"),
+ minLit: iceberg.NewLiteral("a"), maxLit:
iceberg.NewLiteral("b"),
+ lower: iceberg.NewLiteral("m"), upper:
iceberg.NewLiteral("z"),
+ },
+ {
+ name: "string overlaps",
+ typ: iceberg.PrimitiveTypes.String,
+ expr: iceberg.IsIn(iceberg.Reference("value"), "a",
"b"),
+ minLit: iceberg.NewLiteral("a"), maxLit:
iceberg.NewLiteral("b"),
+ lower: iceberg.NewLiteral("a"), upper:
iceberg.NewLiteral("a"),
+ want: true,
+ },
+ {
+ name: "decimal upper disjoint",
+ typ: iceberg.DecimalTypeOf(12, 2),
+ expr: iceberg.IsIn(iceberg.Reference("value"),
decimal(100), decimal(200)),
+ minLit: iceberg.NewLiteral(decimal(100)), maxLit:
iceberg.NewLiteral(decimal(200)),
+ lower: iceberg.NewLiteral(decimal(-200)), upper:
iceberg.NewLiteral(decimal(0)),
+ },
+ {
+ name: "decimal overlaps",
+ typ: iceberg.DecimalTypeOf(12, 2),
+ expr: iceberg.IsIn(iceberg.Reference("value"),
decimal(100), decimal(200)),
+ minLit: iceberg.NewLiteral(decimal(100)), maxLit:
iceberg.NewLiteral(decimal(200)),
+ lower: iceberg.NewLiteral(decimal(100)), upper:
iceberg.NewLiteral(decimal(150)),
+ want: true,
+ },
+ {
+ name: "binary lower disjoint",
+ typ: iceberg.PrimitiveTypes.Binary,
+ expr: iceberg.IsIn(iceberg.Reference("value"),
[]byte{1}, []byte{2}),
+ minLit: iceberg.NewLiteral([]byte{1}), maxLit:
iceberg.NewLiteral([]byte{2}),
+ lower: iceberg.NewLiteral([]byte{10}), upper:
iceberg.NewLiteral([]byte{20}),
+ },
+ {
+ name: "binary overlaps",
+ typ: iceberg.PrimitiveTypes.Binary,
+ expr: iceberg.IsIn(iceberg.Reference("value"),
[]byte{1}, []byte{2}),
+ minLit: iceberg.NewLiteral([]byte{1}), maxLit:
iceberg.NewLiteral([]byte{2}),
+ lower: iceberg.NewLiteral([]byte{1}), upper:
iceberg.NewLiteral([]byte{1}),
+ want: true,
+ },
+ {
+ name: "date upper disjoint",
+ typ: iceberg.PrimitiveTypes.Date,
+ expr: iceberg.IsIn(iceberg.Reference("value"),
iceberg.Date(10), iceberg.Date(20)),
+ minLit: iceberg.NewLiteral(iceberg.Date(10)), maxLit:
iceberg.NewLiteral(iceberg.Date(20)),
+ lower: iceberg.NewLiteral(iceberg.Date(-10)), upper:
iceberg.NewLiteral(iceberg.Date(0)),
+ },
+ {
+ name: "date overlaps",
+ typ: iceberg.PrimitiveTypes.Date,
+ expr: iceberg.IsIn(iceberg.Reference("value"),
iceberg.Date(10), iceberg.Date(20)),
+ minLit: iceberg.NewLiteral(iceberg.Date(10)), maxLit:
iceberg.NewLiteral(iceberg.Date(20)),
+ lower: iceberg.NewLiteral(iceberg.Date(10)), upper:
iceberg.NewLiteral(iceberg.Date(15)),
+ want: true,
+ },
+ {
+ name: "timestamp lower disjoint",
+ typ: iceberg.PrimitiveTypes.Timestamp,
+ expr: iceberg.IsIn(iceberg.Reference("value"),
iceberg.Timestamp(10), iceberg.Timestamp(20)),
+ minLit: iceberg.NewLiteral(iceberg.Timestamp(10)),
maxLit: iceberg.NewLiteral(iceberg.Timestamp(20)),
+ lower: iceberg.NewLiteral(iceberg.Timestamp(30)),
upper: iceberg.NewLiteral(iceberg.Timestamp(40)),
+ },
+ {
+ name: "timestamp overlaps",
+ typ: iceberg.PrimitiveTypes.Timestamp,
+ expr: iceberg.IsIn(iceberg.Reference("value"),
iceberg.Timestamp(10), iceberg.Timestamp(20)),
+ minLit: iceberg.NewLiteral(iceberg.Timestamp(10)),
maxLit: iceberg.NewLiteral(iceberg.Timestamp(20)),
+ lower: iceberg.NewLiteral(iceberg.Timestamp(10)),
upper: iceberg.NewLiteral(iceberg.Timestamp(15)),
+ want: true,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ schema := iceberg.NewSchema(1, iceberg.NestedField{ID:
1, Name: "value", Type: tt.typ})
+ bound, err := iceberg.BindExpr(schema, tt.expr, true)
+ require.NoError(t, err)
+ pred, ok := bound.(iceberg.BoundSetPredicate)
+ require.True(t, ok)
+
+ lower, err := tt.lower.MarshalBinary()
+ require.NoError(t, err)
+ upper, err := tt.upper.MarshalBinary()
+ require.NoError(t, err)
+ newVisitor := func() *inclusiveMetricsEval {
+ return &inclusiveMetricsEval{metricsEvaluator:
metricsEvaluator{
+ valueCounts: map[int]int64{1: 10},
+ nullCounts: map[int]int64{1: 0},
+ nanCounts: map[int]int64{1: 0},
+ lowerBounds: map[int][]byte{1: lower},
+ upperBounds: map[int][]byte{1: upper},
+ }}
+ }
+
+ slowSet := &inExtremaTrackingSet{Set: pred.Literals()}
+ fastSet := &inExtremaTrackingSet{Set: pred.Literals()}
+ slow := newVisitor().VisitIn(pred.Term(), slowSet)
+ fast := newVisitor().VisitInWithExtrema(pred.Term(),
fastSet, tt.minLit, tt.maxLit)
+ require.Equal(t, slow, fast)
+ require.Equal(t, tt.want, fast)
+ require.Equal(t, 1, slowSet.memberCalls)
+ if tt.want {
+ // Overlapping ranges must still visit the
exact set members.
+ require.Equal(t, 1, fastSet.memberCalls)
+ } else {
+ require.Zero(t, fastSet.memberCalls)
+ }
+ })
+ }
+}
+
+type inExtremaTrackingSet struct {
+ iceberg.Set[iceberg.Literal]
+ memberCalls int
+}
+
+func (s *inExtremaTrackingSet) Members() []iceberg.Literal {
+ s.memberCalls++
+
+ return s.Set.Members()
+}
+
+func TestInclusiveMetricsEvaluatorInPredicateExtremaUsesFastPath(t *testing.T)
{
+ decimal := func(value int64) iceberg.Decimal {
+ return iceberg.Decimal{Val: decimal128.FromI64(value), Scale: 2}
+ }
+ // Both paths must prune these disjoint files. The extrema path must do
so
+ // without materializing the set members, using the actual bound
extrema.
+ tests := []struct {
+ name string
+ typ iceberg.Type
+ bounds []iceberg.Literal
+ minLit, maxLit iceberg.Literal
+ }{
+ {
+ "string", iceberg.PrimitiveTypes.String,
+ []iceberg.Literal{iceberg.NewLiteral("m"),
iceberg.NewLiteral("n")},
+ iceberg.NewLiteral("a"), iceberg.NewLiteral("b"),
+ },
+ {
+ "decimal", iceberg.DecimalTypeOf(12, 2),
+ []iceberg.Literal{iceberg.NewLiteral(decimal(5000)),
iceberg.NewLiteral(decimal(6000))},
+ iceberg.NewLiteral(decimal(100)),
iceberg.NewLiteral(decimal(200)),
+ },
+ {
+ "binary", iceberg.PrimitiveTypes.Binary,
+ []iceberg.Literal{iceberg.NewLiteral([]byte("m")),
iceberg.NewLiteral([]byte("n"))},
+ iceberg.NewLiteral([]byte("a")),
iceberg.NewLiteral([]byte("b")),
+ },
+ {
+ "date", iceberg.PrimitiveTypes.Date,
+ []iceberg.Literal{iceberg.NewLiteral(iceberg.Date(50)),
iceberg.NewLiteral(iceberg.Date(60))},
+ iceberg.NewLiteral(iceberg.Date(1)),
iceberg.NewLiteral(iceberg.Date(2)),
+ },
+ {
+ "timestamp", iceberg.PrimitiveTypes.Timestamp,
+
[]iceberg.Literal{iceberg.NewLiteral(iceberg.Timestamp(50)),
iceberg.NewLiteral(iceberg.Timestamp(60))},
+ iceberg.NewLiteral(iceberg.Timestamp(1)),
iceberg.NewLiteral(iceberg.Timestamp(2)),
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ schema := iceberg.NewSchema(1, iceberg.NestedField{ID:
1, Name: "value", Type: tt.typ})
+ bound, err := iceberg.BindExpr(schema,
iceberg.SetPredicate(
+ iceberg.OpIn, iceberg.Reference("value"),
[]iceberg.Literal{tt.minLit, tt.maxLit},
+ ), true)
+ require.NoError(t, err)
+ pred, ok := bound.(iceberg.BoundSetPredicate)
+ require.True(t, ok)
+
+ lower, err := tt.bounds[0].MarshalBinary()
+ require.NoError(t, err)
+ upper, err := tt.bounds[1].MarshalBinary()
+ require.NoError(t, err)
+ newVisitor := func() *inclusiveMetricsEval {
+ return &inclusiveMetricsEval{metricsEvaluator:
metricsEvaluator{
+ valueCounts: map[int]int64{1: 10},
+ nullCounts: map[int]int64{1: 0},
+ nanCounts: map[int]int64{1: 0},
+ lowerBounds: map[int][]byte{1: lower},
+ upperBounds: map[int][]byte{1: upper},
+ }}
+ }
+
+ slowSet := &inExtremaTrackingSet{Set: pred.Literals()}
+ require.False(t, newVisitor().VisitIn(pred.Term(),
slowSet))
+ require.Equal(t, 1, slowSet.memberCalls)
+
+ fastSet := &inExtremaTrackingSet{Set: pred.Literals()}
+ require.False(t,
newVisitor().VisitInWithExtrema(pred.Term(), fastSet, tt.minLit, tt.maxLit))
+ require.Zero(t, fastSet.memberCalls)
+ })
+ }
+
+ t.Run("supplied extrema are used, not recomputed", func(t *testing.T) {
+ schema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 1, Name: "value", Type:
iceberg.PrimitiveTypes.Int32,
+ })
+ bound, err := iceberg.BindExpr(
+ schema, iceberg.IsIn(iceberg.Reference("value"),
int32(50), int32(100)), true,
+ )
+ require.NoError(t, err)
+ pred, ok := bound.(iceberg.BoundSetPredicate)
+ require.True(t, ok)
+
+ lower, err := iceberg.NewLiteral(int32(40)).MarshalBinary()
+ require.NoError(t, err)
+ upper, err := iceberg.NewLiteral(int32(60)).MarshalBinary()
+ require.NoError(t, err)
+ visitor := &inclusiveMetricsEval{metricsEvaluator:
metricsEvaluator{
+ valueCounts: map[int]int64{1: 10},
+ nullCounts: map[int]int64{1: 0},
+ nanCounts: map[int]int64{1: 0},
+ lowerBounds: map[int][]byte{1: lower},
+ upperBounds: map[int][]byte{1: upper},
+ }}
+
+ slowSet := &inExtremaTrackingSet{Set: pred.Literals()}
+ require.True(t, visitor.VisitIn(pred.Term(), slowSet))
+ require.Equal(t, 1, slowSet.memberCalls)
+
+ // Binding always supplies real extrema. Deliberately incorrect
values
+ // distinguish the fast-path contract from recomputing or
scanning members.
+ fastSet := &inExtremaTrackingSet{Set: pred.Literals()}
+ require.False(t, visitor.VisitInWithExtrema(
+ pred.Term(), fastSet, iceberg.NewLiteral(int32(1)),
iceberg.NewLiteral(int32(2)),
+ ))
+ require.Zero(t, fastSet.memberCalls)
+ })
+}
+
+func TestInclusiveMetricsEvaluatorInPredicateSlowPathLowerBeforeUpper(t
*testing.T) {
+ tests := []struct {
+ name string
+ typ iceberg.Type
+ members []iceberg.Literal
+ lower iceberg.Literal
+ upper iceberg.Literal
+ malformedUpper bool
+ }{
+ {
+ name: "malformed upper",
+ typ: iceberg.PrimitiveTypes.Int32,
+ members:
[]iceberg.Literal{iceberg.NewLiteral(int32(1)), iceberg.NewLiteral(int32(2))},
+ lower: iceberg.NewLiteral(int32(3)),
+ malformedUpper: true,
+ },
+ {
+ name: "NaN upper",
+ typ: iceberg.PrimitiveTypes.Float64,
+ members:
[]iceberg.Literal{iceberg.NewLiteral(float64(1)),
iceberg.NewLiteral(float64(2))},
+ lower: iceberg.NewLiteral(float64(3)),
+ upper: iceberg.NewLiteral(math.NaN()),
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ schema := iceberg.NewSchema(1, iceberg.NestedField{ID:
1, Name: "value", Type: tt.typ})
+ bound, err := iceberg.BindExpr(schema,
iceberg.SetPredicate(iceberg.OpIn, iceberg.Reference("value"), tt.members),
true)
+ require.NoError(t, err)
+ pred, ok := bound.(iceberg.BoundSetPredicate)
+ require.True(t, ok)
+
+ lower, err := tt.lower.MarshalBinary()
+ require.NoError(t, err)
+ upper := []byte{1}
+ if !tt.malformedUpper {
+ upper, err = tt.upper.MarshalBinary()
+ require.NoError(t, err)
+ }
+ newVisitor := func() *inclusiveMetricsEval {
+ return &inclusiveMetricsEval{metricsEvaluator:
metricsEvaluator{
+ valueCounts: map[int]int64{1: 10},
+ nullCounts: map[int]int64{1: 0},
+ nanCounts: map[int]int64{1: 0},
+ lowerBounds: map[int][]byte{1: lower},
+ upperBounds: map[int][]byte{1: upper},
+ }}
+ }
+
+ // Both slow entry points must prune without even
decoding the upper bound.
+ require.False(t, newVisitor().VisitIn(pred.Term(),
pred.Literals()))
+ require.False(t,
newVisitor().VisitInWithExtrema(pred.Term(), pred.Literals(), tt.lower, nil))
+ require.False(t,
newVisitor().VisitInWithExtrema(pred.Term(), pred.Literals(), nil, tt.lower))
+ })
+ }
+}
+
+func TestInclusiveMetricsEvaluatorInPredicateExtremaPartialNilUsesSlowPath(t
*testing.T) {
+ schema := iceberg.NewSchema(1, iceberg.NestedField{ID: 1, Name:
"value", Type: iceberg.PrimitiveTypes.Int32})
+ bound, err := iceberg.BindExpr(
+ schema, iceberg.IsIn(iceberg.Reference("value"), int32(1),
int32(100)), true,
+ )
+ require.NoError(t, err)
+ pred := bound.(iceberg.BoundSetPredicate)
+ lower, err := iceberg.NewLiteral(int32(40)).MarshalBinary()
+ require.NoError(t, err)
+ upper, err := iceberg.NewLiteral(int32(60)).MarshalBinary()
+ require.NoError(t, err)
+
+ newVisitor := func() *inclusiveMetricsEval {
+ return &inclusiveMetricsEval{metricsEvaluator: metricsEvaluator{
+ valueCounts: map[int]int64{1: 10},
+ nullCounts: map[int]int64{1: 0},
+ nanCounts: map[int]int64{1: 0},
+ lowerBounds: map[int][]byte{1: lower},
+ upperBounds: map[int][]byte{1: upper},
+ }}
+ }
+ slow := newVisitor().VisitIn(pred.Term(), pred.Literals())
+ minLit := iceberg.NewLiteral(int32(1))
+ maxLit := iceberg.NewLiteral(int32(100))
+ require.Equal(t, slow, newVisitor().VisitInWithExtrema(pred.Term(),
pred.Literals(), minLit, nil))
+ require.Equal(t, slow, newVisitor().VisitInWithExtrema(pred.Term(),
pred.Literals(), nil, maxLit))
+}
+
+func TestInclusiveMetricsEvaluatorGeoInUsesFallback(t *testing.T) {
+ geometry := iceberg.GeometryType{}
+ first, err := iceberg.LiteralFromBytes(geometry,
[]byte("1234567890abcdef"))
+ require.NoError(t, err)
+ second, err := iceberg.LiteralFromBytes(geometry,
[]byte("fedcba9876543210"))
+ require.NoError(t, err)
+
+ schema := iceberg.NewSchema(1, iceberg.NestedField{ID: 1, Name:
"value", Type: geometry})
+ pred := iceberg.SetPredicate(iceberg.OpIn, iceberg.Reference("value"),
[]iceberg.Literal{first, second})
+ eval, err := newInclusiveMetricsEvaluator(schema, pred, true, true)
+ require.NoError(t, err)
+
+ // Geo has no total ordering, so bound dispatch must avoid extrema
comparisons.
+ // Missing bounds leave metrics pruning conservative.
+ mightMatch, err := eval(inclusiveMetricsInTestFile(t, nil, nil))
+ require.NoError(t, err)
+ require.True(t, mightMatch)
+}
+
+func TestInclusiveMetricsEvaluatorInPredicateExtremaVariantExtract(t
*testing.T) {
+ schema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 1, Name: "payload", Type: iceberg.VariantType{},
+ })
+ expr := iceberg.IsIn(
+ iceberg.Extract("payload", "$.a", iceberg.PrimitiveTypes.Int64),
+ int64(1), int64(2),
+ )
+ bound, err := iceberg.BindExpr(schema, expr, true)
+ require.NoError(t, err)
+ pred, ok := bound.(iceberg.BoundSetPredicate)
+ require.True(t, ok)
+
+ newVisitor := func() *inclusiveMetricsEval {
+ return &inclusiveMetricsEval{metricsEvaluator: metricsEvaluator{
+ valueCounts: map[int]int64{1: 10},
+ nullCounts: map[int]int64{1: 0},
+ nanCounts: map[int]int64{1: 0},
+ lowerBounds: map[int][]byte{1:
variantMetricBoundInt64(t, "$['a']", 10)},
+ upperBounds: map[int][]byte{1:
variantMetricBoundInt64(t, "$['a']", 20)},
+ }}
+ }
+
+ slow := newVisitor().VisitIn(pred.Term(), pred.Literals())
+ fast := newVisitor().VisitInWithExtrema(
+ pred.Term(), pred.Literals(), iceberg.NewLiteral(int64(1)),
iceberg.NewLiteral(int64(2)),
+ )
+ require.False(t, slow)
+ require.Equal(t, slow, fast)
+}
+
+func variantMetricBoundInt64(t *testing.T, path string, value int64) []byte {
+ t.Helper()
+
+ var builder variant.Builder
+ start := builder.Offset()
+ entries := []variant.FieldEntry{builder.NextField(start, path)}
+ require.NoError(t, builder.AppendInt(value))
+ require.NoError(t, builder.FinishObject(start, entries))
+ v, err := builder.Build()
+ require.NoError(t, err)
+
+ return append(append([]byte{}, v.Metadata().Bytes()...), v.Bytes()...)
+}
+
+func inclusiveMetricsInTestFile(t *testing.T, lowerBound, upperBound []byte)
iceberg.DataFile {
+ t.Helper()
+
+ builder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentData,
"file.parquet", iceberg.ParquetFile,
+ nil, nil, nil, 10, 100,
+ )
+ require.NoError(t, err)
+
+ lowerBounds := map[int][]byte{}
+ if lowerBound != nil {
+ lowerBounds[1] = lowerBound
+ }
+ upperBounds := map[int][]byte{}
+ if upperBound != nil {
+ upperBounds[1] = upperBound
+ }
+
+ return builder.
+ ValueCounts(map[int]int64{1: 10}).
+ NullValueCounts(map[int]int64{1: 0}).
+ NaNValueCounts(map[int]int64{1: 0}).
+ LowerBoundValues(lowerBounds).
+ UpperBoundValues(upperBounds).
+ Build()
+}
+
func TestRemoveBoundCheckSupportsTimestampNanos(t *testing.T) {
bound := iceberg.NewLiteral(iceberg.TimestampNano(2))
values := []iceberg.Literal{
diff --git a/visitors.go b/visitors.go
index 896cb2b65..da3c266d9 100644
--- a/visitors.go
+++ b/visitors.go
@@ -83,8 +83,8 @@ type BoundGeospatialExprVisitor[T any] interface {
}
// boundSetExtremaExprVisitor is consulted only by the borrowed
-// VisitBoundPredicateRef path. The table package's manifest evaluator is
-// currently the only implementation.
+// VisitBoundPredicateRef path. It is used by trusted evaluators that can prune
+// an IN predicate from its precomputed minimum and maximum literals.
type boundSetExtremaExprVisitor[T any] interface {
VisitInWithExtrema(BoundTerm, Set[Literal], Literal, Literal) T
}
