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 21563b22f perf(table): reuse materialized row-lineage batches (#2225)
21563b22f is described below
commit 21563b22f3c1024882c3948d823ceec71fa71b9b
Author: Minh Vu <[email protected]>
AuthorDate: Tue Oct 20 19:16:09 2026 +0200
perf(table): reuse materialized row-lineage batches (#2225)
* perf(table): reuse fully populated row-lineage batches
Skip rebuilding existing non-null lineage arrays and bulk-advance row
position
cursors. Add regression coverage and baseline-compatible benchmarks.
* fix(table): satisfy nlreturn lint in lineage benchmark
---
table/arrow_scanner.go | 71 ++++++++++++++++--
table/row_lineage_materialized_bench_test.go | 104 +++++++++++++++++++++++++++
table/row_lineage_materialized_test.go | 104 +++++++++++++++++++++++++++
3 files changed, 272 insertions(+), 7 deletions(-)
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index b417dad6e..89c7dfdb2 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -2071,6 +2071,42 @@ func (c *rowPositionCursor) next() int64 {
return pos
}
+// advance consumes rows without materializing their positions. This is used
+// when a row-id column is already complete and only cursor alignment matters.
+// Keep the cursor at an exact span boundary until the next read, matching
next.
+func (c *rowPositionCursor) advance(rows int64) {
+ if rows <= 0 {
+ return
+ }
+
+ spans := c.src.spans
+ if len(spans) == 0 {
+ c.consumed += rows
+
+ return
+ }
+
+ for c.spanIdx < len(spans)-1 {
+ remaining := spans[c.spanIdx].NumRows - c.consumed
+ if remaining <= 0 {
+ c.spanIdx++
+ c.consumed = 0
+
+ continue
+ }
+ if rows <= remaining {
+ c.consumed += rows
+
+ return
+ }
+
+ rows -= remaining
+ c.spanIdx++
+ c.consumed = 0
+ }
+ c.consumed += rows
+}
+
// synthesizeRowLineageColumns fills the requested row-lineage columns from
task
// constants: per the v3 spec a null value inherits first_row_id + position /
// data_sequence_number, a non-null value is kept. A column absent from the
batch
@@ -2092,8 +2128,10 @@ func synthesizeRowLineageColumns(
schema := batch.Schema()
nrows := batch.NumRows()
- fields := append([]arrow.Field(nil), schema.Fields()...)
- newCols := append([]arrow.Array(nil), batch.Columns()...)
+ // Copy the schema and column lists only when a value must be
synthesized.
+ // Completely materialized lineage columns can reuse the entire input
batch.
+ var fields []arrow.Field
+ var newCols []arrow.Array
var built []arrow.Array
defer func() {
for _, a := range built {
@@ -2101,19 +2139,32 @@ func synthesizeRowLineageColumns(
}
}()
- // perRow, when set, runs once for every physical row in order —
including
- // rows whose explicit value wins — so a cursor stays in lockstep with
the
- // batch even when value is never called.
+ // The row-position cursor must consume every physical row, including
+ // rows whose explicit value wins. A complete row-id column can advance
+ // the cursor in bulk; other cases call perRow for each row.
synth := func(name string, fieldID int, perRow func(k int64), value
func(k int64) int64) error {
idx := fieldIndexByID(schema, fieldID)
var existing *array.Int64
if idx >= 0 {
var ok bool
- if existing, ok = newCols[idx].(*array.Int64); !ok {
- return fmt.Errorf("row-lineage column %s is %s,
want int64", name, newCols[idx].DataType())
+ if existing, ok = batch.Column(idx).(*array.Int64); !ok
{
+ return fmt.Errorf("row-lineage column %s is %s,
want int64", name, batch.Column(idx).DataType())
}
}
+ if existing != nil && existing.NullN() == 0 {
+ // The input already contains every value. In the
row-id case,
+ // advance past the batch without reconstructing each
position.
+ if perRow != nil {
+ cursor.advance(nrows)
+ }
+ return nil
+ }
+
+ if newCols == nil {
+ fields = append([]arrow.Field(nil), schema.Fields()...)
+ newCols = append([]arrow.Array(nil), batch.Columns()...)
+ }
bldr := array.NewInt64Builder(alloc)
defer bldr.Release()
bldr.Reserve(int(nrows))
@@ -2166,6 +2217,12 @@ func synthesizeRowLineageColumns(
}
}
+ if newCols == nil {
+ batch.Retain()
+
+ return batch, nil
+ }
+
meta := schema.Metadata()
return array.NewRecordBatch(arrow.NewSchema(fields, &meta), newCols,
nrows), nil
diff --git a/table/row_lineage_materialized_bench_test.go
b/table/row_lineage_materialized_bench_test.go
new file mode 100644
index 000000000..49e47047b
--- /dev/null
+++ b/table/row_lineage_materialized_bench_test.go
@@ -0,0 +1,104 @@
+// 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/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/compute"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/iceberg-go"
+ tblutils "github.com/apache/iceberg-go/table/internal"
+)
+
+// makeMaterializedLineageBatch builds two reserved columns with optional
nulls.
+// A null index of -1 means the column is already fully populated.
+func makeMaterializedLineageBatch(alloc memory.Allocator, rows, rowIDNullAt,
seqNullAt int) arrow.RecordBatch {
+ values := make([]int64, rows)
+ sequences := make([]int64, rows)
+ for i := range values {
+ values[i] = 1_000 + int64(i)
+ sequences[i] = 10 + int64(i)
+ }
+
+ build := func(values []int64, nullAt int) arrow.Array {
+ var valid []bool
+ if nullAt >= 0 {
+ valid = make([]bool, len(values))
+ for i := range valid {
+ valid[i] = true
+ }
+ valid[nullAt] = false
+ }
+ builder := array.NewInt64Builder(alloc)
+ defer builder.Release()
+ builder.AppendValues(values, valid)
+
+ return builder.NewArray()
+ }
+
+ rowID := build(values, rowIDNullAt)
+ sequence := build(sequences, seqNullAt)
+ schema := arrow.NewSchema([]arrow.Field{
+ lineageArrowField(iceberg.RowIDColumnName,
iceberg.RowIDFieldID),
+ lineageArrowField(iceberg.LastUpdatedSequenceNumberColumnName,
iceberg.LastUpdatedSequenceNumberFieldID),
+ }, nil)
+ batch := array.NewRecordBatch(schema, []arrow.Array{rowID, sequence},
int64(rows))
+ rowID.Release()
+ sequence.Release()
+
+ return batch
+}
+
+func BenchmarkSynthesizeRowLineageColumns(b *testing.B) {
+ for _, rows := range []int{1_024, 65_536} {
+ for _, scenario := range []struct {
+ name string
+ rowNullAt, seqNullAt int
+ }{
+ {name: "complete", rowNullAt: -1, seqNullAt: -1},
+ {name: "row_id_sparse", rowNullAt: rows / 2, seqNullAt:
-1},
+ {name: "both_sparse", rowNullAt: rows / 2, seqNullAt:
rows / 4},
+ } {
+ b.Run(fmt.Sprintf("rows=%d/%s", rows, scenario.name),
func(b *testing.B) {
+ batch :=
makeMaterializedLineageBatch(memory.DefaultAllocator, rows, scenario.rowNullAt,
scenario.seqNullAt)
+ defer batch.Release()
+ firstRowID, sequence := int64(1_000), int64(10)
+ task := FileScanTask{FirstRowID: &firstRowID,
DataSequenceNumber: &sequence}
+ source := &rowPositionSource{spans:
[]tblutils.RowGroupSpan{
+ {FirstRowPos: 0, NumRows: int64(rows /
2)},
+ {FirstRowPos: int64(rows * 2), NumRows:
int64(rows / 2)},
+ }}
+ ctx := compute.WithAllocator(b.Context(),
memory.DefaultAllocator)
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ cursor := source.cursor()
+ output, err :=
synthesizeRowLineageColumns(ctx, cursor, task, batch, true, true)
+ if err != nil {
+ b.Fatal(err)
+ }
+ output.Release()
+ }
+ })
+ }
+ }
+}
diff --git a/table/row_lineage_materialized_test.go
b/table/row_lineage_materialized_test.go
new file mode 100644
index 000000000..e2d578692
--- /dev/null
+++ b/table/row_lineage_materialized_test.go
@@ -0,0 +1,104 @@
+// 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/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/compute"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ tblutils "github.com/apache/iceberg-go/table/internal"
+ "github.com/stretchr/testify/require"
+)
+
+func TestSynthesizeRowLineageFullyMaterialized(t *testing.T) {
+ alloc := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer alloc.AssertSize(t, 0)
+ batch := makeMaterializedLineageBatch(alloc, 4, -1, -1)
+ defer batch.Release()
+
+ firstRowID, sequence := int64(1_000), int64(10)
+ task := FileScanTask{FirstRowID: &firstRowID, DataSequenceNumber:
&sequence}
+ source := &rowPositionSource{spans: []tblutils.RowGroupSpan{
+ {FirstRowPos: 5, NumRows: 2},
+ {FirstRowPos: 20, NumRows: 2},
+ }}
+ cursor := source.cursor()
+ ctx := compute.WithAllocator(t.Context(), alloc)
+
+ output, err := synthesizeRowLineageColumns(ctx, cursor, task, batch,
true, true)
+ require.NoError(t, err)
+ defer output.Release()
+
+ // The caller owns an extra reference to the input batch, not a rebuild.
+ require.Same(t, batch, output)
+ require.Equal(t, int64(1_002), output.Column(0).(*array.Int64).Value(2))
+ require.Equal(t, int64(12), output.Column(1).(*array.Int64).Value(2))
+ require.Equal(t, int64(22), cursor.next()) // preserved position after
a pruned group
+}
+
+func TestSynthesizeRowLineageOnlyRebuildsNullColumn(t *testing.T) {
+ alloc := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer alloc.AssertSize(t, 0)
+ batch := makeMaterializedLineageBatch(alloc, 4, 2, -1)
+ defer batch.Release()
+
+ firstRowID, sequence := int64(1_000), int64(10)
+ task := FileScanTask{FirstRowID: &firstRowID, DataSequenceNumber:
&sequence}
+ source := &rowPositionSource{spans: []tblutils.RowGroupSpan{
+ {FirstRowPos: 5, NumRows: 2},
+ {FirstRowPos: 20, NumRows: 2},
+ }}
+ cursor := source.cursor()
+ output, err :=
synthesizeRowLineageColumns(compute.WithAllocator(t.Context(), alloc), cursor,
task, batch, true, true)
+ require.NoError(t, err)
+ defer output.Release()
+
+ require.NotSame(t, batch, output)
+ require.NotSame(t, batch.Column(0), output.Column(0))
+ require.Same(t, batch.Column(1), output.Column(1))
+ require.Equal(t, int64(1_020), output.Column(0).(*array.Int64).Value(2))
+ require.Equal(t, int64(12), output.Column(1).(*array.Int64).Value(2))
+ require.Equal(t, int64(22), cursor.next())
+}
+
+func TestRowPositionCursorAdvanceEquivalent(t *testing.T) {
+ spans := [][]tblutils.RowGroupSpan{
+ nil,
+ {{FirstRowPos: 5, NumRows: 2}},
+ {{FirstRowPos: 5, NumRows: 2}, {FirstRowPos: 20, NumRows: 3}},
+ {{FirstRowPos: 0, NumRows: 0}, {FirstRowPos: 10, NumRows: 3}},
+ {{FirstRowPos: 0, NumRows: 1}, {FirstRowPos: 10, NumRows: 0},
{FirstRowPos: 20, NumRows: 2}},
+ }
+ for _, groups := range spans {
+ for _, counts := range [][]int64{{0, 1, 0, 2, 3}, {2, 2, 1,
12}, {100}} {
+ src := &rowPositionSource{spans: groups}
+ expected, actual := src.cursor(), src.cursor()
+ for _, n := range counts {
+ for range n {
+ expected.next()
+ }
+ actual.advance(n)
+ require.Equal(t, expected.spanIdx,
actual.spanIdx, "spans=%v n=%d", groups, n)
+ require.Equal(t, expected.consumed,
actual.consumed, "spans=%v n=%d", groups, n)
+ }
+ require.Equal(t, expected.next(), actual.next(),
"spans=%v counts=%v", groups, counts)
+ }
+ }
+}
