From 45ca19d4272e867cb73ca9243d48cd6d144c9d05 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sat, 1 Aug 2026 19:00:03 +0200 Subject: [PATCH 1/4] fix(table): guard DV filtering against stale row counts --- table/arrow_scanner.go | 40 ++++++++++++++++++++++++++--------- table/dv_scanner_read_test.go | 22 +++++++++++++++++++ 2 files changed, 52 insertions(+), 10 deletions(-) diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go index 027018f6a..3f12321ae 100644 --- a/table/arrow_scanner.go +++ b/table/arrow_scanner.go @@ -560,16 +560,36 @@ 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() - - return compute.FilterRecordBatch(ctx, r, sliced, compute.DefaultFilterOptions()) + 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()) + } + + // 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++ { + if pos >= rowCount { + bldr.Append(true) + } else { + bldr.Append(keepBits[pos>>3]&(1<<(uint(pos)&7)) != 0) + } + } + mask := bldr.NewBooleanArray() + defer mask.Release() + + return compute.FilterRecordBatch(ctx, r, mask, compute.DefaultFilterOptions()) } bldr := array.NewBooleanBuilder(mem) diff --git a/table/dv_scanner_read_test.go b/table/dv_scanner_read_test.go index 54d3d4cf2..301188b73 100644 --- a/table/dv_scanner_read_test.go +++ b/table/dv_scanner_read_test.go @@ -347,3 +347,25 @@ 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() + + bitmap := dv.NewRoaringPositionBitmap() + bitmap.Set(1) + filter := filterByDeletionVector(ctx, bitmap, 2, (&rowPositionSource{}).cursor()) + + bldr := array.NewInt64Builder(mem) + defer bldr.Release() + bldr.AppendValues([]int64{0, 1, 2}, nil) + col := bldr.NewArray() + defer col.Release() + schema := arrow.NewSchema([]arrow.Field{{Name: "pos", Type: arrow.PrimitiveTypes.Int64}}, nil) + batch := array.NewRecordBatch(schema, []arrow.Array{col}, 3) + + out, err := filter(batch) + require.NoError(t, err) + defer out.Release() + assert.Equal(t, []int64{0, 2}, out.Column(0).(*array.Int64).Int64Values()) +} From ea422b372c36617c108009448a6c719af9d267d3 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sat, 1 Aug 2026 19:08:23 +0200 Subject: [PATCH 2/4] refactor(table): share DV mask bounds logic --- table/arrow_scanner.go | 19 ++++++------------- table/dv_scanner_read_test.go | 31 ++++++++++++++++++++----------- 2 files changed, 26 insertions(+), 24 deletions(-) diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go index 3f12321ae..a1e53a06b 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. @@ -580,11 +584,7 @@ func filterByDeletionVector(ctx context.Context, bitmap *dv.RoaringPositionBitma defer bldr.Release() bldr.Reserve(int(nrows)) for pos := currentIdx; pos < nextIdx; pos++ { - if pos >= rowCount { - bldr.Append(true) - } else { - bldr.Append(keepBits[pos>>3]&(1<<(uint(pos)&7)) != 0) - } + bldr.Append(deletionVectorKeepsRow(keepBits, rowCount, pos)) } mask := bldr.NewBooleanArray() defer mask.Release() @@ -600,14 +600,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 301188b73..dd06e94ba 100644 --- a/table/dv_scanner_read_test.go +++ b/table/dv_scanner_read_test.go @@ -354,18 +354,27 @@ func TestFilterByDeletionVectorStaleRowCount(t *testing.T) { bitmap := dv.NewRoaringPositionBitmap() bitmap.Set(1) - filter := filterByDeletionVector(ctx, bitmap, 2, (&rowPositionSource{}).cursor()) + bitmap.Set(3) + filter := filterByDeletionVector(ctx, bitmap, 4, (&rowPositionSource{}).cursor()) - bldr := array.NewInt64Builder(mem) - defer bldr.Release() - bldr.AppendValues([]int64{0, 1, 2}, nil) - col := bldr.NewArray() - defer col.Release() - schema := arrow.NewSchema([]arrow.Field{{Name: "pos", Type: arrow.PrimitiveTypes.Int64}}, nil) - batch := array.NewRecordBatch(schema, []arrow.Array{col}, 3) + 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) - out, err := filter(batch) + return array.NewRecordBatch(schema, []arrow.Array{col}, int64(len(values))) + } + + withinCount, err := filter(mkBatch(0, 1, 2)) require.NoError(t, err) - defer out.Release() - assert.Equal(t, []int64{0, 2}, out.Column(0).(*array.Int64).Int64Values()) + defer withinCount.Release() + assert.Equal(t, []int64{0, 2}, withinCount.Column(0).(*array.Int64).Int64Values()) + + beyondCount, err := filter(mkBatch(3, 4, 5)) + require.NoError(t, err) + defer beyondCount.Release() + assert.Equal(t, []int64{4, 5}, beyondCount.Column(0).(*array.Int64).Int64Values()) } From 7852abd08425946b7649ad0548e5b815fcbdc9f1 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sat, 1 Aug 2026 19:11:35 +0200 Subject: [PATCH 3/4] perf(table): pass through rows beyond DV bounds --- table/arrow_scanner.go | 5 +++++ table/dv_scanner_read_test.go | 5 +++++ 2 files changed, 10 insertions(+) diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go index a1e53a06b..49819ea3e 100644 --- a/table/arrow_scanner.go +++ b/table/arrow_scanner.go @@ -576,6 +576,11 @@ func filterByDeletionVector(ctx context.Context, bitmap *dv.RoaringPositionBitma 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 diff --git a/table/dv_scanner_read_test.go b/table/dv_scanner_read_test.go index dd06e94ba..13ffa9cd3 100644 --- a/table/dv_scanner_read_test.go +++ b/table/dv_scanner_read_test.go @@ -377,4 +377,9 @@ func TestFilterByDeletionVectorStaleRowCount(t *testing.T) { require.NoError(t, err) defer beyondCount.Release() assert.Equal(t, []int64{4, 5}, beyondCount.Column(0).(*array.Int64).Int64Values()) + + fullyBeyondCount, err := filter(mkBatch(6, 7)) + require.NoError(t, err) + defer fullyBeyondCount.Release() + assert.Equal(t, []int64{6, 7}, fullyBeyondCount.Column(0).(*array.Int64).Int64Values()) } From dabefc1342cb1340e01e3af37845d4980dea677a Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sun, 2 Aug 2026 15:13:45 +0200 Subject: [PATCH 4/4] chore(ci): retrigger integration checks