You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/08/12 18:06:04 UTC
(iceberg-go) branch main updated: fix(table): apply per-task residual filters in ReadTasks (#1676)
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 a00734424 fix(table): apply per-task residual filters in ReadTasks (#1676)
a00734424 is described below
commit a00734424daae5bcef16d92c515f1bd15e105126
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Wed Aug 12 20:05:57 2026 +0200
fix(table): apply per-task residual filters in ReadTasks (#1676)
* fix(table): apply per-task scan residuals
* test(table): cover per-task residual edge cases
---
table/arrow_scanner.go | 92 ++++++++++++++----
table/projjson_roundtrip_internal_test.go | 2 +-
table/scanner.go | 29 ++++--
table/scanner_test.go | 34 +++++++
table/task_residual_test.go | 151 ++++++++++++++++++++++++++++++
5 files changed, 283 insertions(+), 25 deletions(-)
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index 28a7d3ebf..677106288 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -710,7 +710,7 @@ func collectLeafIDs(typ iceberg.Type, fieldID int, idset set[int]) {
}
}
-func (as *arrowScan) projectedFieldIDs() (set[int], error) {
+func (as *arrowScan) projectedFieldIDs(rowFilter iceberg.BooleanExpression) (set[int], error) {
idset := set[int]{}
// Collect leaf field IDs for column pruning.
// For nested types (map, list, struct), we recursively descend to find
@@ -719,8 +719,8 @@ func (as *arrowScan) projectedFieldIDs() (set[int], error) {
collectLeafIDs(field.Type, field.ID, idset)
}
- if as.boundRowFilter != nil {
- extracted, err := iceberg.ExtractFieldIDs(as.boundRowFilter)
+ if rowFilter != nil {
+ extracted, err := iceberg.ExtractFieldIDs(rowFilter)
if err != nil {
return nil, err
}
@@ -739,8 +739,8 @@ type enumeratedRecord struct {
Err error
}
-func (as *arrowScan) prepareToRead(ctx context.Context, file iceberg.DataFile) (*iceberg.Schema, []int, tblutils.FileReader, error) {
- ids, err := as.projectedFieldIDs()
+func (as *arrowScan) prepareToRead(ctx context.Context, file iceberg.DataFile, rowFilter iceberg.BooleanExpression) (*iceberg.Schema, []int, tblutils.FileReader, error) {
+ ids, err := as.projectedFieldIDs(rowFilter)
if err != nil {
return nil, nil, nil, err
}
@@ -776,12 +776,12 @@ func (as *arrowScan) prepareToRead(ctx context.Context, file iceberg.DataFile) (
return iceSchema, colIndices, rdr, nil
}
-func (as *arrowScan) getRecordFilter(ctx context.Context, fileSchema *iceberg.Schema) (recProcessFn, bool, error) {
- if as.boundRowFilter == nil || as.boundRowFilter.Equals(iceberg.AlwaysTrue{}) {
+func (as *arrowScan) getRecordFilter(ctx context.Context, fileSchema *iceberg.Schema, rowFilter iceberg.BooleanExpression) (recProcessFn, bool, error) {
+ if rowFilter == nil || rowFilter.Equals(iceberg.AlwaysTrue{}) {
return nil, false, nil
}
- translatedFilter, err := iceberg.TranslateColumnNames(as.boundRowFilter, fileSchema)
+ translatedFilter, err := iceberg.TranslateColumnNames(rowFilter, fileSchema)
if err != nil {
return nil, false, err
}
@@ -809,6 +809,53 @@ func (as *arrowScan) getRecordFilter(ctx context.Context, fileSchema *iceberg.Sc
return nil, false, nil
}
+type filterBindingState struct {
+ hasBound bool
+ hasUnbound bool
+}
+
+type filterBindingVisitor struct{}
+
+func (filterBindingVisitor) VisitTrue() filterBindingState { return filterBindingState{} }
+func (filterBindingVisitor) VisitFalse() filterBindingState { return filterBindingState{} }
+func (filterBindingVisitor) VisitNot(child filterBindingState) filterBindingState {
+ return child
+}
+func (filterBindingVisitor) VisitAnd(left, right filterBindingState) filterBindingState {
+ return filterBindingState{
+ hasBound: left.hasBound || right.hasBound,
+ hasUnbound: left.hasUnbound || right.hasUnbound,
+ }
+}
+func (filterBindingVisitor) VisitOr(left, right filterBindingState) filterBindingState {
+ return filterBindingVisitor{}.VisitAnd(left, right)
+}
+func (filterBindingVisitor) VisitUnbound(iceberg.UnboundPredicate) filterBindingState {
+ return filterBindingState{hasUnbound: true}
+}
+func (filterBindingVisitor) VisitBound(iceberg.BoundPredicate) filterBindingState {
+ return filterBindingState{hasBound: true}
+}
+
+func bindTaskFilter(schema *iceberg.Schema, filter iceberg.BooleanExpression, caseSensitive bool) (iceberg.BooleanExpression, error) {
+ if filter == nil {
+ return nil, nil
+ }
+
+ state, err := iceberg.VisitExpr(filter, filterBindingVisitor{})
+ if err != nil {
+ return nil, err
+ }
+ if state.hasBound && state.hasUnbound {
+ return nil, fmt.Errorf("%w: scan task residual mixes bound and unbound predicates", iceberg.ErrInvalidArgument)
+ }
+ if !state.hasUnbound {
+ return filter, nil
+ }
+
+ return iceberg.BindExpr(schema, filter, caseSensitive)
+}
+
// fieldIndexByID returns the index of the field carrying fieldID in its Arrow
// metadata, or -1. Resolving by reserved id (not name) matches how
// SchemaWithRowLineageColumns and projection identify lineage columns.
@@ -987,6 +1034,7 @@ func (as *arrowScan) processRecords(
ctx context.Context,
task tblutils.Enumerated[FileScanTask],
fileSchema *iceberg.Schema,
+ rowFilter iceberg.BooleanExpression,
rdr tblutils.FileReader,
columns []int,
pipeline []recProcessFn,
@@ -997,6 +1045,9 @@ func (as *arrowScan) processRecords(
testRowGroups any
recRdr array.RecordReader
)
+ if rowFilter == nil {
+ rowFilter = iceberg.AlwaysTrue{}
+ }
// Row-group stats/bloom pruning skips whole groups, so emitted batches no
// longer cover contiguous file positions. Steps that key on the original
@@ -1006,12 +1057,12 @@ func (as *arrowScan) processRecords(
// stays enabled.
switch {
case task.Value.File.FileFormat() == iceberg.ParquetFile:
- statsFn, err := newParquetRowGroupStatsEvaluator(fileSchema, as.boundRowFilter, false)
+ statsFn, err := newParquetRowGroupStatsEvaluator(fileSchema, rowFilter, false)
if err != nil {
return err
}
- bloomPreds, err := newBloomFilterPredicates(as.boundRowFilter)
+ bloomPreds, err := newBloomFilterPredicates(rowFilter)
if err != nil {
return err
}
@@ -1098,7 +1149,16 @@ func (as *arrowScan) recordsFromTask(ctx context.Context, task tblutils.Enumerat
dropFile bool
)
- iceSchema, colIndices, rdr, err = as.prepareToRead(ctx, task.Value.File)
+ rowFilter := as.boundRowFilter
+ rowFilter, err = bindTaskFilter(as.metadata.CurrentSchema(), task.Value.Residual, as.caseSensitive)
+ if task.Value.Residual == nil {
+ rowFilter = as.boundRowFilter
+ }
+ if err != nil {
+ return err
+ }
+
+ iceSchema, colIndices, rdr, err = as.prepareToRead(ctx, task.Value.File, rowFilter)
if err != nil {
return err
}
@@ -1175,7 +1235,7 @@ func (as *arrowScan) recordsFromTask(ctx context.Context, task tblutils.Enumerat
pipeline = append(pipeline, eqFn)
}
- filterFunc, dropFile, err = as.getRecordFilter(ctx, iceSchema)
+ filterFunc, dropFile, err = as.getRecordFilter(ctx, iceSchema, rowFilter)
if err != nil {
return err
}
@@ -1206,7 +1266,7 @@ func (as *arrowScan) recordsFromTask(ctx context.Context, task tblutils.Enumerat
})
})
- err = as.processRecords(ctx, task, iceSchema, rdr, colIndices, pipeline, posSource, out)
+ err = as.processRecords(ctx, task, iceSchema, rowFilter, rdr, colIndices, pipeline, posSource, out)
return err
}
@@ -1226,7 +1286,7 @@ func (as *arrowScan) producePosDeletesFromTask(ctx context.Context, task tblutil
dropFile bool
)
- iceSchema, colIndices, rdr, err = as.prepareToRead(ctx, task.Value.File)
+ iceSchema, colIndices, rdr, err = as.prepareToRead(ctx, task.Value.File, as.boundRowFilter)
if err != nil {
return err
}
@@ -1252,7 +1312,7 @@ func (as *arrowScan) producePosDeletesFromTask(ctx context.Context, task tblutil
pipeline = append(pipeline, processPositionalDeletes(ctx, deletes, posSource.cursor()))
}
- filterFunc, dropFile, err = as.getRecordFilter(ctx, iceSchema)
+ filterFunc, dropFile, err = as.getRecordFilter(ctx, iceSchema, as.boundRowFilter)
if err != nil {
return err
}
@@ -1280,7 +1340,7 @@ func (as *arrowScan) producePosDeletesFromTask(ctx context.Context, task tblutil
return ToRequestedSchema(ctx, iceberg.PositionalDeleteSchema, enrichedIcebergSchema, r, SchemaOptions{IncludeFieldIDs: true, UseLargeTypes: as.useLargeTypes})
})
- err = as.processRecords(ctx, task, iceSchema, rdr, colIndices, pipeline, posSource, out)
+ err = as.processRecords(ctx, task, iceSchema, as.boundRowFilter, rdr, colIndices, pipeline, posSource, out)
return err
}
diff --git a/table/projjson_roundtrip_internal_test.go b/table/projjson_roundtrip_internal_test.go
index c7f3cb71e..d18de1b2c 100644
--- a/table/projjson_roundtrip_internal_test.go
+++ b/table/projjson_roundtrip_internal_test.go
@@ -91,7 +91,7 @@ func TestWriteRecordsRecoversExactProjJSONCRSOnRead(t *testing.T) {
projectedSchema: tbl.Schema(),
concurrency: 1,
nameMapping: tbl.Metadata().NameMapping(),
- }).prepareToRead(ctx, dataFiles[0])
+ }).prepareToRead(ctx, dataFiles[0], nil)
require.NoError(t, err)
defer rdr.Close()
diff --git a/table/scanner.go b/table/scanner.go
index 45ac9458b..c98fb5971 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -1015,11 +1015,8 @@ type FileScanTask struct {
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
- // file metadata; nil means the caller did not provide a task residual.
- // ReadTasks currently applies the Scan's original row filter and does not
- // consume this per-task value. Remote integration must preserve that original
- // filter until per-task residual evaluation is wired; otherwise tasks read
- // outside their originating Scan could under-filter rows.
+ // file metadata; nil means the caller did not provide a task residual and
+ // ReadTasks falls back to the Scan's original row filter.
Residual iceberg.BooleanExpression
// Row lineage (v3): constants used when reading to synthesize _row_id and _last_updated_sequence_number.
@@ -1047,8 +1044,8 @@ func (scan *Scan) ToArrowRecords(ctx context.Context) (*arrow.Schema, iter.Seq2[
}
// ReadTasks reads Arrow records from a specific set of FileScanTasks, applying the
-// scan's projection, row filters, and positional delete handling. This is useful when
-// the caller has already planned or selected specific tasks to read.
+// scan's projection, per-task residual filters, and positional delete handling. This
+// is useful when the caller has already planned or selected specific tasks to read.
func (scan *Scan) ReadTasks(ctx context.Context, tasks []FileScanTask) (*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) {
if scan.selectorErr != nil {
return nil, nil, scan.selectorErr
@@ -1076,6 +1073,22 @@ func (scan *Scan) ReadTasks(ctx context.Context, tasks []FileScanTask) (*arrow.S
return nil, nil, err
}
+ // 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)
+ }
+ }
+
// A plan-scoped FileIO (from remote planning) takes precedence over the
// table's default FileIO and is closed once the returned iterator finishes.
var fs io.IO
@@ -1097,7 +1110,7 @@ func (scan *Scan) ReadTasks(ctx context.Context, tasks []FileScanTask) (*arrow.S
rowLimit: scan.limit,
options: scan.options,
concurrency: scan.concurrency,
- }).GetRecords(ctx, tasks)
+ }).GetRecords(ctx, readTasks)
if err != nil {
// No iterator to drive cleanup on a setup error, so close here.
if scan.planIO != nil {
diff --git a/table/scanner_test.go b/table/scanner_test.go
index 11c8e205d..729bdd1f2 100644
--- a/table/scanner_test.go
+++ b/table/scanner_test.go
@@ -113,6 +113,40 @@ func (s *ScannerSuite) TestScanner() {
s.Len(tasks, tt.expectedNumTasks)
})
}
+
+ s.Run("task residual", func() {
+ ctx := compute.WithAllocator(s.ctx, mem)
+ scan := tbl.Scan(table.WithRowFilter(iceberg.AlwaysTrue{}),
+ table.WithSelectedFields("number"))
+ tasks, err := scan.PlanFiles(ctx)
+ s.Require().NoError(err)
+ s.Require().Len(tasks, 1)
+
+ // A remote planner can return a residual that is narrower than the
+ // original scan filter. ReadTasks must apply it even when the caller
+ // supplies the planned tasks directly.
+ tasks[0].Residual = iceberg.GreaterThanEqual(ref, "i")
+ _, itr, err := scan.ReadTasks(ctx, tasks)
+ s.Require().NoError(err)
+
+ next, stop := iter.Pull2(itr)
+ defer stop()
+ rec, err, valid := next()
+ s.Require().True(valid)
+ s.Require().NoError(err)
+ defer rec.Release()
+
+ arr, _, err := array.FromJSON(mem, arrow.PrimitiveTypes.Int32, strings.NewReader(`[10, 11, 12]`))
+ s.Require().NoError(err)
+ defer arr.Release()
+ expected := array.NewRecord(expectedSchema, []arrow.Array{arr}, int64(arr.Len()))
+ defer expected.Release()
+ s.True(array.RecordEqual(expected, rec), "task residual must filter the planned task")
+
+ _, err, valid = next()
+ s.Require().NoError(err)
+ s.Require().False(valid)
+ })
}
func (s *ScannerSuite) TestScannerWithDeletes() {
diff --git a/table/task_residual_test.go b/table/task_residual_test.go
new file mode 100644
index 000000000..877c6db3e
--- /dev/null
+++ b/table/task_residual_test.go
@@ -0,0 +1,151 @@
+// 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_test
+
+import (
+ "context"
+ "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/iceberg-go"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/require"
+)
+
+func TestReadTasksUsesTaskResidual(t *testing.T) {
+ ctx := context.Background()
+ tbl := newV3RowLineageTestTable(t)
+
+ arrowSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false},
+ {Name: "data", Type: arrow.BinaryTypes.String, Nullable: true},
+ }, nil)
+ data, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema, []string{
+ `[{"id":1,"data":"a"},{"id":2,"data":"b"},{"id":3,"data":"c"}]`,
+ })
+ require.NoError(t, err)
+ t.Cleanup(data.Release)
+
+ tbl, err = tbl.Append(ctx, array.NewTableReader(data, -1), nil)
+ require.NoError(t, err)
+
+ scan := tbl.Scan(table.WithRowFilter(iceberg.AlwaysTrue{}), table.WithSelectedFields("id"))
+ tasks, err := scan.PlanFiles(ctx)
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+
+ // This is the unbound residual a remote planner would return. It is
+ // narrower than the original scan filter and must be applied by ReadTasks.
+ tasks[0].Residual = iceberg.GreaterThan(iceberg.Reference("id"), int64(1))
+ originalResidual := tasks[0].Residual
+ _, records, err := scan.ReadTasks(ctx, tasks)
+ require.NoError(t, err)
+ require.Same(t, originalResidual, tasks[0].Residual,
+ "ReadTasks must not replace the caller's task residual")
+
+ var got []int64
+ for record, err := range records {
+ require.NoError(t, err)
+ values := record.Column(0).(*array.Int64)
+ for i := 0; i < values.Len(); i++ {
+ got = append(got, values.Value(i))
+ }
+ record.Release()
+ }
+
+ require.Equal(t, []int64{2, 3}, got)
+}
+
+func TestReadTasksAppliesDifferentResidualsPerTask(t *testing.T) {
+ ctx := context.Background()
+ tbl := newV3RowLineageTestTable(t)
+
+ arrowSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false},
+ {Name: "data", Type: arrow.BinaryTypes.String, Nullable: true},
+ }, nil)
+ data, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema, []string{
+ `[{"id":1,"data":"a"},{"id":2,"data":"b"},{"id":3,"data":"c"}]`,
+ })
+ require.NoError(t, err)
+ t.Cleanup(data.Release)
+
+ tbl, err = tbl.Append(ctx, array.NewTableReader(data, -1), nil)
+ require.NoError(t, err)
+
+ scan := tbl.Scan(table.WithRowFilter(iceberg.AlwaysTrue{}), table.WithSelectedFields("id"))
+ tasks, err := scan.PlanFiles(ctx)
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+
+ first := tasks[0]
+ second := tasks[0]
+ first.Residual = iceberg.LessThan(iceberg.Reference("id"), int64(2))
+ second.Residual = iceberg.GreaterThan(iceberg.Reference("id"), int64(2))
+ readTasks := []table.FileScanTask{first, second}
+
+ _, records, err := scan.ReadTasks(ctx, readTasks)
+ require.NoError(t, err)
+
+ var got []int64
+ for record, err := range records {
+ require.NoError(t, err)
+ values := record.Column(0).(*array.Int64)
+ for i := 0; i < values.Len(); i++ {
+ got = append(got, values.Value(i))
+ }
+ record.Release()
+ }
+
+ require.Equal(t, []int64{1, 3}, got)
+}
+
+func TestReadTasksAlwaysFalseResidualCompletes(t *testing.T) {
+ ctx := context.Background()
+ tbl := newV3RowLineageTestTable(t)
+
+ arrowSchema := arrow.NewSchema([]arrow.Field{{
+ Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false,
+ }}, nil)
+ data, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema, []string{
+ `[{"id":1},{"id":2}]`,
+ })
+ require.NoError(t, err)
+ t.Cleanup(data.Release)
+
+ tbl, err = tbl.Append(ctx, array.NewTableReader(data, -1), nil)
+ require.NoError(t, err)
+
+ scan := tbl.Scan(table.WithRowFilter(iceberg.AlwaysTrue{}), table.WithSelectedFields("id"))
+ tasks, err := scan.PlanFiles(ctx)
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ tasks[0].Residual = iceberg.AlwaysFalse{}
+
+ _, records, err := scan.ReadTasks(ctx, tasks)
+ require.NoError(t, err)
+ count := 0
+ for record, err := range records {
+ require.NoError(t, err)
+ count += int(record.NumRows())
+ record.Release()
+ }
+ require.Zero(t, count)
+}