laskoviymishka commented on code in PR #1624:
URL: https://github.com/apache/iceberg-go/pull/1624#discussion_r3719814722
##########
table/arrow_scanner.go:
##########
@@ -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()
Review Comment:
This `Retain` is the one genuinely subtle line in the change — it's
counteracting the unconditional `defer r.Release()` at the top of the closure
so the caller ends up with the batch at its original refcount.
Without a comment I think a future reader could easily read it as a leak and
"fix" it by deleting the `Retain`, which turns the pending `Release` into a
use-after-free that won't surface until GC or the sanitizer. I'd drop a
one-line note right here to pin that down.
##########
table/arrow_scanner.go:
##########
@@ -530,6 +530,10 @@ func processPositionalDeletes(ctx context.Context, deletes
set[int64], cursor *r
}
}
Review Comment:
When we pulled this out of the slow-path loop it lost the bit-layout note
that used to sit above it. Could we keep a one-liner here — byte `pos>>3`, bit
`pos&7`, LSB-first (the layout `array.NewBoolean` reads in the fast path) — so
the next reader doesn't wonder whether the `uint(pos)` cast is guarding
negative positions?
##########
table/dv_scanner_read_test.go:
##########
@@ -347,3 +347,39 @@ 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)
+ bitmap.Set(3)
+ filter := filterByDeletionVector(ctx, bitmap, 4,
(&rowPositionSource{}).cursor())
Review Comment:
Two edges of the three-way split aren't pinned by these batches. The test
steps from `nextIdx=3` (case 1) straight to `nextIdx=6` (straddle), so neither
exact boundary — `nextIdx==rowCount` nor `currentIdx==rowCount` — ever gets
hit. Those conditions are `<=` / `>=`, and that's load-bearing: the `<=` is
what keeps `array.NewSlice(full, currentIdx, rowCount)` valid when a batch ends
exactly on `rowCount`. If a refactor tightened either to a strict inequality, a
batch landing on the boundary would silently take the wrong branch and still
produce the right rows here. A two-batch variant where batch 1 ends exactly at
`rowCount` and batch 2 starts there would catch that.
Separately, `rowCount=0` (a fully stale manifest) is a valid value that's
untested — every batch should fall to the pass-through branch. Worth one more
case.
##########
table/dv_scanner_read_test.go:
##########
@@ -347,3 +347,39 @@ 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)
+ bitmap.Set(3)
+ filter := filterByDeletionVector(ctx, bitmap, 4,
(&rowPositionSource{}).cursor())
+
+ 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)))
+ }
+
+ withinCount, err := filter(mkBatch(0, 1, 2))
+ require.NoError(t, err)
+ defer withinCount.Release()
+ assert.Equal(t, []int64{0, 2},
withinCount.Column(0).(*array.Int64).Int64Values())
+
+ beyondCount, err := filter(mkBatch(3, 4, 5))
Review Comment:
I think this straddle case passes even against a subtly-wrong
implementation. In `mkBatch(3, 4, 5)` the only in-range position is 3, and 3 is
deleted — so 4 and 5 (beyond `rowCount`) are the only kept rows, and the
assertion is `[]int64{4, 5}`. Nothing here exercises an in-range position that
should be *kept*: a straddle impl that just dropped every in-range row would
produce the same `[4, 5]` and pass all three assertions.
I'd reshape so the straddle batch carries an in-range keep, e.g. batch1 =
`(0,1)` [case 1], batch2 = `(2,3,4,5)` [straddle]: pos 2 in-range and kept, 3
in-range and deleted, 4/5 beyond and kept → expect `[2, 4, 5]`. That pins the
branch that actually matters.
##########
table/arrow_scanner.go:
##########
@@ -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
Review Comment:
A stale `record_count` almost always means the writer emitted more physical
rows than the manifest declares — a real defect in whatever wrote the file. We
handle it correctly, but silently, so an operator has no way to find the
offending file.
Would it be worth a one-time `slog.Warn` (data file path, `rowCount`, the
batch range) the first time we cross this boundary, guarded so it fires at most
once per file rather than per batch? `dv/deletion_vector.go` already warns on
the missing-cardinality case, so there's precedent for treating this as a
writer defect worth surfacing. wdyt?
##########
table/dv_scanner_read_test.go:
##########
@@ -347,3 +347,39 @@ 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)
+ bitmap.Set(3)
+ filter := filterByDeletionVector(ctx, bitmap, 4,
(&rowPositionSource{}).cursor())
+
+ 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)))
+ }
+
+ withinCount, err := filter(mkBatch(0, 1, 2))
+ require.NoError(t, err)
+ 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())
+
+ fullyBeyondCount, err := filter(mkBatch(6, 7))
Review Comment:
Small thing, but these two names are inverted: `beyondCount` above is
actually the straddle result (`currentIdx 3 < rowCount 4 < nextIdx 6`), while
this `mkBatch(6, 7)` case is the fully-beyond one. I'd rename to something like
`straddleResult` / `beyondResult` so the intent reads straight — the
value/position coincidence (values happening to equal absolute positions)
already makes this test a little hard to follow, so anything that names the
branch helps.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
