This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 61318c413 perf(table): avoid cloning ReadTasks for already-bound
residuals (#2106)
61318c413 is described below
commit 61318c41303d3968f4a846e1703297e8eb3a68d8
Author: Minh Vu <[email protected]>
AuthorDate: Sun Oct 11 20:30:34 2026 +0200
perf(table): avoid cloning ReadTasks for already-bound residuals (#2106)
---
table/arrow_scanner.go | 32 +-
table/arrow_scanner_bench_test.go | 58 +++
table/arrow_scanner_pruning_internal_test.go | 18 +-
table/incremental_append_scan.go | 2 +-
table/incremental_changelog_scan.go | 2 +-
table/readtasks_residual_binding_internal_test.go | 426 ++++++++++++++++++++++
table/scanner.go | 97 +++--
7 files changed, 593 insertions(+), 42 deletions(-)
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index c1ebe9fd4..a482a2964 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -1248,6 +1248,9 @@ type arrowScan struct {
filterPlanCache compiledFileFilterPlanCache
fileReadPlanCache preparedFileReadPlanCache
cacheFileReadPlan bool
+ // taskResidualsBound means every non-nil residual passed to GetRecords
was
+ // bound and validated against filterSchema, so rowFilterForTask can
skip binding.
+ taskResidualsBound bool
useLargeTypes bool
concurrency int
@@ -1694,42 +1697,55 @@ func validateBoundFilter(schema *iceberg.Schema, filter
iceberg.BooleanExpressio
return validationErr
}
-func bindTaskFilter(schema *iceberg.Schema, filter iceberg.BooleanExpression,
caseSensitive bool) (iceberg.BooleanExpression, error) {
+// bindTaskFilter returns changed=true only when binding an unbound residual
+// produces a new expression. When changed is false, the returned expression is
+// the exact input value, after validating any already-bound predicates.
+func bindTaskFilter(schema *iceberg.Schema, filter iceberg.BooleanExpression,
caseSensitive bool) (iceberg.BooleanExpression, bool, error) {
if filter == nil {
- return nil, nil
+ return nil, false, nil
}
state, err := iceberg.VisitExpr(filter, filterBindingVisitor{})
if err != nil {
- return nil, err
+ return nil, false, err
}
if state.hasBound && state.hasUnbound {
- return nil, fmt.Errorf("%w: scan task residual mixes bound and
unbound predicates", iceberg.ErrInvalidArgument)
+ return nil, false, fmt.Errorf("%w: scan task residual mixes
bound and unbound predicates", iceberg.ErrInvalidArgument)
}
if !state.hasUnbound {
if state.hasBound {
if err := validateBoundFilter(schema, filter); err !=
nil {
- return nil, err
+ return nil, false, err
}
}
- return filter, nil
+ return filter, false, nil
+ }
+
+ bound, err := iceberg.BindExpr(schema, filter, caseSensitive)
+ if err != nil {
+ return nil, false, err
}
- return iceberg.BindExpr(schema, filter, caseSensitive)
+ return bound, true, nil
}
func (as *arrowScan) rowFilterForTask(task FileScanTask)
(iceberg.BooleanExpression, error) {
if task.Residual == nil {
return as.boundRowFilter, nil
}
+ if as.taskResidualsBound {
+ return task.Residual, nil
+ }
filterSchema := as.scanSchema
if as.filterSchema != nil {
filterSchema = as.filterSchema
}
- return bindTaskFilter(filterSchema, task.Residual, as.caseSensitive)
+ bound, _, err := bindTaskFilter(filterSchema, task.Residual,
as.caseSensitive)
+
+ return bound, err
}
// fieldIndexByID returns the index of the field carrying fieldID in its Arrow
diff --git a/table/arrow_scanner_bench_test.go
b/table/arrow_scanner_bench_test.go
index f3a1e5aa3..b5c5dcc20 100644
--- a/table/arrow_scanner_bench_test.go
+++ b/table/arrow_scanner_bench_test.go
@@ -687,3 +687,61 @@ func benchmarkTaskResidualJSON(rowCount int) string {
return result.String()
}
+
+func BenchmarkArrowScanReadTasksResidualBinding(b *testing.B) {
+ schema := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ )
+ metadata, err := NewMetadata(
+ schema, iceberg.UnpartitionedSpec, UnsortedSortOrder,
"mem://benchmark/read-tasks", nil,
+ )
+ if err != nil {
+ b.Fatal(err)
+ }
+ fs := iceio.NewMemFS()
+ scan := New(Identifier{"benchmark", "read_tasks"}, metadata,
"metadata.json",
+ func(context.Context) (iceio.IO, error) { return fs, nil },
nil).Scan()
+ boundResidual, err := iceberg.BindExpr(schema,
+ iceberg.GreaterThan(iceberg.Reference("id"), int64(1)), true)
+ if err != nil {
+ b.Fatal(err)
+ }
+ unboundResidual := iceberg.GreaterThan(iceberg.Reference("id"),
int64(1))
+
+ for _, workload := range []struct {
+ taskCount int
+ name string
+ residual iceberg.BooleanExpression
+ }{
+ {taskCount: 1_000, name: "bound", residual: boundResidual},
+ {taskCount: 10_000, name: "bound", residual: boundResidual},
+ {taskCount: 100_000, name: "bound", residual: boundResidual},
+ {taskCount: 100_000, name: "nil"},
+ {taskCount: 100_000, name: "unbound", residual:
unboundResidual},
+ {taskCount: 100_000, name: "late_unbound", residual:
boundResidual},
+ } {
+ b.Run(fmt.Sprintf("tasks=%d/residual=%s", workload.taskCount,
workload.name), func(b *testing.B) {
+ tasks := make([]FileScanTask, workload.taskCount)
+ for i := range tasks {
+ tasks[i].Residual = workload.residual
+ }
+ if workload.name == "late_unbound" {
+ tasks[len(tasks)-1].Residual = unboundResidual
+ }
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ _, records, err := scan.ReadTasks(b.Context(),
tasks)
+ if err != nil {
+ b.Fatal(err)
+ }
+ if records == nil {
+ b.Fatal("ReadTasks returned a nil
iterator")
+ }
+ // Do not drain the iterator: this benchmark
isolates setup and residual binding.
+ }
+ b.ReportMetric(float64(len(tasks)), "tasks/op")
+ })
+ }
+}
diff --git a/table/arrow_scanner_pruning_internal_test.go
b/table/arrow_scanner_pruning_internal_test.go
index 007e3d0d2..b34e6d1b8 100644
--- a/table/arrow_scanner_pruning_internal_test.go
+++ b/table/arrow_scanner_pruning_internal_test.go
@@ -253,16 +253,26 @@ func TestBindTaskFilterValidatesBoundSchema(t *testing.T)
{
require.NoError(t, err)
t.Run("matching schema is accepted", func(t *testing.T) {
- got, err := bindTaskFilter(int64Schema, bound, true)
+ got, changed, err := bindTaskFilter(int64Schema, bound, true)
require.NoError(t, err)
+ require.False(t, changed)
require.Same(t, bound, got)
})
+ t.Run("unbound filter is bound", func(t *testing.T) {
+ unbound := iceberg.EqualTo(iceberg.Reference("id"), int64(42))
+ got, changed, err := bindTaskFilter(int64Schema, unbound, true)
+ require.NoError(t, err)
+ require.True(t, changed)
+ require.NotNil(t, got)
+ require.NotSame(t, unbound, got)
+ })
+
t.Run("missing field is rejected", func(t *testing.T) {
otherSchema := iceberg.NewSchema(0,
iceberg.NestedField{ID: 2, Name: "other", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
)
- _, err := bindTaskFilter(otherSchema, bound, true)
+ _, _, err := bindTaskFilter(otherSchema, bound, true)
require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
require.ErrorContains(t, err, "field ID 1")
})
@@ -271,7 +281,7 @@ func TestBindTaskFilterValidatesBoundSchema(t *testing.T) {
otherSchema := iceberg.NewSchema(0,
iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int32, Required: true},
)
- _, err := bindTaskFilter(otherSchema, bound, true)
+ _, _, err := bindTaskFilter(otherSchema, bound, true)
require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
require.ErrorContains(t, err, "type")
})
@@ -289,7 +299,7 @@ func TestBindTaskFilterValidatesBoundSchema(t *testing.T) {
iceberg.NestedField{ID: 10, Name: "payload", Type:
&iceberg.StructType{FieldList: []iceberg.NestedField{field}}},
)
- _, err = bindTaskFilter(otherSchema, boundNested, true)
+ _, _, err = bindTaskFilter(otherSchema, boundNested, true)
require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
require.ErrorContains(t, err, "accessor path")
})
diff --git a/table/incremental_append_scan.go b/table/incremental_append_scan.go
index cdb0dc565..ed77dbe1b 100644
--- a/table/incremental_append_scan.go
+++ b/table/incremental_append_scan.go
@@ -123,7 +123,7 @@ func (s *IncrementalAppendScan) PlanFiles(ctx
context.Context) ([]FileScanTask,
if err != nil {
return nil, err
}
- residual, err := bindTaskFilter(schema, planningScan.rowFilter,
planningScan.caseSensitive)
+ residual, _, err := bindTaskFilter(schema, planningScan.rowFilter,
planningScan.caseSensitive)
if err != nil {
return nil, fmt.Errorf("bind incremental scan residual: %w",
err)
}
diff --git a/table/incremental_changelog_scan.go
b/table/incremental_changelog_scan.go
index dcb5f2620..8b8006adc 100644
--- a/table/incremental_changelog_scan.go
+++ b/table/incremental_changelog_scan.go
@@ -136,7 +136,7 @@ func (s *IncrementalChangelogScan) PlanFiles(ctx
context.Context) ([]ChangelogSc
if err != nil {
return nil, err
}
- residual, err := bindTaskFilter(schema, planningScan.rowFilter,
planningScan.caseSensitive)
+ residual, _, err := bindTaskFilter(schema, planningScan.rowFilter,
planningScan.caseSensitive)
if err != nil {
return nil, fmt.Errorf("bind incremental changelog scan
residual: %w", err)
}
diff --git a/table/readtasks_residual_binding_internal_test.go
b/table/readtasks_residual_binding_internal_test.go
new file mode 100644
index 000000000..f395ee0f1
--- /dev/null
+++ b/table/readtasks_residual_binding_internal_test.go
@@ -0,0 +1,426 @@
+// 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 (
+ "context"
+ "fmt"
+ "iter"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/pqarrow"
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/stretchr/testify/require"
+)
+
+func residualBindingTestScan(t *testing.T) (*Scan, *iceberg.Schema) {
+ t.Helper()
+
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64},
+ )
+ metadata, err := NewMetadata(
+ schema, iceberg.UnpartitionedSpec, UnsortedSortOrder,
"mem://mixed-residuals", nil,
+ )
+ require.NoError(t, err)
+ memFS := iceio.NewMemFS()
+ tbl := New(
+ Identifier{"db", "tbl"}, metadata, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return memFS, nil },
nil,
+ )
+
+ return tbl.Scan(), schema
+}
+
+func writeResidualBindingParquetFile(t testing.TB, path string, schema
*iceberg.Schema, jsonData string) iceberg.DataFile {
+ t.Helper()
+
+ // Real Iceberg field IDs are needed for the Parquet reader to reach
residual filtering.
+ arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false)
+ require.NoError(t, err)
+ record, _, err := array.RecordFromJSON(memory.DefaultAllocator,
arrowSchema, strings.NewReader(jsonData))
+ require.NoError(t, err)
+ defer record.Release()
+
+ writer, err := (iceio.LocalFS{}).Create(path)
+ require.NoError(t, err)
+ defer writer.Close()
+
+ tbl := array.NewTableFromRecords(arrowSchema,
[]arrow.RecordBatch{record})
+ defer tbl.Release()
+
+ props := parquet.NewWriterProperties(parquet.WithStats(true))
+ // WriteTable flushes and closes the file before returning.
+ require.NoError(t, pqarrow.WriteTable(
+ tbl, writer, record.NumRows(), props,
pqarrow.DefaultWriterProps(),
+ ))
+
+ info, err := os.Stat(path)
+ require.NoError(t, err)
+ builder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec,
+ iceberg.EntryContentData,
+ path,
+ iceberg.ParquetFile,
+ nil, nil, nil,
+ record.NumRows(),
+ info.Size(),
+ )
+ require.NoError(t, err)
+
+ return builder.Build()
+}
+
+func TestBindReadTasksResidualsCopyOnWrite(t *testing.T) {
+ _, schema := residualBindingTestScan(t)
+ unbound := iceberg.GreaterThan(iceberg.Reference("id"), int64(1))
+ bound, err := iceberg.BindExpr(schema, unbound, true)
+ require.NoError(t, err)
+
+ tests := []struct {
+ name string
+ residuals []iceberg.BooleanExpression
+ wantAlias bool
+ wantSame []bool
+ }{
+ {
+ name: "all bound and nil",
+ residuals: []iceberg.BooleanExpression{bound, nil,
bound},
+ wantAlias: true,
+ wantSame: []bool{true, false, true},
+ },
+ {
+ name: "all nil",
+ residuals: []iceberg.BooleanExpression{nil, nil, nil},
+ wantAlias: true,
+ wantSame: []bool{false, false, false},
+ },
+ {
+ name: "mixed",
+ residuals: []iceberg.BooleanExpression{bound, nil,
unbound, bound},
+ wantSame: []bool{true, false, false, true},
+ },
+ {
+ name: "first task unbound",
+ residuals: []iceberg.BooleanExpression{unbound, bound},
+ wantSame: []bool{false, true},
+ },
+ {
+ name: "last task unbound",
+ residuals: []iceberg.BooleanExpression{bound, nil,
unbound},
+ wantSame: []bool{true, false, false},
+ },
+ {
+ name: "all unbound",
+ residuals: []iceberg.BooleanExpression{unbound,
unbound},
+ wantSame: []bool{false, false},
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ tasks := make([]FileScanTask, len(tt.residuals))
+ for i, residual := range tt.residuals {
+ tasks[i].Residual = residual
+ }
+
+ got, err := bindReadTasksResiduals(schema, tasks, true)
+ require.NoError(t, err)
+ require.Len(t, got, len(tasks))
+ if len(tasks) > 0 {
+ require.Equal(t, tt.wantAlias, &got[0] ==
&tasks[0])
+ }
+
+ for i, original := range tt.residuals {
+ if original == nil {
+ require.Nil(t, got[i].Residual)
+
+ continue
+ }
+
+ // The input plan is never rewritten, even when
the output needs binding.
+ require.Same(t, original, tasks[i].Residual)
+ if tt.wantSame[i] {
+ require.Same(t, original,
got[i].Residual)
+ } else {
+ require.NotSame(t, original,
got[i].Residual)
+ }
+ state, visitErr :=
iceberg.VisitExpr(got[i].Residual, filterBindingVisitor{})
+ require.NoError(t, visitErr)
+ require.True(t, state.hasBound)
+ require.False(t, state.hasUnbound)
+ }
+ })
+ }
+}
+
+func TestReadTasksResidualPlanIsReusable(t *testing.T) {
+ scan, schema := residualBindingTestScan(t)
+ unbound := iceberg.GreaterThan(iceberg.Reference("id"), int64(1))
+ bound, err := iceberg.BindExpr(schema, unbound, true)
+ require.NoError(t, err)
+
+ wrongSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 2, Name: "other", Type:
iceberg.PrimitiveTypes.Int64},
+ )
+ wrongBound, err := iceberg.BindExpr(
+ wrongSchema, iceberg.EqualTo(iceberg.Reference("other"),
int64(1)), true,
+ )
+ require.NoError(t, err)
+
+ tests := []struct {
+ name string
+ tasks []FileScanTask
+ wantErr bool
+ }{
+ {
+ name: "mixed plan",
+ tasks: []FileScanTask{{Residual: bound}, {}, {Residual:
unbound}, {Residual: bound}},
+ },
+ {
+ name: "invalid bound residual",
+ tasks: []FileScanTask{{Residual: bound}, {Residual:
unbound}, {Residual: wrongBound}},
+ wantErr: true,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ originals := make([]iceberg.BooleanExpression,
len(tt.tasks))
+ for i := range tt.tasks {
+ originals[i] = tt.tasks[i].Residual
+ }
+
+ // Run twice to prove the caller-owned plan remains
reusable.
+ for range 2 {
+ _, _, err := scan.ReadTasks(t.Context(),
tt.tasks)
+ if tt.wantErr {
+ require.ErrorIs(t, err,
iceberg.ErrInvalidArgument)
+ require.ErrorContains(t, err, "bind
residual for task 2")
+ require.ErrorContains(t, err, "field ID
2")
+ } else {
+ require.NoError(t, err)
+ }
+ for i, original := range originals {
+ if original == nil {
+ require.Nil(t,
tt.tasks[i].Residual)
+ } else {
+ require.Same(t, original,
tt.tasks[i].Residual)
+ }
+ }
+ }
+ })
+ }
+}
+
+func TestReadTasksConcurrentScansPreserveInputTasks(t *testing.T) {
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64},
+ )
+ location := t.TempDir()
+ metadata, err := NewMetadata(
+ schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, location,
nil,
+ )
+ require.NoError(t, err)
+ tbl := New(
+ Identifier{"db", "tbl"}, metadata, filepath.Join(location,
"metadata.json"),
+ func(context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }, nil,
+ )
+
+ bound, err := iceberg.BindExpr(schema,
iceberg.GreaterThan(iceberg.Reference("id"), int64(1)), true)
+ require.NoError(t, err)
+ unbound := iceberg.LessThan(iceberg.Reference("id"), int64(3))
+ files := []iceberg.DataFile{
+ writeResidualBindingParquetFile(t, filepath.Join(location,
"data-1.parquet"), schema, `[{"id":2}]`),
+ writeResidualBindingParquetFile(t, filepath.Join(location,
"data-2.parquet"), schema, `[{"id":3}]`),
+ }
+
+ tests := []struct {
+ name string
+ residuals []iceberg.BooleanExpression
+ wantRows int64
+ }{
+ {name: "aliased bound plan", residuals:
[]iceberg.BooleanExpression{bound, bound}, wantRows: 2},
+ {name: "mixed copy-on-write plan", residuals:
[]iceberg.BooleanExpression{bound, unbound}, wantRows: 1},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ tasks := []FileScanTask{
+ {File: files[0], Residual: tt.residuals[0]},
+ {File: files[1], Residual: tt.residuals[1]},
+ }
+ snapshot := append([]FileScanTask(nil), tasks...)
+
+ type result struct {
+ rows int64
+ err error
+ }
+ results := make(chan result, 2)
+ for range 2 {
+ go func() {
+ scan := tbl.Scan(WithMaxConcurrency(4))
+ _, records, readErr :=
scan.ReadTasks(t.Context(), tasks)
+ if readErr != nil {
+ results <- result{err: readErr}
+
+ return
+ }
+
+ var rows int64
+ for record, iterErr := range records {
+ if iterErr != nil {
+ results <- result{err:
iterErr}
+
+ return
+ }
+ rows += record.NumRows()
+ record.Release()
+ }
+ results <- result{rows: rows}
+ }()
+ }
+
+ // Join both readers before any assertion can end the
subtest and
+ // remove fixtures still in use by the other reader.
+ completed := []result{<-results, <-results}
+ for _, got := range completed {
+ require.NoError(t, got.err)
+ require.Equal(t, tt.wantRows, got.rows)
+ }
+ require.Equal(t, snapshot, tasks)
+ for i := range tasks {
+ require.Same(t, snapshot[i].Residual,
tasks[i].Residual)
+ }
+ })
+ }
+}
+
+func TestReadTasksPassesBoundResidualsToGetRecords(t *testing.T) {
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64},
+ )
+ location := t.TempDir()
+ metadata, err := NewMetadata(
+ schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, location,
nil,
+ )
+ require.NoError(t, err)
+ tbl := New(
+ Identifier{"db", "tbl"}, metadata, filepath.Join(location,
"metadata.json"),
+ func(context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }, nil,
+ )
+
+ unbound := iceberg.GreaterThan(iceberg.Reference("id"), int64(1))
+ lowerBound, err := iceberg.BindExpr(schema, unbound, true)
+ require.NoError(t, err)
+ upperBound, err := iceberg.BindExpr(schema,
iceberg.LessThan(iceberg.Reference("id"), int64(5)), true)
+ require.NoError(t, err)
+
+ // Distinct data and residuals expose both an unbound handoff and an
+ // incorrect residual-to-task index, rather than just asserting row
totals.
+ files := []iceberg.DataFile{
+ writeResidualBindingParquetFile(t, filepath.Join(location,
"data-1.parquet"), schema, `[{"id":1},{"id":2},{"id":3}]`),
+ writeResidualBindingParquetFile(t, filepath.Join(location,
"data-2.parquet"), schema, `[{"id":4},{"id":5},{"id":6}]`),
+ writeResidualBindingParquetFile(t, filepath.Join(location,
"data-3.parquet"), schema, `[{"id":7},{"id":8},{"id":9}]`),
+ }
+
+ tests := []struct {
+ name string
+ residuals []iceberg.BooleanExpression
+ wantAlias bool
+ wantIDs []int64
+ }{
+ {
+ name: "bound and nil",
+ residuals: []iceberg.BooleanExpression{lowerBound,
upperBound, nil},
+ wantAlias: true,
+ wantIDs: []int64{2, 3, 4, 7, 8, 9},
+ },
+ {
+ name: "all nil",
+ residuals: []iceberg.BooleanExpression{nil, nil, nil},
+ wantAlias: true,
+ wantIDs: []int64{1, 2, 3, 4, 5, 6, 7, 8, 9},
+ },
+ {
+ name: "mixed",
+ residuals: []iceberg.BooleanExpression{unbound,
upperBound, nil},
+ wantIDs: []int64{2, 3, 4, 7, 8, 9},
+ },
+ }
+
+ for _, tt := range tests {
+ for _, concurrency := range []int{1, 4} {
+ t.Run(fmt.Sprintf("%s/concurrency=%d", tt.name,
concurrency), func(t *testing.T) {
+ tasks := make([]FileScanTask, len(files))
+ for i, file := range files {
+ tasks[i] = FileScanTask{File: file,
Residual: tt.residuals[i]}
+ }
+
+ scan :=
tbl.Scan(WithMaxConcurrency(concurrency))
+ readerCalls := 0
+ scan.readRecords = func(reader *arrowScan, ctx
context.Context, received []FileScanTask) (*arrow.Schema,
iter.Seq2[arrow.RecordBatch, error], error) {
+ readerCalls++
+ require.Len(t, received, len(tasks))
+ // Check the actual
ReadTasks-to-GetRecords handoff, so an
+ // unconditional clone outside the
binding helper fails too.
+ require.Equal(t, tt.wantAlias,
&received[0] == &tasks[0])
+ for i, task := range received {
+ if task.Residual != nil {
+ _, bound :=
task.Residual.(iceberg.BoundPredicate)
+ require.True(t, bound,
"residual for task %d must be bound", i)
+ }
+ }
+
+ return reader.GetRecords(ctx, received)
+ }
+ _, records, err := scan.ReadTasks(t.Context(),
tasks)
+ require.NoError(t, err)
+ require.Equal(t, 1, readerCalls)
+
+ var ids []int64
+ for record, readErr := range records {
+ require.NoError(t, readErr)
+ values, ok :=
record.Column(0).(*array.Int64)
+ require.True(t, ok)
+ for i := range values.Len() {
+ ids = append(ids,
values.Value(i))
+ }
+ record.Release()
+ }
+
+ require.Equal(t, tt.wantIDs, ids)
+ for i, residual := range tt.residuals {
+ if residual == nil {
+ require.Nil(t,
tasks[i].Residual)
+ } else {
+ require.Same(t, residual,
tasks[i].Residual)
+ }
+ }
+ })
+ }
+ }
+}
diff --git a/table/scanner.go b/table/scanner.go
index 60507598d..f6bf14c35 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -582,6 +582,10 @@ type Scan struct {
// map cannot silently drop it.
arrowBatchSize int
+ // readRecords is a scan-local reader seam; nil uses
(*arrowScan).GetRecords.
+ // Tests can wrap it to observe the task slice handed to the Arrow
reader.
+ readRecords func(*arrowScan, context.Context, []FileScanTask)
(*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error)
+
reporter metrics.Reporter
}
@@ -2121,8 +2125,11 @@ type FileScanTask struct {
Start, Length int64
// Residual is the portion of the scan filter that must still be
evaluated
// 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.
+ // file metadata; nil means the caller did not provide a task residual.
Callers
+ // may supply either a bound or unbound expression; ReadTasks validates
bound
+ // residuals and binds unbound residuals against the scan's effective
schema.
+ // Residuals that mix bound and unbound predicates are rejected.
ReadTasks
+ // applies the scan's original row filter and each task residual.
Residual iceberg.BooleanExpression
// Row lineage (v3): constants used when reading to synthesize _row_id
and _last_updated_sequence_number.
@@ -2155,6 +2162,37 @@ func (scan *Scan) ToArrowRecords(ctx context.Context)
(*arrow.Schema, iter.Seq2[
return scan.ReadTasks(ctx, tasks)
}
+// bindReadTasksResiduals keeps the caller's task slice untouched. If every
+// residual is nil or already bound, the returned top-level slice aliases
tasks.
+// Otherwise it shallow-clones once, on the first residual that needs binding.
+// Nested slices remain shared in either case.
+func bindReadTasksResiduals(
+ schema *iceberg.Schema, tasks []FileScanTask, caseSensitive bool,
+) ([]FileScanTask, error) {
+ readTasks := tasks
+ clonedTasks := false
+ for i := range tasks {
+ if tasks[i].Residual == nil {
+ continue
+ }
+
+ boundResidual, changed, err := bindTaskFilter(schema,
tasks[i].Residual, caseSensitive)
+ if err != nil {
+ return nil, fmt.Errorf("bind residual for task %d: %w",
i, err)
+ }
+ if !changed {
+ continue
+ }
+ if !clonedTasks {
+ readTasks = slices.Clone(tasks)
+ clonedTasks = true
+ }
+ readTasks[i].Residual = boundResidual
+ }
+
+ return readTasks, nil
+}
+
// ReadTasks reads Arrow records from a specific set of FileScanTasks,
applying the
// scan's projection, per-task residual filters, and delete handling. This is
useful
// when the caller has already planned or selected specific tasks to read.
@@ -2167,6 +2205,13 @@ func (scan *Scan) ToArrowRecords(ctx context.Context)
(*arrow.Schema, iter.Seq2[
// reached; if no such task is processed, the file is not read and its error
is not
// returned. The returned iterator is single-use.
//
+// On success, callers must not modify tasks or its elements until
+// the range over the returned iterator has finished, including an early break.
+// If the iterator is never ranged over, callers must ensure it will not be
used
+// before modifying tasks. ReadTasks retains the caller's backing array when no
+// residual needs binding (each is nil or already bound). Any clone is shallow,
+// so nested slices (for example DeleteFiles) remain shared with the caller.
+//
// With [WithMaxConcurrency] above one, tasks are decoded in parallel and the
// batches are returned in task order. Each worker holds the decoded batches of
// at most one task until the iterator has returned them, so while an early
task
@@ -2206,19 +2251,10 @@ func (scan *Scan) ReadTasks(ctx context.Context, tasks
[]FileScanTask) (*arrow.S
}
// Bind task residuals against the schema selected by this scan, which
may
- // be an older snapshot schema rather than the table's current schema.
Keep
- // the caller's task slice untouched because the same plan may be
reused.
- readTasks := slices.Clone(tasks)
- for i := range readTasks {
- if readTasks[i].Residual == nil {
- continue
- }
-
- readTasks[i].Residual, err = bindTaskFilter(effectiveSchema,
- readTasks[i].Residual, scan.caseSensitive)
- if err != nil {
- return nil, nil, fmt.Errorf("bind residual for task %d:
%w", i, err)
- }
+ // be an older snapshot schema rather than the table's current schema.
+ readTasks, err := bindReadTasksResiduals(effectiveSchema, tasks,
scan.caseSensitive)
+ if err != nil {
+ return nil, nil, err
}
// A plan-scoped FileIO (from remote planning) takes precedence over the
@@ -2239,19 +2275,24 @@ func (scan *Scan) ReadTasks(ctx context.Context, tasks
[]FileScanTask) (*arrow.S
return nil, nil, err
}
- outSchema, records, err := (&arrowScan{
- metadata: scan.metadata,
- fs: fs,
- scanSchema: effectiveSchema,
- projectedSchema: schema,
- boundRowFilter: boundFilter,
- filterSchema: effectiveSchema,
- caseSensitive: scan.caseSensitive,
- rowLimit: scan.limit,
- options: scan.options,
- concurrency: scan.concurrency,
- arrowBatchSize: scan.arrowBatchSize,
- }).GetRecords(ctx, readTasks)
+ readRecords := scan.readRecords
+ if readRecords == nil {
+ readRecords = (*arrowScan).GetRecords
+ }
+ outSchema, records, err := readRecords(&arrowScan{
+ metadata: scan.metadata,
+ fs: fs,
+ scanSchema: effectiveSchema,
+ projectedSchema: schema,
+ boundRowFilter: boundFilter,
+ filterSchema: effectiveSchema,
+ caseSensitive: scan.caseSensitive,
+ taskResidualsBound: true,
+ rowLimit: scan.limit,
+ options: scan.options,
+ concurrency: scan.concurrency,
+ arrowBatchSize: scan.arrowBatchSize,
+ }, ctx, readTasks)
if err != nil {
// No iterator to drive cleanup on a setup error, so release
here.
if releasePlanIO != nil {
