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 d785b279 fix(table): guard DV filtering against stale row counts
(#1624)
d785b279 is described below
commit d785b279495fb19c54fc370ad660633b13f5bb6b
Author: Minh Vu <[email protected]>
AuthorDate: Wed Aug 5 20:10:59 2026 +0200
fix(table): guard DV filtering against stale row counts (#1624)
* fix(table): guard DV filtering against stale row counts
* refactor(table): share DV mask bounds logic
* perf(table): pass through rows beyond DV bounds
* chore(ci): retrigger integration checks
* test(table): cover stale DV row count boundaries
---
table/arrow_scanner.go | 52 +++++++++++++++++++++++++-------------
table/dv_scanner_read_test.go | 59 +++++++++++++++++++++++++++++++++++++++++++
2 files changed, 94 insertions(+), 17 deletions(-)
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index 027018f6..49819ea3 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -530,6 +530,10 @@ 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
+}
+
// 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.
@@ -560,16 +564,37 @@ func filterByDeletionVector(ctx context.Context, bitmap
*dv.RoaringPositionBitma
currentIdx := nextIdx
nextIdx += nrows
- // 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()
+ 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.
+ bldr := array.NewBooleanBuilder(mem)
+ defer bldr.Release()
+ bldr.Reserve(int(nrows))
+ for pos := currentIdx; pos < nextIdx; pos++ {
+ bldr.Append(deletionVectorKeepsRow(keepBits,
rowCount, pos))
+ }
+ mask := bldr.NewBooleanArray()
+ defer mask.Release()
- return compute.FilterRecordBatch(ctx, r, sliced,
compute.DefaultFilterOptions())
+ return compute.FilterRecordBatch(ctx, r, mask,
compute.DefaultFilterOptions())
}
bldr := array.NewBooleanBuilder(mem)
@@ -580,14 +605,7 @@ func filterByDeletionVector(ctx context.Context, bitmap
*dv.RoaringPositionBitma
// 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.
- if pos >= rowCount {
- bldr.Append(true)
-
- continue
- }
- // Test keep bit at absolute position pos: byte pos/8,
bit pos%8,
- // LSB-first (the layout the fast path's
array.NewBoolean reads).
- bldr.Append(keepBits[pos>>3]&(1<<(uint(pos)&7)) != 0)
+ bldr.Append(deletionVectorKeepsRow(keepBits, rowCount,
pos))
}
mask := bldr.NewBooleanArray()
defer mask.Release()
diff --git a/table/dv_scanner_read_test.go b/table/dv_scanner_read_test.go
index 54d3d4cf..b2453832 100644
--- a/table/dv_scanner_read_test.go
+++ b/table/dv_scanner_read_test.go
@@ -347,3 +347,62 @@ func TestFilterByDeletionVectorOutOfBoundsPosition(t
*testing.T) {
defer out.Release()
assert.Equal(t, []int64{0, 1, 2},
out.Column(0).(*array.Int64).Int64Values())
}
+
+func TestFilterByDeletionVectorStaleRowCount(t *testing.T) {
+ ctx := context.Background()
+ mem := memory.NewGoAllocator()
+
+ mkBatch := func(values ...int64) arrow.RecordBatch {
+ bldr := array.NewInt64Builder(mem)
+ defer bldr.Release()
+ bldr.AppendValues(values, nil)
+ col := bldr.NewArray()
+ defer col.Release()
+ schema := arrow.NewSchema([]arrow.Field{{Name: "pos", Type:
arrow.PrimitiveTypes.Int64}}, nil)
+
+ return array.NewRecordBatch(schema, []arrow.Array{col},
int64(len(values)))
+ }
+
+ bitmap := dv.NewRoaringPositionBitmap()
+ bitmap.Set(1)
+ bitmap.Set(3)
+ filter := filterByDeletionVector(ctx, bitmap, 4,
(&rowPositionSource{}).cursor())
+
+ withinCount, err := filter(mkBatch(0, 1))
+ require.NoError(t, err)
+ defer withinCount.Release()
+ assert.Equal(t, []int64{0},
withinCount.Column(0).(*array.Int64).Int64Values())
+
+ boundaryCount, err := filter(mkBatch(2, 3))
+ require.NoError(t, err)
+ defer boundaryCount.Release()
+ assert.Equal(t, []int64{2},
boundaryCount.Column(0).(*array.Int64).Int64Values())
+
+ straddlingFilter := filterByDeletionVector(ctx, bitmap, 4,
(&rowPositionSource{}).cursor())
+ straddlingPrefix, err := straddlingFilter(mkBatch(0, 1))
+ require.NoError(t, err)
+ defer straddlingPrefix.Release()
+ assert.Equal(t, []int64{0},
straddlingPrefix.Column(0).(*array.Int64).Int64Values())
+
+ straddlingCount, err := straddlingFilter(mkBatch(2, 3, 4, 5))
+ require.NoError(t, err)
+ defer straddlingCount.Release()
+ assert.Equal(t, []int64{2, 4, 5},
straddlingCount.Column(0).(*array.Int64).Int64Values())
+
+ boundaryFilter := filterByDeletionVector(ctx, bitmap, 4,
(&rowPositionSource{}).cursor())
+ boundaryBatch, err := boundaryFilter(mkBatch(0, 1, 2, 3))
+ require.NoError(t, err)
+ defer boundaryBatch.Release()
+ assert.Equal(t, []int64{0, 2},
boundaryBatch.Column(0).(*array.Int64).Int64Values())
+
+ afterBoundary, err := boundaryFilter(mkBatch(4, 5))
+ require.NoError(t, err)
+ defer afterBoundary.Release()
+ assert.Equal(t, []int64{4, 5},
afterBoundary.Column(0).(*array.Int64).Int64Values())
+
+ zeroRowCountFilter := filterByDeletionVector(ctx, bitmap, 0,
(&rowPositionSource{}).cursor())
+ zeroRowCount, err := zeroRowCountFilter(mkBatch(6, 7))
+ require.NoError(t, err)
+ defer zeroRowCount.Release()
+ assert.Equal(t, []int64{6, 7},
zeroRowCount.Column(0).(*array.Int64).Int64Values())
+}
