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 a58fdf414 perf(table): limit parquet row-group stats to filter columns
(#2156)
a58fdf414 is described below
commit a58fdf414476a10686ca731e9fa383d74a68a932
Author: Minh Vu <[email protected]>
AuthorDate: Fri Oct 9 19:30:52 2026 +0200
perf(table): limit parquet row-group stats to filter columns (#2156)
Signed-off-by: Minh Vu <[email protected]>
---
table/arrow_filter_plan.go | 6 +
table/arrow_filter_plan_stats_test.go | 60 ++++++
table/arrow_scanner.go | 15 +-
table/arrow_scanner_pruning_internal_test.go | 4 +
table/internal/parquet_files.go | 20 +-
table/internal/parquet_stats_projection_test.go | 212 +++++++++++++++++++++
table/row_group_stats_filter_columns_bench_test.go | 101 ++++++++++
7 files changed, 413 insertions(+), 5 deletions(-)
diff --git a/table/arrow_filter_plan.go b/table/arrow_filter_plan.go
index 0c4227c82..7d693663b 100644
--- a/table/arrow_filter_plan.go
+++ b/table/arrow_filter_plan.go
@@ -40,6 +40,7 @@ import (
// per-row-group maps.
type compiledFileFilterPlan struct {
statsFilter iceberg.BooleanExpression
+ statsFieldIDs []int
bloomPreds []tblutils.RowGroupBloomPred
dictionaryPreds []tblutils.RowGroupDictionaryPred
@@ -370,8 +371,13 @@ func compileFileFilterPlan(
if err != nil {
return nil, err
}
+ statsFieldIDs, err := iceberg.ExtractFieldIDs(statsFilter)
+ if err != nil {
+ return nil, err
+ }
plan.statsFilter = statsFilter
+ plan.statsFieldIDs = statsFieldIDs
plan.bloomPreds = bloomPreds
plan.dictionaryPreds = dictionaryPreds
}
diff --git a/table/arrow_filter_plan_stats_test.go
b/table/arrow_filter_plan_stats_test.go
new file mode 100644
index 000000000..4be33343c
--- /dev/null
+++ b/table/arrow_filter_plan_stats_test.go
@@ -0,0 +1,60 @@
+// 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"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestCompiledFileFilterPlanTracksStatsFieldIDs(t *testing.T) {
+ schema := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2, Name: "value", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 3, Name: "payload", Type:
iceberg.PrimitiveTypes.String},
+ )
+ filter, err := iceberg.BindExpr(schema, iceberg.NewAnd(
+ iceberg.EqualTo(iceberg.Reference("id"), int64(7)),
+ iceberg.GreaterThan(iceberg.Reference("value"), int64(10)),
+ ), true)
+ require.NoError(t, err)
+
+ plan, err := compileFileFilterPlan(schema, filter, true, false, true)
+ require.NoError(t, err)
+
+ assert.ElementsMatch(t, []int{1, 2}, plan.statsFieldIDs)
+}
+
+func TestCompiledFileFilterPlanKeepsEmptyStatsFieldIDsForAlwaysFalse(t
*testing.T) {
+ schema := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ )
+ plan, err := compileFileFilterPlan(schema, iceberg.AlwaysFalse{}, true,
false, true)
+ require.NoError(t, err)
+ require.NotNil(t, plan.statsFieldIDs)
+ assert.Empty(t, plan.statsFieldIDs)
+
+ eval := plan.statsEvaluator()
+ require.NotNil(t, eval)
+ keep, err := eval(buildRowGroupMetricsMetadata(t, 1, 1,
true).RowGroup(0), nil)
+ require.NoError(t, err)
+ assert.False(t, keep)
+}
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index c7fe27198..c1ebe9fd4 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -881,6 +881,10 @@ func newPositionDeleteRowGroupTester(schema *arrow.Schema,
targets map[string]st
if err != nil {
return nil, err
}
+ statsFieldIDs, err := iceberg.ExtractFieldIDs(filter)
+ if err != nil {
+ return nil, err
+ }
statsFn, err :=
newParquetRowGroupStatsEvaluator(iceberg.PositionalDeleteSchema, filter, false)
if err != nil {
@@ -892,8 +896,9 @@ func newPositionDeleteRowGroupTester(schema *arrow.Schema,
targets map[string]st
}
return &tblutils.ParquetRowGroupTester{
- StatsFn: statsFn,
- BloomPreds: bloomPreds,
+ StatsFn: statsFn,
+ StatsFieldIDs: statsFieldIDs,
+ BloomPreds: bloomPreds,
}, nil
}
@@ -1968,6 +1973,7 @@ func (as *arrowScan) processRecordsWithPlans(
if plans != nil && plans.pruning != nil {
tester = &tblutils.ParquetRowGroupTester{
StatsFn: plans.pruning.statsEvaluator(),
+ StatsFieldIDs: plans.pruning.statsFieldIDs,
BloomPreds: plans.pruning.bloomPreds,
DictionaryPreds: plans.pruning.dictionaryPreds,
}
@@ -2004,6 +2010,10 @@ func (as *arrowScan) processRecordsWithPlans(
if err != nil {
return err
}
+ statsFieldIDs, err :=
iceberg.ExtractFieldIDs(filePruningFilter)
+ if err != nil {
+ return err
+ }
statsFn, err :=
newParquetRowGroupStatsEvaluator(fileSchema, filePruningFilter, false)
if err != nil {
@@ -2022,6 +2032,7 @@ func (as *arrowScan) processRecordsWithPlans(
tester = &tblutils.ParquetRowGroupTester{
StatsFn: statsFn,
+ StatsFieldIDs: statsFieldIDs,
BloomPreds: bloomPreds,
DictionaryPreds: dictionaryPreds,
}
diff --git a/table/arrow_scanner_pruning_internal_test.go
b/table/arrow_scanner_pruning_internal_test.go
index 4af6adc80..007e3d0d2 100644
--- a/table/arrow_scanner_pruning_internal_test.go
+++ b/table/arrow_scanner_pruning_internal_test.go
@@ -118,6 +118,7 @@ func TestProcessRecordsUsesRowGroupFilterForPruning(t
*testing.T) {
require.NoError(t, err)
require.NotNil(t, reader.tester)
+ assert.ElementsMatch(t, []int{1}, reader.tester.StatsFieldIDs)
require.Len(t, reader.tester.BloomPreds, 1)
assert.Equal(t, 1, reader.tester.BloomPreds[0].FieldID)
assert.Len(t, reader.tester.BloomPreds[0].PhysBytes, 1)
@@ -183,6 +184,7 @@ func
TestProcessRecordsRebindsRowGroupFilterToPromotedFileSchema(t *testing.T) {
require.NoError(t, err)
assert.True(t, reader.statsResult, "matching INT32 row-group stats
should be retained")
+ assert.Equal(t, []int{1}, reader.tester.StatsFieldIDs)
require.Len(t, reader.tester.BloomPreds, 1)
assert.Equal(t, []byte{1, 0, 0, 0},
reader.tester.BloomPreds[0].PhysBytes[0],
"the bloom predicate should use the INT32 file encoding")
@@ -231,6 +233,8 @@ func TestProcessRecordsDoesNotPruneMissingInitialDefault(t
*testing.T) {
require.NoError(t, err)
require.NotNil(t, reader.tester)
+ require.NotNil(t, reader.tester.StatsFieldIDs)
+ assert.Empty(t, reader.tester.StatsFieldIDs)
assert.Empty(t, reader.tester.BloomPreds,
"a missing field with an initial-default must disable bloom
pruning")
assert.Empty(t, reader.tester.DictionaryPreds,
diff --git a/table/internal/parquet_files.go b/table/internal/parquet_files.go
index 76e4d04f4..7a8b36f37 100644
--- a/table/internal/parquet_files.go
+++ b/table/internal/parquet_files.go
@@ -1914,7 +1914,11 @@ type RowGroupDictionaryPred struct {
// group pruning.
// Pass it as the tester argument to wrapPqArrowReader.GetRecords.
type ParquetRowGroupTester struct {
- StatsFn func(*metadata.RowGroupMetaData, []int) (bool, error)
+ StatsFn func(*metadata.RowGroupMetaData, []int) (bool, error)
+ // StatsFieldIDs limits StatsFn to Parquet columns referenced by the
pruning
+ // expression. Nil preserves the historical behavior of using all read
columns.
+ // A non-nil empty slice requests no column statistics.
+ StatsFieldIDs []int
BloomPreds []RowGroupBloomPred // nil = no bloom filter pass
DictionaryPreds []RowGroupDictionaryPred // nil = no dictionary filter
pass
// RangeSet indicates that Start and Length came from an explicit scan
task.
@@ -2064,9 +2068,19 @@ func (w wrapPqArrowReader) GetRecords(ctx
context.Context, cols []int, tester an
dictionaryReader *file.Reader
)
- if len(rowGroupTester.BloomPreds) > 0 ||
len(rowGroupTester.DictionaryPreds) > 0 {
+ if (rowGroupTester.StatsFn != nil &&
len(rowGroupTester.StatsFieldIDs) > 0) ||
+ len(rowGroupTester.BloomPreds) > 0 ||
len(rowGroupTester.DictionaryPreds) > 0 {
fieldIDToColIdx = buildFieldIDToColIdx(fileMeta)
}
+ statsCols := cols
+ if rowGroupTester.StatsFn != nil &&
rowGroupTester.StatsFieldIDs != nil {
+ statsCols = make([]int, 0,
len(rowGroupTester.StatsFieldIDs))
+ for _, fieldID := range rowGroupTester.StatsFieldIDs {
+ if colIdx, ok := fieldIDToColIdx[fieldID]; ok {
+ statsCols = append(statsCols, colIdx)
+ }
+ }
+ }
var dictionaryPredsByColumn map[int][]int
if len(rowGroupTester.DictionaryPreds) > 0 {
dictionaryPredsByColumn =
groupRowGroupDictionaryPredicates(
@@ -2130,7 +2144,7 @@ func (w wrapPqArrowReader) GetRecords(ctx
context.Context, cols []int, tester an
}
if use && rowGroupTester.StatsFn != nil {
var err error
- use, err = rowGroupTester.StatsFn(rgMeta, cols)
+ use, err = rowGroupTester.StatsFn(rgMeta,
statsCols)
if err != nil {
return nil, err
}
diff --git a/table/internal/parquet_stats_projection_test.go
b/table/internal/parquet_stats_projection_test.go
new file mode 100644
index 000000000..222c9f608
--- /dev/null
+++ b/table/internal/parquet_stats_projection_test.go
@@ -0,0 +1,212 @@
+// 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 internal_test
+
+import (
+ "bytes"
+ "context"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/file"
+ "github.com/apache/arrow-go/v18/parquet/metadata"
+ "github.com/apache/arrow-go/v18/parquet/schema"
+ "github.com/apache/iceberg-go/table/internal"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestParquetRowGroupStatsUseRequestedFieldIDs(t *testing.T) {
+ data := buildMultiColumnDictionaryTestParquet(t,
multiColumnDictionaryTestRowGroup{
+ ids: []int32{1},
+ categories: []string{"a"},
+ })
+
+ for _, test := range []struct {
+ name string
+ fieldIDs []int
+ wantCols []int
+ }{
+ {name: "nil uses read columns", wantCols: []int{0, 1}},
+ {name: "filter column only", fieldIDs: []int{2}, wantCols:
[]int{1}},
+ {name: "no filter columns", fieldIDs: []int{}, wantCols:
[]int{}},
+ {name: "missing field", fieldIDs: []int{999}, wantCols:
[]int{}},
+ } {
+ t.Run(test.name, func(t *testing.T) {
+ rdr := openBloomTestReader(t, data)
+ defer rdr.Close()
+
+ var calls [][]int
+ tester := &internal.ParquetRowGroupTester{
+ StatsFn: func(_ *metadata.RowGroupMetaData,
cols []int) (bool, error) {
+ calls = append(calls, append([]int{},
cols...))
+
+ return true, nil
+ },
+ StatsFieldIDs: test.fieldIDs,
+ }
+ rr, err := rdr.GetRecords(context.Background(),
[]int{0, 1}, tester)
+ require.NoError(t, err)
+ assert.Equal(t, int64(1), countRecords(t, rr))
+ require.Len(t, calls, 1)
+ assert.Equal(t, test.wantCols, calls[0])
+ })
+ }
+}
+
+func TestParquetRowGroupStatsPruneUsingUnprojectedColumn(t *testing.T) {
+ data := buildMultiColumnDictionaryTestParquet(t,
+ multiColumnDictionaryTestRowGroup{ids: []int32{9}, categories:
[]string{"drop"}},
+ multiColumnDictionaryTestRowGroup{ids: []int32{42}, categories:
[]string{"keep"}},
+ )
+ rdr := openBloomTestReader(t, data)
+ defer rdr.Close()
+
+ var survivors []internal.RowGroupSpan
+ var calls int
+ tester := &internal.ParquetRowGroupTester{
+ StatsFieldIDs: []int{2},
+ Survivors: &survivors,
+ StatsFn: func(rg *metadata.RowGroupMetaData, cols []int) (bool,
error) {
+ calls++
+ require.Equal(t, []int{1}, cols, "stats must use the
filter column, independent of projection")
+ column, err := rg.ColumnChunk(cols[0])
+ require.NoError(t, err)
+ stats, err := column.Statistics()
+ require.NoError(t, err)
+ require.NotNil(t, stats)
+
+ return bytes.Equal(stats.EncodeMin(), []byte("keep")),
nil
+ },
+ }
+ rr, err := rdr.GetRecords(t.Context(), []int{0}, tester)
+ require.NoError(t, err)
+ defer rr.Release()
+
+ require.True(t, rr.Next())
+ record := rr.RecordBatch()
+ require.Equal(t, int64(1), record.NumCols(), "pruning must not add the
filter column to the output")
+ assert.Equal(t, "id", record.Schema().Field(0).Name)
+ assert.Equal(t, []int32{42},
record.Column(0).(*array.Int32).Int32Values())
+ assert.False(t, rr.Next())
+ require.NoError(t, rr.Err())
+ assert.Equal(t, 2, calls)
+ assert.Equal(t, []internal.RowGroupSpan{{FirstRowPos: 1, NumRows: 1}},
survivors)
+}
+
+func TestParquetRowGroupStatsUseNestedFieldIDs(t *testing.T) {
+ before := schema.NewInt32Node("before", parquet.Repetitions.Required, 1)
+ target := schema.NewInt32Node("target", parquet.Repetitions.Required, 3)
+ other := schema.NewInt32Node("other", parquet.Repetitions.Required, 4)
+ nested, err := schema.NewGroupNode("nested",
parquet.Repetitions.Required,
+ schema.FieldList{target, other}, 2)
+ require.NoError(t, err)
+ after := schema.NewInt32Node("after", parquet.Repetitions.Required, 5)
+ root, err := schema.NewGroupNode("schema", parquet.Repetitions.Required,
+ schema.FieldList{before, nested, after}, -1)
+ require.NoError(t, err)
+
+ var buf bytes.Buffer
+ writer := file.NewParquetWriter(&buf, root,
+
file.WithWriterProps(parquet.NewWriterProperties(parquet.WithStats(true))))
+ rg, err := writer.AppendRowGroupChecked()
+ require.NoError(t, err)
+ for _, value := range []int32{10, 20, 30, 40} {
+ column, err := rg.NextColumn()
+ require.NoError(t, err)
+ _, err =
column.(*file.Int32ColumnChunkWriter).WriteBatch([]int32{value}, nil, nil)
+ require.NoError(t, err)
+ require.NoError(t, column.Close())
+ }
+ require.NoError(t, rg.Close())
+ require.NoError(t, writer.Close())
+
+ rdr := openBloomTestReader(t, buf.Bytes())
+ defer rdr.Close()
+
+ var calls [][]int
+ tester := &internal.ParquetRowGroupTester{
+ StatsFn: func(_ *metadata.RowGroupMetaData, cols []int) (bool,
error) {
+ calls = append(calls, append([]int{}, cols...))
+
+ return true, nil
+ },
+ StatsFieldIDs: []int{4},
+ }
+ rr, err := rdr.GetRecords(context.Background(), []int{0, 1, 2, 3},
tester)
+ require.NoError(t, err)
+ assert.Equal(t, int64(1), countRecords(t, rr))
+ require.Len(t, calls, 1)
+ assert.Equal(t, []int{2}, calls[0])
+}
+
+func TestParquetRowGroupStatsShareMappingWithBloomAndDictionaryPruning(t
*testing.T) {
+ tests := []struct {
+ name string
+ bloom bool
+ dictionary bool
+ }{
+ {name: "stats only"},
+ {name: "stats and bloom", bloom: true},
+ {name: "stats and dictionary", dictionary: true},
+ {name: "stats and both", bloom: true, dictionary: true},
+ }
+ data := buildBloomTestParquet(t, 1)
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ rdr := openBloomTestReader(t, data)
+ defer rdr.Close()
+
+ var calls [][]int
+ tester := &internal.ParquetRowGroupTester{
+ StatsFn: func(_ *metadata.RowGroupMetaData,
cols []int) (bool, error) {
+ calls = append(calls, append([]int{},
cols...))
+
+ return true, nil
+ },
+ StatsFieldIDs: []int{1},
+ }
+ if test.bloom {
+ tester.BloomPreds =
[]internal.RowGroupBloomPred{{
+ FieldID: 1,
+ PhysBytes: [][]byte{
+ int32PhysBytes(1),
+ int32PhysBytes(2),
+ },
+ }}
+ }
+ if test.dictionary {
+ tester.DictionaryPreds =
[]internal.RowGroupDictionaryPred{{
+ FieldID: 1,
+ PhysBytes: [][]byte{
+ int32PhysBytes(1),
+ int32PhysBytes(2),
+ },
+ }}
+ }
+
+ rr, err := rdr.GetRecords(context.Background(),
[]int{0}, tester)
+ require.NoError(t, err)
+ assert.Equal(t, int64(2), countRecords(t, rr))
+ require.Len(t, calls, 2)
+ assert.Equal(t, []int{0}, calls[0])
+ assert.Equal(t, []int{0}, calls[1])
+ })
+ }
+}
diff --git a/table/row_group_stats_filter_columns_bench_test.go
b/table/row_group_stats_filter_columns_bench_test.go
new file mode 100644
index 000000000..5bdae99a7
--- /dev/null
+++ b/table/row_group_stats_filter_columns_bench_test.go
@@ -0,0 +1,101 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "fmt"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+)
+
+var rowGroupStatsFilterColumnsBenchmarkSink int
+
+func BenchmarkParquetRowGroupStatsFilterColumns(b *testing.B) {
+ const rowGroups = 1024
+
+ for _, columnCount := range []int{32, 128} {
+ meta := buildRowGroupMetricsMetadata(b, rowGroups, columnCount,
true)
+ fields := make([]iceberg.NestedField, columnCount)
+ readCols := make([]int, columnCount)
+ for i := range columnCount {
+ fields[i] = iceberg.NestedField{
+ ID: i + 1, Name: fmt.Sprintf("field_%d", i),
+ Type: iceberg.PrimitiveTypes.String, Required:
true,
+ }
+ readCols[i] = i
+ }
+ schema := iceberg.NewSchema(0, fields...)
+
+ filterCounts := []int{1, 4}
+ if columnCount == 128 {
+ filterCounts = append(filterCounts, columnCount)
+ }
+ for _, filterCount := range filterCounts {
+ var filter iceberg.BooleanExpression
+ for i := range filterCount {
+ pred :=
iceberg.EqualTo(iceberg.Reference(fmt.Sprintf("field_%d", i)), "m")
+ if filter == nil {
+ filter = pred
+ } else {
+ filter = iceberg.NewAnd(filter, pred)
+ }
+ }
+ bound, err := iceberg.BindExpr(schema, filter, true)
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ for _, test := range []struct {
+ name string
+ cols []int
+ }{
+ {name: "all_read_columns", cols: readCols},
+ {name: "filter_columns", cols:
readCols[:filterCount]},
+ } {
+
b.Run(fmt.Sprintf("columns=%d/filter_columns=%d/%s", columnCount, filterCount,
test.name), func(b *testing.B) {
+ eval, err :=
newParquetRowGroupStatsEvaluator(schema, bound, false)
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ matched := 0
+ for rowGroup := range
meta.NumRowGroups() {
+ keep, err :=
eval(meta.RowGroup(rowGroup), test.cols)
+ if err != nil {
+ b.Fatal(err)
+ }
+ if keep {
+ matched++
+ }
+ }
+
rowGroupStatsFilterColumnsBenchmarkSink = matched
+ }
+ b.StopTimer()
+
+ if
rowGroupStatsFilterColumnsBenchmarkSink != rowGroups {
+ b.Fatalf("expected %d matches,
got %d", rowGroups, rowGroupStatsFilterColumnsBenchmarkSink)
+ }
+ })
+ }
+ }
+ }
+}
