From 635798d5c99e76284941976f3a521475e48c4db0 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sat, 29 Aug 2026 14:32:01 +0200 Subject: [PATCH] perf(table): compute partition residuals for local scan tasks --- table/arrow_scanner.go | 17 +- table/arrow_scanner_test.go | 37 ++- table/partition_residual.go | 318 +++++++++++++++++++++++ table/partition_residual_bench_test.go | 106 ++++++++ table/partition_residual_test.go | 338 +++++++++++++++++++++++++ table/scanner.go | 36 ++- 6 files changed, 847 insertions(+), 5 deletions(-) create mode 100644 table/partition_residual.go create mode 100644 table/partition_residual_bench_test.go create mode 100644 table/partition_residual_test.go diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go index 73a3cf8ff..160e649c9 100644 --- a/table/arrow_scanner.go +++ b/table/arrow_scanner.go @@ -869,7 +869,10 @@ func (as *arrowScan) projectedFieldIDs(rowFilter iceberg.BooleanExpression, equa } func (as *arrowScan) scanInvariants(tableProperties iceberg.Properties) (*arrowScanInvariants, error) { - projectedIDs, err := as.projectedFieldIDs(as.boundRowFilter, nil) + // Filter columns are added per task below. A local task may have a + // partition-elided residual that no longer references the original filter + // fields. + projectedIDs, err := as.projectedFieldIDs(nil, nil) if err != nil { return nil, err } @@ -883,6 +886,18 @@ func (as *arrowScan) scanInvariants(tableProperties iceberg.Properties) (*arrowS } func (as *arrowScan) addTaskProjectedFieldIDs(invariants *arrowScanInvariants, tasks []FileScanTask) error { + // Tasks without a residual still use the scan's original filter. Add those + // fields once, then add only the actual residual fields for other tasks. + for _, task := range tasks { + if task.Residual == nil { + if err := addFilterFieldIDs(invariants.projectedIDs, as.boundRowFilter); err != nil { + return err + } + + break + } + } + for _, task := range tasks { if task.Residual == nil { continue diff --git a/table/arrow_scanner_test.go b/table/arrow_scanner_test.go index 8e9cd511d..70cf5fcb4 100644 --- a/table/arrow_scanner_test.go +++ b/table/arrow_scanner_test.go @@ -149,7 +149,7 @@ func TestArrowScanSnapshotsInvariants(t *testing.T) { assert.Equal(t, 1, metadata.nameMappingCalls) } -func TestArrowScanAddTaskProjectedFieldIDsSkipsNilResiduals(t *testing.T) { +func TestArrowScanAddTaskProjectedFieldIDsUsesTaskResiduals(t *testing.T) { schema := iceberg.NewSchema(1, iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64}, iceberg.NestedField{ID: 2, Name: "data", Type: iceberg.PrimitiveTypes.String}, @@ -162,9 +162,9 @@ func TestArrowScanAddTaskProjectedFieldIDsSkipsNilResiduals(t *testing.T) { scanSchema: schema, boundRowFilter: boundFilter, } - invariants := &arrowScanInvariants{projectedIDs: set[int]{1: {}}} + invariants := &arrowScanInvariants{projectedIDs: set[int]{}} tasks := []FileScanTask{ - {}, + {Residual: iceberg.AlwaysTrue{}}, {Residual: iceberg.EqualTo(iceberg.Reference("data"), "value")}, {}, } @@ -174,6 +174,37 @@ func TestArrowScanAddTaskProjectedFieldIDsSkipsNilResiduals(t *testing.T) { assert.Equal(t, set[int]{1: {}, 2: {}}, invariants.projectedIDs) } +func TestArrowScanInvariantsStartWithRequestedFields(t *testing.T) { + schema := iceberg.NewSchema(1, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64}, + iceberg.NestedField{ID: 2, Name: "data", Type: iceberg.PrimitiveTypes.String}, + ) + projected, err := schema.Select(true, "data") + require.NoError(t, err) + metadata, err := NewMetadata( + schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "mem://test/table", iceberg.Properties{}, + ) + require.NoError(t, err) + + boundFilter, err := iceberg.BindExpr(schema, + iceberg.EqualTo(iceberg.Reference("id"), int64(1)), true) + require.NoError(t, err) + + scanner := &arrowScan{ + metadata: metadata, + scanSchema: schema, + projectedSchema: projected, + boundRowFilter: boundFilter, + } + invariants, err := scanner.scanInvariants(metadata.Properties()) + require.NoError(t, err) + assert.Equal(t, set[int]{2: {}}, invariants.projectedIDs) + + err = scanner.addTaskProjectedFieldIDs(invariants, []FileScanTask{{Residual: iceberg.AlwaysTrue{}}}) + require.NoError(t, err) + assert.Equal(t, set[int]{2: {}}, invariants.projectedIDs) +} + func TestEnrichRecordsWithPosDeleteFields(t *testing.T) { testSchema := arrow.NewSchema([]arrow.Field{ {Name: "first_name", Type: &arrow.StringType{}, Nullable: false}, diff --git a/table/partition_residual.go b/table/partition_residual.go new file mode 100644 index 000000000..e57035cf0 --- /dev/null +++ b/table/partition_residual.go @@ -0,0 +1,318 @@ +// 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" + + "github.com/apache/iceberg-go" +) + +// partitionResidualEvaluator partially evaluates a bound row filter using a +// file's partition values. It is built once per partition spec and can be +// reused for every data file in that spec. +type partitionResidualEvaluator struct { + root *partitionResidualNode + partitionType *iceberg.StructType +} + +type partitionResidualNodeKind uint8 + +const ( + partitionResidualOpaque partitionResidualNodeKind = iota + partitionResidualTrue + partitionResidualFalse + partitionResidualPredicateKind + partitionResidualNot + partitionResidualAnd + partitionResidualOr +) + +type partitionResidualNode struct { + kind partitionResidualNodeKind + expr iceberg.BooleanExpression + predicate *partitionResidualPredicate + left *partitionResidualNode + right *partitionResidualNode + child *partitionResidualNode +} + +type partitionResidualPredicate struct { + original iceberg.BoundPredicate + projections []partitionResidualProjection +} + +type partitionResidualProjection struct { + // strictComplement evaluates the inclusive projection of the negated + // predicate. If it is false, the original predicate is true for every + // value in this partition. + strictComplement func(_ iceberg.StructLike) (bool, error) + inclusive func(_ iceberg.StructLike) (bool, error) +} + +type partitionResidualBuilder struct { + spec iceberg.PartitionSpec + partitionSchema *iceberg.Schema + caseSensitive bool + hasProjection bool + err error +} + +// newPartitionResidualEvaluator creates a Java-style residual evaluator for +// one partition spec. The filter must already be bound to the scan schema. +// Unsupported transforms and predicates simply keep the original predicate. +func newPartitionResidualEvaluator( + schema *iceberg.Schema, + spec *iceberg.PartitionSpec, + filter iceberg.BooleanExpression, + caseSensitive bool, +) (*partitionResidualEvaluator, error) { + if schema == nil || spec == nil || filter == nil { + return nil, nil + } + + partitionType := spec.PartitionType(schema) + builder := &partitionResidualBuilder{ + spec: *spec, + partitionSchema: iceberg.NewSchema(0, partitionType.FieldList...), + caseSensitive: caseSensitive, + } + root, err := iceberg.VisitExpr(filter, builder) + if err != nil { + return nil, err + } + if builder.err != nil { + return nil, builder.err + } + if !builder.hasProjection { + return nil, nil + } + + return &partitionResidualEvaluator{ + root: root, + partitionType: partitionType, + }, nil +} + +func (b *partitionResidualBuilder) VisitTrue() *partitionResidualNode { + return &partitionResidualNode{kind: partitionResidualTrue, expr: iceberg.AlwaysTrue{}} +} + +func (b *partitionResidualBuilder) VisitFalse() *partitionResidualNode { + return &partitionResidualNode{kind: partitionResidualFalse, expr: iceberg.AlwaysFalse{}} +} + +func (b *partitionResidualBuilder) VisitNot(child *partitionResidualNode) *partitionResidualNode { + return &partitionResidualNode{ + kind: partitionResidualNot, + expr: iceberg.NewNot(child.expr), + child: child, + } +} + +func (b *partitionResidualBuilder) VisitAnd(left, right *partitionResidualNode) *partitionResidualNode { + return &partitionResidualNode{ + kind: partitionResidualAnd, + expr: iceberg.NewAnd(left.expr, right.expr), + left: left, + right: right, + } +} + +func (b *partitionResidualBuilder) VisitOr(left, right *partitionResidualNode) *partitionResidualNode { + return &partitionResidualNode{ + kind: partitionResidualOr, + expr: iceberg.NewOr(left.expr, right.expr), + left: left, + right: right, + } +} + +func (b *partitionResidualBuilder) VisitUnbound(pred iceberg.UnboundPredicate) *partitionResidualNode { + return &partitionResidualNode{kind: partitionResidualOpaque, expr: pred} +} + +func (b *partitionResidualBuilder) VisitBound(pred iceberg.BoundPredicate) *partitionResidualNode { + parts := b.spec.FieldsBySourceID(pred.Ref().Field().ID) + if len(parts) == 0 { + return &partitionResidualNode{kind: partitionResidualOpaque, expr: pred} + } + + projections := make([]partitionResidualProjection, 0, len(parts)) + for _, part := range parts { + projection, err := newPartitionResidualProjection( + b.partitionSchema, part, pred, b.caseSensitive) + if err != nil { + if b.err == nil { + b.err = fmt.Errorf("build partition residual for %s: %w", pred, err) + } + + return &partitionResidualNode{kind: partitionResidualOpaque, expr: pred} + } + if projection.strictComplement == nil && projection.inclusive == nil { + continue + } + + projections = append(projections, projection) + } + if len(projections) == 0 { + return &partitionResidualNode{kind: partitionResidualOpaque, expr: pred} + } + + b.hasProjection = true + + return &partitionResidualNode{ + kind: partitionResidualPredicateKind, + expr: pred, + predicate: &partitionResidualPredicate{ + original: pred, + projections: projections, + }, + } +} + +func newPartitionResidualProjection( + partitionSchema *iceberg.Schema, + part iceberg.PartitionField, + pred iceberg.BoundPredicate, + caseSensitive bool, +) (partitionResidualProjection, error) { + var projection partitionResidualProjection + + negated, ok := pred.Negate().(iceberg.BoundPredicate) + if ok { + strictComplement, err := bindPartitionProjection( + partitionSchema, part, negated, caseSensitive) + if err != nil { + return projection, err + } + projection.strictComplement = strictComplement + } + + inclusive, err := bindPartitionProjection(partitionSchema, part, pred, caseSensitive) + if err != nil { + return projection, err + } + projection.inclusive = inclusive + + return projection, nil +} + +func bindPartitionProjection( + partitionSchema *iceberg.Schema, + part iceberg.PartitionField, + pred iceberg.BoundPredicate, + caseSensitive bool, +) (func(_ iceberg.StructLike) (bool, error), error) { + projected, err := part.Transform.Project(part.Name, pred) + if err != nil || projected == nil { + return nil, err + } + + return iceberg.ExpressionEvaluator(partitionSchema, projected, caseSensitive) +} + +// residual returns the portion of the original filter that still needs to be +// evaluated for a file. The changed result is false when no partition value +// simplified the filter, allowing callers to retain the nil-residual fallback. +func (p *partitionResidualEvaluator) residual( + partition map[int]any, +) (iceberg.BooleanExpression, bool, error) { + residual, changed, err := p.root.residual( + borrowedPartitionRecord{partition: partition, partitionType: p.partitionType}, + ) + if !changed { + return nil, false, err + } + + return residual, true, err +} + +func (n *partitionResidualNode) residual( + partition iceberg.StructLike, +) (iceberg.BooleanExpression, bool, error) { + switch n.kind { + case partitionResidualOpaque: + return n.expr, false, nil + case partitionResidualTrue, partitionResidualFalse: + return n.expr, false, nil + case partitionResidualPredicateKind: + for _, projection := range n.predicate.projections { + if projection.strictComplement != nil { + matches, err := projection.strictComplement(partition) + if err != nil { + return nil, false, err + } + if !matches { + return iceberg.AlwaysTrue{}, true, nil + } + } + + if projection.inclusive != nil { + matches, err := projection.inclusive(partition) + if err != nil { + return nil, false, err + } + if !matches { + return iceberg.AlwaysFalse{}, true, nil + } + } + } + + return n.predicate.original, false, nil + case partitionResidualNot: + child, changed, err := n.child.residual(partition) + if err != nil { + return nil, false, err + } + + return iceberg.NewNot(child), changed, nil + case partitionResidualAnd: + left, leftChanged, err := n.left.residual(partition) + if err != nil { + return nil, false, err + } + if _, alwaysFalse := left.(iceberg.AlwaysFalse); alwaysFalse { + return left, leftChanged, nil + } + + right, rightChanged, err := n.right.residual(partition) + if err != nil { + return nil, false, err + } + + return iceberg.NewAnd(left, right), leftChanged || rightChanged, nil + case partitionResidualOr: + left, leftChanged, err := n.left.residual(partition) + if err != nil { + return nil, false, err + } + if _, alwaysTrue := left.(iceberg.AlwaysTrue); alwaysTrue { + return left, leftChanged, nil + } + + right, rightChanged, err := n.right.residual(partition) + if err != nil { + return nil, false, err + } + + return iceberg.NewOr(left, right), leftChanged || rightChanged, nil + } + + return n.expr, false, nil +} diff --git a/table/partition_residual_bench_test.go b/table/partition_residual_bench_test.go new file mode 100644 index 000000000..f7ae51bad --- /dev/null +++ b/table/partition_residual_bench_test.go @@ -0,0 +1,106 @@ +// 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 ( + "testing" + + "github.com/apache/iceberg-go" +) + +var partitionResidualBenchmarkSink iceberg.BooleanExpression + +func BenchmarkPartitionResidualPlanning(b *testing.B) { + for _, tc := range []struct { + name string + transform iceberg.Transform + filter iceberg.BooleanExpression + partition func(int) any + }{ + { + name: "identity", + transform: iceberg.IdentityTransform{}, + filter: iceberg.NewAnd( + iceberg.EqualTo(iceberg.Reference("tenant_id"), "acme"), + iceberg.GreaterThan(iceberg.Reference("amount"), int64(100)), + ), + partition: func(i int) any { + if i%2 == 0 { + return "acme" + } + + return "other" + }, + }, + { + name: "day-range", + transform: iceberg.DayTransform{}, + filter: iceberg.NewAnd( + iceberg.GreaterThanEqual(iceberg.Reference("event_ts"), "2022-11-27T10:00:00"), + iceberg.LessThan(iceberg.Reference("event_ts"), "2022-11-30T10:00:00"), + ), + partition: func(i int) any { + return iceberg.Date(19323 + i%5) + }, + }, + } { + b.Run(tc.name+"/files=4096", func(b *testing.B) { + var schema *iceberg.Schema + if tc.name == "identity" { + schema = partitionResidualTestSchema() + } else { + schema = iceberg.NewSchema(1, iceberg.NestedField{ + ID: 1, Name: "event_ts", Type: iceberg.PrimitiveTypes.Timestamp, + }) + } + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, FieldID: 1000, Name: "partition", Transform: tc.transform, + }) + bound, err := iceberg.BindExpr(schema, tc.filter, true) + if err != nil { + b.Fatal(err) + } + evaluator, err := newPartitionResidualEvaluator(schema, &spec, bound, true) + if err != nil { + b.Fatal(err) + } + if evaluator == nil { + b.Fatal("expected partition residual evaluator") + } + + partitions := make([]map[int]any, 4096) + for i := range partitions { + partitions[i] = map[int]any{1000: tc.partition(i)} + } + + b.ReportAllocs() + b.ResetTimer() + for b.Loop() { + for _, partition := range partitions { + residual, _, err := evaluator.residual(partition) + if err != nil { + b.Fatal(err) + } + partitionResidualBenchmarkSink = residual + } + } + b.StopTimer() + b.ReportMetric(float64(len(partitions)), "files/op") + }) + } +} diff --git a/table/partition_residual_test.go b/table/partition_residual_test.go new file mode 100644 index 000000000..061c26a60 --- /dev/null +++ b/table/partition_residual_test.go @@ -0,0 +1,338 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "bytes" + "context" + "testing" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func partitionResidualTestSchema() *iceberg.Schema { + return iceberg.NewSchema(1, + iceberg.NestedField{ID: 1, Name: "tenant_id", Type: iceberg.PrimitiveTypes.String}, + iceberg.NestedField{ID: 2, Name: "amount", Type: iceberg.PrimitiveTypes.Int64}, + ) +} + +func boundPartitionResidualEvaluator( + t *testing.T, + schema *iceberg.Schema, + spec iceberg.PartitionSpec, + filter iceberg.BooleanExpression, +) *partitionResidualEvaluator { + t.Helper() + + bound, err := iceberg.BindExpr(schema, filter, true) + require.NoError(t, err) + + evaluator, err := newPartitionResidualEvaluator(schema, &spec, bound, true) + require.NoError(t, err) + require.NotNil(t, evaluator) + + return evaluator +} + +func TestPartitionResidualEvaluatorElidesSatisfiedIdentityPredicate(t *testing.T) { + schema := partitionResidualTestSchema() + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, FieldID: 1000, Name: "tenant_id", Transform: iceberg.IdentityTransform{}, + }) + filter := iceberg.NewAnd( + iceberg.EqualTo(iceberg.Reference("tenant_id"), "acme"), + iceberg.GreaterThan(iceberg.Reference("amount"), int64(100)), + ) + evaluator := boundPartitionResidualEvaluator(t, schema, spec, filter) + + residual, simplified, err := evaluator.residual(map[int]any{1000: "acme"}) + require.NoError(t, err) + require.True(t, simplified) + + want, err := iceberg.BindExpr(schema, + iceberg.GreaterThan(iceberg.Reference("amount"), int64(100)), true) + require.NoError(t, err) + assert.True(t, residual.Equals(want), "expected %s, got %s", want, residual) + + residual, simplified, err = evaluator.residual(map[int]any{1000: "other"}) + require.NoError(t, err) + require.True(t, simplified) + assert.Equal(t, iceberg.AlwaysFalse{}, residual) +} + +func TestPartitionResidualEvaluatorSimplifiesBooleanCombinations(t *testing.T) { + schema := partitionResidualTestSchema() + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, FieldID: 1000, Name: "tenant_id", Transform: iceberg.IdentityTransform{}, + }) + + t.Run("or", func(t *testing.T) { + filter := iceberg.NewOr( + iceberg.EqualTo(iceberg.Reference("tenant_id"), "acme"), + iceberg.GreaterThan(iceberg.Reference("amount"), int64(100)), + ) + evaluator := boundPartitionResidualEvaluator(t, schema, spec, filter) + + residual, simplified, err := evaluator.residual(map[int]any{1000: "acme"}) + require.NoError(t, err) + require.True(t, simplified) + assert.Equal(t, iceberg.AlwaysTrue{}, residual) + + residual, simplified, err = evaluator.residual(map[int]any{1000: "other"}) + require.NoError(t, err) + require.True(t, simplified) + want, err := iceberg.BindExpr(schema, + iceberg.GreaterThan(iceberg.Reference("amount"), int64(100)), true) + require.NoError(t, err) + assert.True(t, residual.Equals(want)) + }) + + t.Run("not", func(t *testing.T) { + filter := iceberg.NewNot(iceberg.EqualTo(iceberg.Reference("tenant_id"), "acme")) + evaluator := boundPartitionResidualEvaluator(t, schema, spec, filter) + + residual, simplified, err := evaluator.residual(map[int]any{1000: "acme"}) + require.NoError(t, err) + require.True(t, simplified) + assert.Equal(t, iceberg.AlwaysFalse{}, residual) + + residual, simplified, err = evaluator.residual(map[int]any{1000: "other"}) + require.NoError(t, err) + require.True(t, simplified) + assert.Equal(t, iceberg.AlwaysTrue{}, residual) + }) +} + +func TestPartitionResidualEvaluatorHandlesNullAndSetPredicates(t *testing.T) { + schema := partitionResidualTestSchema() + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, FieldID: 1000, Name: "tenant_id", Transform: iceberg.IdentityTransform{}, + }) + + t.Run("is null", func(t *testing.T) { + evaluator := boundPartitionResidualEvaluator(t, schema, spec, + iceberg.IsNull(iceberg.Reference("tenant_id"))) + + residual, simplified, err := evaluator.residual(map[int]any{1000: nil}) + require.NoError(t, err) + require.True(t, simplified) + assert.Equal(t, iceberg.AlwaysTrue{}, residual) + + residual, simplified, err = evaluator.residual(map[int]any{1000: "acme"}) + require.NoError(t, err) + require.True(t, simplified) + assert.Equal(t, iceberg.AlwaysFalse{}, residual) + }) + + t.Run("in", func(t *testing.T) { + evaluator := boundPartitionResidualEvaluator(t, schema, spec, + iceberg.IsIn(iceberg.Reference("tenant_id"), "acme", "iceberg")) + + residual, simplified, err := evaluator.residual(map[int]any{1000: "acme"}) + require.NoError(t, err) + require.True(t, simplified) + assert.Equal(t, iceberg.AlwaysTrue{}, residual) + + residual, simplified, err = evaluator.residual(map[int]any{1000: "other"}) + require.NoError(t, err) + require.True(t, simplified) + assert.Equal(t, iceberg.AlwaysFalse{}, residual) + }) +} + +func TestPartitionResidualEvaluatorComputesDayBoundaries(t *testing.T) { + schema := iceberg.NewSchema(1, + iceberg.NestedField{ID: 1, Name: "event_ts", Type: iceberg.PrimitiveTypes.Timestamp}, + ) + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, FieldID: 1000, Name: "event_day", Transform: iceberg.DayTransform{}, + }) + filter := iceberg.NewAnd( + iceberg.GreaterThanEqual(iceberg.Reference("event_ts"), "2022-11-27T10:00:00"), + iceberg.LessThan(iceberg.Reference("event_ts"), "2022-11-30T10:00:00"), + ) + evaluator := boundPartitionResidualEvaluator(t, schema, spec, filter) + + lower, err := iceberg.BindExpr(schema, + iceberg.GreaterThanEqual(iceberg.Reference("event_ts"), "2022-11-27T10:00:00"), true) + require.NoError(t, err) + upper, err := iceberg.BindExpr(schema, + iceberg.LessThan(iceberg.Reference("event_ts"), "2022-11-30T10:00:00"), true) + require.NoError(t, err) + + tests := []struct { + name string + partition iceberg.Date + want iceberg.BooleanExpression + }{ + {name: "lower boundary", partition: iceberg.Date(19323), want: lower}, + {name: "interior", partition: iceberg.Date(19324), want: iceberg.AlwaysTrue{}}, + {name: "upper boundary", partition: iceberg.Date(19326), want: upper}, + {name: "outside", partition: iceberg.Date(19327), want: iceberg.AlwaysFalse{}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + residual, simplified, err := evaluator.residual(map[int]any{1000: tt.partition}) + require.NoError(t, err) + require.True(t, simplified) + assert.True(t, residual.Equals(tt.want), "expected %s, got %s", tt.want, residual) + }) + } +} + +func TestPartitionResidualEvaluatorComputesTruncateBoundaries(t *testing.T) { + schema := iceberg.NewSchema(1, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32}, + ) + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, FieldID: 1000, Name: "id_truncated", Transform: iceberg.TruncateTransform{Width: 10}, + }) + filter := iceberg.NewAnd( + iceberg.GreaterThanEqual(iceberg.Reference("id"), int32(25)), + iceberg.LessThan(iceberg.Reference("id"), int32(65)), + ) + evaluator := boundPartitionResidualEvaluator(t, schema, spec, filter) + + lower, err := iceberg.BindExpr(schema, + iceberg.GreaterThanEqual(iceberg.Reference("id"), int32(25)), true) + require.NoError(t, err) + upper, err := iceberg.BindExpr(schema, + iceberg.LessThan(iceberg.Reference("id"), int32(65)), true) + require.NoError(t, err) + + tests := []struct { + name string + partition int32 + want iceberg.BooleanExpression + }{ + {name: "before range", partition: 10, want: iceberg.AlwaysFalse{}}, + {name: "lower boundary", partition: 20, want: lower}, + {name: "interior", partition: 40, want: iceberg.AlwaysTrue{}}, + {name: "upper boundary", partition: 60, want: upper}, + {name: "after range", partition: 70, want: iceberg.AlwaysFalse{}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + residual, simplified, err := evaluator.residual(map[int]any{1000: tt.partition}) + require.NoError(t, err) + require.True(t, simplified) + assert.True(t, residual.Equals(tt.want), "expected %s, got %s", tt.want, residual) + }) + } +} + +func TestPartitionResidualEvaluatorHandlesTransformedFilterTerms(t *testing.T) { + schema := iceberg.NewSchema(1, + iceberg.NestedField{ID: 1, Name: "event_ts", Type: iceberg.PrimitiveTypes.Timestamp}, + ) + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, FieldID: 1000, Name: "event_day", Transform: iceberg.DayTransform{}, + }) + filter := iceberg.EqualTo( + iceberg.NewUnboundTransform(iceberg.DayTransform{}, iceberg.Reference("event_ts")), + iceberg.Date(19323), + ) + evaluator := boundPartitionResidualEvaluator(t, schema, spec, filter) + + residual, simplified, err := evaluator.residual(map[int]any{1000: iceberg.Date(19323)}) + require.NoError(t, err) + require.True(t, simplified) + assert.Equal(t, iceberg.AlwaysTrue{}, residual) + + residual, simplified, err = evaluator.residual(map[int]any{1000: iceberg.Date(19324)}) + require.NoError(t, err) + require.True(t, simplified) + assert.Equal(t, iceberg.AlwaysFalse{}, residual) +} + +func TestPartitionResidualEvaluatorKeepsBucketPredicateConservative(t *testing.T) { + schema := partitionResidualTestSchema() + transform := iceberg.BucketTransform{NumBuckets: 16} + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, FieldID: 1000, Name: "tenant_bucket", Transform: transform, + }) + filter := iceberg.EqualTo(iceberg.Reference("tenant_id"), "acme") + evaluator := boundPartitionResidualEvaluator(t, schema, spec, filter) + + partition := transform.Apply(iceberg.Optional[iceberg.Literal]{ + Valid: true, Val: iceberg.StringLiteral("acme"), + }) + require.True(t, partition.Valid) + residual, simplified, err := evaluator.residual(map[int]any{1000: partition.Val.Any()}) + require.NoError(t, err) + assert.False(t, simplified, "a bucket match does not prove equality because buckets can collide") + assert.Nil(t, residual) +} + +func TestPartitionResidualEvaluatorLeavesUnpartitionedFiltersUnchanged(t *testing.T) { + schema := partitionResidualTestSchema() + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, FieldID: 1000, Name: "tenant_bucket", + Transform: iceberg.BucketTransform{NumBuckets: 16}, + }) + filter := iceberg.GreaterThan(iceberg.Reference("amount"), int64(100)) + + bound, err := iceberg.BindExpr(schema, filter, true) + require.NoError(t, err) + evaluator, err := newPartitionResidualEvaluator(schema, &spec, bound, true) + require.NoError(t, err) + assert.Nil(t, evaluator) +} + +func TestPlanFilesLocalPopulatesPartitionResidual(t *testing.T) { + const ( + manifestPath = "mem://default/table/metadata/manifest.avro" + manifestListPath = "mem://default/table/metadata/snap-7.avro" + dataPath = "mem://default/table/data/file.parquet" + ) + + spec := partitionedSpec() + fs := iceio.NewMemFS() + scan, schema, snapshotID := newSchemaEvolutionScanWithSnapshot( + t, &spec, fs, manifestListPath, nil) + dataFile := newTestDataFile(t, spec, dataPath, map[int]any{1000: int32(5)}) + entry := iceberg.NewManifestEntryBuilder( + iceberg.EntryStatusADDED, &snapshotID, dataFile, + ).SequenceNum(1).Build() + + var manifestBuffer bytes.Buffer + manifest, err := iceberg.WriteManifest( + manifestPath, &manifestBuffer, 2, spec, schema, snapshotID, + []iceberg.ManifestEntry{entry}, + ) + require.NoError(t, err) + require.NoError(t, fs.WriteFile(manifestPath, manifestBuffer.Bytes())) + + var listBuffer bytes.Buffer + sequenceNumber := int64(1) + require.NoError(t, iceberg.WriteManifestList( + 2, &listBuffer, snapshotID, nil, &sequenceNumber, 0, + []iceberg.ManifestFile{manifest}, + )) + require.NoError(t, fs.WriteFile(manifestListPath, listBuffer.Bytes())) + + tasks, err := scan.PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 1) + assert.Equal(t, iceberg.AlwaysTrue{}, tasks[0].Residual, + "the identity partition already proves id == 5") +} diff --git a/table/scanner.go b/table/scanner.go index f5599d932..f9a1f186b 100644 --- a/table/scanner.go +++ b/table/scanner.go @@ -1117,6 +1117,18 @@ func (scan *Scan) planFilesLocal(ctx context.Context, acc *scanMetricsAccumulato return nil, err } + var boundRowFilter iceberg.BooleanExpression + if scan.rowFilter != nil && !scan.rowFilter.Equals(iceberg.AlwaysTrue{}) { + boundRowFilter, err = iceberg.BindExpr(schema, scan.rowFilter, scan.caseSensitive) + if err != nil { + return nil, err + } + } + var residualEvaluators map[int]*partitionResidualEvaluator + if boundRowFilter != nil { + residualEvaluators = make(map[int]*partitionResidualEvaluator) + } + // Step 3: Index positional deletes and match them to data files. posDeleteIndex, err := buildPositionalDeleteIndex(entries.positionalDeleteEntries) if err != nil { @@ -1161,6 +1173,28 @@ func (scan *Scan) planFilesLocal(ctx context.Context, acc *scanMetricsAccumulato Start: 0, Length: e.DataFile().FileSizeBytes(), } + if boundRowFilter != nil { + specID := int(e.DataFile().SpecID()) + residualEvaluator, found := residualEvaluators[specID] + if !found { + residualEvaluator, err = newPartitionResidualEvaluator( + schema, scan.metadata.PartitionSpecByID(specID), boundRowFilter, scan.caseSensitive) + if err != nil { + return nil, fmt.Errorf("build partition residual evaluator for spec %d: %w", specID, err) + } + residualEvaluators[specID] = residualEvaluator + } + if residualEvaluator != nil { + var simplified bool + task.Residual, simplified, err = residualEvaluator.residual(dataFilePartition(e.DataFile())) + if err != nil { + return nil, fmt.Errorf("evaluate partition residual for %s: %w", e.DataFile().FilePath(), err) + } + if !simplified { + task.Residual = nil + } + } + } // Row lineage constants: readers use these to synthesize _row_id and // _last_updated_sequence_number when requested. Per spec the // synthesized _last_updated_sequence_number is the manifest entry's @@ -1435,7 +1469,7 @@ type FileScanTask struct { DeletionVectorFiles []iceberg.DataFile // deletion vectors (puffin files) Start, Length int64 // Residual is the portion of the scan filter that must still be evaluated - // for this task. Remote planners may simplify the original filter using + // for this task. Local and remote planners may simplify the original filter using // file metadata; nil means the caller did not provide a task residual. // ReadTasks applies the scan's original row filter and each task residual. Residual iceberg.BooleanExpression