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 2ecacb874 perf(table): build deletion-vector masks per affected batch
(#2211)
2ecacb874 is described below
commit 2ecacb8742479f96c215883e0b3d555cae412ffe
Author: Minh Vu <[email protected]>
AuthorDate: Mon Oct 19 21:59:31 2026 +0200
perf(table): build deletion-vector masks per affected batch (#2211)
* perf(table): filter deletion vectors with batch-local keep masks
* style: satisfy lint in sparse deletion-vector masks
* style: satisfy lint in sparse deletion-vector masks
* style: satisfy lint in sparse deletion-vector masks
* style: finish nlreturn cleanup in sparse DV scan
* style: finish nlreturn cleanup in sparse DV scan
---
table/arrow_scanner.go | 89 ++++++++++++------------
table/dv/roaring_bitmap.go | 54 +++++++++++++++
table/dv/roaring_bitmap_range_bench_test.go | 54 +++++++++++++++
table/dv/roaring_bitmap_range_test.go | 102 ++++++++++++++++++++++++++++
table/dv_scan_sparse_mask_test.go | 102 ++++++++++++++++++++++++++++
5 files changed, 357 insertions(+), 44 deletions(-)
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index 98556f2f4..b417dad6e 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -1335,66 +1335,55 @@ func processPositionalDeletes(ctx context.Context,
deletes set[int64], cursor *r
}
}
-func deletionVectorKeepsRow(keepBits []byte, rowCount, pos int64) bool {
- return pos >= rowCount || keepBits[pos>>3]&(1<<(uint(pos)&7)) != 0
+func deletionVectorKeepsRow(bitmap *dv.RoaringPositionBitmap, rowCount, pos
int64) bool {
+ return pos >= rowCount || pos < 0 || !bitmap.Contains(uint64(pos))
}
-// filterByDeletionVector returns a pipeline step that drops rows present in
-// the bitmap by precomputing a bit-packed keep-mask covering the whole file
-// once and slicing the relevant range per batch into
compute.FilterRecordBatch.
-//
-// The mask layout matches Arrow's Boolean buffer convention (LSB-first per
-// byte, little-endian word order), so dv.KeepMaskBytes ->
memory.NewBufferBytes
-// -> array.NewBoolean is a zero-copy wrap. The keepBits slice is Go-allocated
-// (GC-friendly) and shared across every per-batch Boolean array; each
-// array.NewBoolean / array.NewSlice pair is released after the batch.
-//
-// rowCount bounds the mask to the data file's row count. cursor maps each
batch
-// row to its original file position: when no row group was pruned the
positions
-// are contiguous and the mask is sliced zero-copy; when pruning skipped groups
-// the emitted rows are non-contiguous, so the keep bits are gathered per row
at
-// the cursor's positions.
+// filterByDeletionVector drops deleted rows without materializing a keep-mask
+// for the whole file. Contiguous batches use a bit-packed mask for only their
+// affected positions; clean batches are returned unchanged. Pruned row groups
+// use the position cursor and build a mask only after finding a deleted row.
+// Positions beyond a stale manifest row count are retained.
func filterByDeletionVector(ctx context.Context, bitmap
*dv.RoaringPositionBitmap, rowCount int64, cursor *rowPositionCursor)
recProcessFn {
nextIdx := int64(0)
- keepBits := bitmap.KeepMaskBytes(rowCount)
- buf := memory.NewBufferBytes(keepBits)
mem := compute.GetAllocator(ctx)
return func(r arrow.RecordBatch) (arrow.RecordBatch, error) {
defer r.Release()
nrows := r.NumRows()
-
if !cursor.src.pruned() {
currentIdx := nextIdx
nextIdx += nrows
- if nextIdx <= rowCount {
- // Wrap (and slice) the shared keep-mask buffer
for this batch.
- // array.NewSlice on a Boolean array tracks the
bit-level offset,
- // so we don't need byte-aligned slicing —
currentIdx can land
- // anywhere within a byte.
- full := array.NewBoolean(int(rowCount), buf,
nil, 0)
- defer full.Release()
- sliced := array.NewSlice(full, currentIdx,
nextIdx).(*array.Boolean)
- defer sliced.Release()
-
- return compute.FilterRecordBatch(ctx, r,
sliced, compute.DefaultFilterOptions())
- }
if currentIdx >= rowCount {
r.Retain()
return r, nil
}
- // A stale manifest count can be smaller than the rows
emitted by
- // the file. Preserve rows beyond the mask rather than
slicing past
- // its bounds and panicking.
+ maskEnd := min(nextIdx, rowCount)
+ keepBits := bitmap.KeepMaskForRange(uint64(currentIdx),
uint64(maskEnd))
+ if keepBits == nil {
+ r.Retain()
+
+ return r, nil
+ }
+
+ if nextIdx <= rowCount {
+ mask := array.NewBoolean(int(nrows),
memory.NewBufferBytes(keepBits), nil, 0)
+ defer mask.Release()
+
+ return compute.FilterRecordBatch(ctx, r, mask,
compute.DefaultFilterOptions())
+ }
+
+ // A stale row count can end inside this batch. Leave
the
+ // out-of-range suffix untouched rather than slicing
past it.
bldr := array.NewBooleanBuilder(mem)
defer bldr.Release()
bldr.Reserve(int(nrows))
for pos := currentIdx; pos < nextIdx; pos++ {
- bldr.Append(deletionVectorKeepsRow(keepBits,
rowCount, pos))
+ bldr.Append(deletionVectorKeepsRow(bitmap,
rowCount, pos))
}
mask := bldr.NewBooleanArray()
defer mask.Release()
@@ -1402,16 +1391,28 @@ func filterByDeletionVector(ctx context.Context, bitmap
*dv.RoaringPositionBitma
return compute.FilterRecordBatch(ctx, r, mask,
compute.DefaultFilterOptions())
}
- bldr := array.NewBooleanBuilder(mem)
- defer bldr.Release()
- bldr.Reserve(int(nrows))
- for range nrows {
+ var bldr *array.BooleanBuilder
+ for i := range nrows {
pos := cursor.next()
- // keepBits is sized for rowCount; a pos past it means
row-group
- // metadata and the manifest's File.Count() disagree.
Keep the row
- // rather than indexing out of bounds.
- bldr.Append(deletionVectorKeepsRow(keepBits, rowCount,
pos))
+ keep := deletionVectorKeepsRow(bitmap, rowCount, pos)
+ if bldr == nil {
+ if keep {
+ continue
+ }
+ bldr = array.NewBooleanBuilder(mem)
+ bldr.Reserve(int(nrows))
+ for range i {
+ bldr.Append(true)
+ }
+ }
+ bldr.Append(keep)
+ }
+ if bldr == nil {
+ r.Retain()
+
+ return r, nil
}
+ defer bldr.Release()
mask := bldr.NewBooleanArray()
defer mask.Release()
diff --git a/table/dv/roaring_bitmap.go b/table/dv/roaring_bitmap.go
index 0a72d4085..5f5a81909 100644
--- a/table/dv/roaring_bitmap.go
+++ b/table/dv/roaring_bitmap.go
@@ -348,3 +348,57 @@ func (b *RoaringPositionBitmap) KeepMaskBytes(length
int64) []byte {
return out
}
+
+// KeepMaskForRange returns a bit-packed Arrow Boolean keep-mask for [start,
end).
+// A nil result means the range contains no deletes. Only an affected range
+// allocates a mask, so a scan does not need one bit for every row in the file.
+// The mask is indexed relative to start, including across 32-bit buckets.
+func (b *RoaringPositionBitmap) KeepMaskForRange(start, end uint64) []byte {
+ if start >= end {
+ return nil
+ }
+
+ var out []byte
+ for key, bm := range b.bitmaps {
+ base := uint64(key) << 32
+ if end <= base {
+ continue
+ }
+
+ lo := uint64(0)
+ if start > base {
+ lo = start - base
+ if lo >= 1<<32 {
+ continue
+ }
+ }
+ hi := min(end-base, uint64(1)<<32)
+ if lo >= hi || bm.CardinalityInRange(lo, hi) == 0 {
+ continue
+ }
+
+ if out == nil {
+ out = make([]byte,
bitutil.BytesForBits(int64(end-start)))
+ memory.Set(out, 0xFF)
+ }
+
+ it := bm.Iterator()
+ it.AdvanceIfNeeded(uint32(lo))
+ for it.HasNext() {
+ pos := uint64(it.Next())
+ if pos >= hi {
+ break
+ }
+ offset := base + pos - start
+ out[offset>>3] &^= byte(1 << (offset & 7))
+ }
+ }
+
+ if out != nil {
+ if pad := int64(len(out))*8 - int64(end-start); pad > 0 {
+ bitutil.SetBitsTo(out, int64(end-start), pad, false)
+ }
+ }
+
+ return out
+}
diff --git a/table/dv/roaring_bitmap_range_bench_test.go
b/table/dv/roaring_bitmap_range_bench_test.go
new file mode 100644
index 000000000..f63cfe42e
--- /dev/null
+++ b/table/dv/roaring_bitmap_range_bench_test.go
@@ -0,0 +1,54 @@
+// 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 dv
+
+import "testing"
+
+var keepMaskBenchmarkSink []byte
+
+func BenchmarkKeepMaskBatchRange(b *testing.B) {
+ const (
+ fileRows int64 = 10_000_000
+ start uint64 = 4_000_000
+ batchSize uint64 = 65_536
+ )
+ bitmap := NewRoaringPositionBitmap()
+ for i := range uint64(16) {
+ bitmap.Set(start + 1 + i*2_048)
+ }
+ bitmap.Set(uint64(fileRows - 1))
+
+ b.Run("full_file", func(b *testing.B) {
+ b.ReportAllocs()
+ for b.Loop() {
+ keepMaskBenchmarkSink = bitmap.KeepMaskBytes(fileRows)
+ }
+ })
+ b.Run("affected_batch", func(b *testing.B) {
+ b.ReportAllocs()
+ for b.Loop() {
+ keepMaskBenchmarkSink = bitmap.KeepMaskForRange(start,
start+batchSize)
+ }
+ })
+ b.Run("clean_batch", func(b *testing.B) {
+ b.ReportAllocs()
+ for b.Loop() {
+ keepMaskBenchmarkSink = bitmap.KeepMaskForRange(0,
batchSize)
+ }
+ })
+}
diff --git a/table/dv/roaring_bitmap_range_test.go
b/table/dv/roaring_bitmap_range_test.go
new file mode 100644
index 000000000..bdb8171b0
--- /dev/null
+++ b/table/dv/roaring_bitmap_range_test.go
@@ -0,0 +1,102 @@
+// 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 dv
+
+import (
+ "math/rand"
+ "testing"
+)
+
+func TestKeepMaskForRange(t *testing.T) {
+ const boundary uint64 = 1 << 32
+
+ for _, tc := range []struct {
+ name string
+ start uint64
+ length uint64
+ positions []uint64
+ }{
+ {name: "empty range", start: 100, positions: []uint64{100}},
+ {name: "clean range", start: 100, length: 9, positions:
[]uint64{99, 109}},
+ {name: "unaligned offset", start: 5, length: 9, positions:
[]uint64{5, 7, 11, 13}},
+ {name: "one byte", start: 10, length: 8, positions:
[]uint64{10, 17}},
+ {name: "across buckets", start: boundary - 3, length: 9,
positions: []uint64{boundary - 1, boundary, boundary + 3}},
+ {name: "high bucket", start: (1 << 63) + 11, length: 11,
positions: []uint64{(1 << 63) + 12, (1 << 63) + 20}},
+ {name: "dense", start: 101, length: 25},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ bitmap := NewRoaringPositionBitmap()
+ for _, pos := range tc.positions {
+ bitmap.Set(pos)
+ }
+ if tc.name == "dense" {
+ bitmap.SetRange(100, 128)
+ bitmap.RunLengthEncode()
+ }
+
+ assertKeepMaskForRange(t, bitmap, tc.start, tc.length)
+ })
+ }
+}
+
+func TestKeepMaskForRangeRandomized(t *testing.T) {
+ random := rand.New(rand.NewSource(27))
+ for _, base := range []uint64{0, 31, (1 << 32) - 50, 1 << 32, (3 << 32)
+ 50} {
+ for _, length := range []uint64{0, 1, 7, 8, 9, 63, 64, 127} {
+ bitmap := NewRoaringPositionBitmap()
+ for range 80 {
+ bitmap.Set(base + uint64(random.Intn(150)))
+ }
+ assertKeepMaskForRange(t, bitmap, base, length)
+ }
+ }
+}
+
+func assertKeepMaskForRange(t *testing.T, bitmap *RoaringPositionBitmap,
start, length uint64) {
+ t.Helper()
+ got := bitmap.KeepMaskForRange(start, start+length)
+ hasDelete := false
+ for i := range length {
+ if bitmap.Contains(start + i) {
+ hasDelete = true
+
+ break
+ }
+ }
+ if !hasDelete {
+ if got != nil {
+ t.Fatalf("clean range [%d, %d): expected nil mask, got
%v", start, start+length, got)
+ }
+
+ return
+ }
+ if want := int((length + 7) / 8); len(got) != want {
+ t.Fatalf("mask length = %d, want %d", len(got), want)
+ }
+ for i := range length {
+ keep := got[i>>3]&(1<<(i&7)) != 0
+ if want := !bitmap.Contains(start + i); keep != want {
+ t.Errorf("position %d: keep = %t, want %t", start+i,
keep, want)
+ }
+ }
+ for i := length; i < uint64(len(got)*8); i++ {
+ if got[i>>3]&(1<<(i&7)) != 0 {
+ t.Errorf("padding bit %d not cleared", i)
+ }
+ }
+}
diff --git a/table/dv_scan_sparse_mask_test.go
b/table/dv_scan_sparse_mask_test.go
new file mode 100644
index 000000000..d140dae0e
--- /dev/null
+++ b/table/dv_scan_sparse_mask_test.go
@@ -0,0 +1,102 @@
+// 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"
+ "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"
+ "github.com/apache/iceberg-go/table/dv"
+ "github.com/apache/iceberg-go/table/internal"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestDeletionVectorHugeFileCleanBatch(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+
+ bitmap := dv.NewRoaringPositionBitmap()
+ bitmap.Set(1 << 39)
+ ctx := compute.WithAllocator(context.Background(), mem)
+ filter := filterByDeletionVector(ctx, bitmap, 1<<40,
(&rowPositionSource{}).cursor())
+
+ in := checkedInt64RecordBatch(mem, 0, 1, 2)
+ out, err := filter(in)
+ require.NoError(t, err)
+ assert.Same(t, in, out, "a clean batch should pass through")
+ out.Release()
+}
+
+func TestDeletionVectorSkipsCleanBatches(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+
+ bitmap := dv.NewRoaringPositionBitmap()
+ bitmap.Set(4)
+ ctx := compute.WithAllocator(context.Background(), mem)
+ filter := filterByDeletionVector(ctx, bitmap, 9,
(&rowPositionSource{}).cursor())
+
+ for i, test := range []struct {
+ values []int64
+ want []int64
+ clean bool
+ }{
+ {values: []int64{0, 1, 2}, want: []int64{0, 1, 2}, clean: true},
+ {values: []int64{3, 4, 5}, want: []int64{3, 5}},
+ {values: []int64{6, 7, 8}, want: []int64{6, 7, 8}, clean: true},
+ } {
+ in := checkedInt64RecordBatch(mem, test.values...)
+ out, err := filter(in)
+ require.NoError(t, err, "batch %d", i)
+ if test.clean {
+ assert.Same(t, in, out, "batch %d", i)
+ }
+ assert.Equal(t, test.want,
out.Column(0).(*array.Int64).Int64Values())
+ out.Release()
+ }
+}
+
+func TestDeletionVectorPrunedBatchSkipsCleanRows(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+
+ bitmap := dv.NewRoaringPositionBitmap()
+ bitmap.Set(53)
+ ctx := compute.WithAllocator(context.Background(), mem)
+ source := &rowPositionSource{spans: []internal.RowGroupSpan{
+ {FirstRowPos: 40, NumRows: 3},
+ {FirstRowPos: 52, NumRows: 3},
+ }}
+ filter := filterByDeletionVector(ctx, bitmap, 100, source.cursor())
+
+ first := checkedInt64RecordBatch(mem, 40, 41, 42)
+ firstOut, err := filter(first)
+ require.NoError(t, err)
+ assert.Same(t, first, firstOut)
+ firstOut.Release()
+
+ second := checkedInt64RecordBatch(mem, 52, 53, 54)
+ secondOut, err := filter(second)
+ require.NoError(t, err)
+ assert.Equal(t, []int64{52, 54},
secondOut.Column(0).(*array.Int64).Int64Values())
+ secondOut.Release()
+}
