fix: handle bloomfilter index overflow by multi-batch read&write

[wojiaodoubao]

When create bloom filter index on a column with more than 1.07B rows, we might ran into arrow byte array offset overflow error. This pr fix it by writing index in batch.

[codecov]

Codecov Report

❌ Patch coverage is 91.96429% with 18 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
rust/lance-index/src/scalar/bloomfilter.rs	91.96%	5 Missing and 13 partials ⚠️

📢 Thoughts on this report? Let us know!

[wojiaodoubao]

Hi @jackye1995 , @BubbleCal , could you help review this when you have time, thanks very much~

[Xuanwo]

@claude review

[github-actions]

Claude Code is working…

I'll analyze this and get back to you.

View job run

[claude]

🔴 The read path in BloomFilterIndex::load bounds its batch size by row count (BLOOMFILTER_READ_BATCH_SIZE = 4096) while the write path is correctly byte-bounded by MAX_BLOOMFILTER_ARRAY_LENGTH (~i32::MAX - 1 MiB). Because read_range returns a single concatenated RecordBatch, a user with non-default number_of_items whose serialized filters exceed ~512 KiB per zone will hit 4096 × 512 KiB = 2 GiB > i32::MAX on load, reproducing the exact BinaryArray i32 offset overflow this PR fixes on the write side. The read chunking should be byte-bounded symmetric to the write side (or switch to LargeBinary).

Extended reasoning...

Summary

The write path in this PR correctly flushes when accumulated bytes would exceed MAX_BLOOMFILTER_ARRAY_LENGTH = i32::MAX - 1 MiB (see BloomFilterBatchWriter::emit). The read path, however, uses a row-count bound (BLOOMFILTER_READ_BATCH_SIZE = 4096):

for start in (0..index_file.num_rows()).step_by(BLOOMFILTER_READ_BATCH_SIZE) {
    let end = (start + BLOOMFILTER_READ_BATCH_SIZE).min(index_file.num_rows());
    let bloom_data = index_file.read_range(start..end, None).await?;
    zones.extend(Self::try_from_serialized(bloom_data)?);
}

This is asymmetric with the write side and leaves the read path vulnerable to the same arrow_array::BinaryArray i32-offset overflow that the PR sets out to fix.

Why existing code doesn't prevent it

IndexReader::read_range (see rust/lance-index/src/scalar/lance_format.rs:191-221) calls read_stream_projected with batch_size = u32::MAX and asserts batches.len() == 1. The result is a single concatenated RecordBatch whose bloom_filter_data column is a BinaryArray with i32 offsets. The write side chunking only limits the per-written-batch bytes; it does not limit what the reader concatenates. If a written file contains batches small enough individually (each ≤ 2 GiB) but together the next 4096 rows happen to cross the 2 GiB boundary, read_range will merge them into one BinaryArray and overflow.

Additionally, the per-filter size cap on write is only max_array_length (~2 GiB), and Sbbf explicitly supports up to 128 MiB per zone (BITSET_LOG2_MAX_BYTES = 27 in sbbf.rs). A user configuring a large LANCE_BLOOMFILTER_DEFAULT_NUMBER_OF_ITEMS / low probability can easily produce filters in the 512 KiB–128 MiB range.

Impact

• Default parameters (8192 items, p=0.00057 → ~16 KiB per filter) give 4096 × 16 KiB = 64 MiB, which is safe. So the primary 1.07B-row scenario this PR targets is fixed.
• Non-default configurations producing per-zone filters above ~512 KiB will hit the exact BinaryArray i32 offset overflow on load, reproducing the bug on the read path. Since the PR explicitly adds a byte cap on write to permit large filters, users can legitimately write indices that cannot be read back.

Step-by-step proof

1. A user sets LANCE_BLOOMFILTER_DEFAULT_NUMBER_OF_ITEMS (or passes params) such that each serialized SBBF is ~1 MiB.
2. Index creation succeeds: BloomFilterBatchWriter flushes every ~2000 filters, keeping each written batch under MAX_BLOOMFILTER_ARRAY_LENGTH.
3. On load, the reader issues read_range(0..4096, None). Even if the file was written as two separate batches of 2000 filters each, read_stream_projected concatenates all rows in the range into one RecordBatch (and asserts batches.len() == 1).
4. 4096 filters × ~1 MiB ≈ 4 GiB of binary data; building the concatenated BinaryArray attempts to fit the offsets into i32 and overflows, producing the same panic/error this PR is trying to eliminate.
5. At the lower threshold: 4096 × 512 KiB = 2 GiB > i32::MAX (2 147 483 647 bytes) is already sufficient to trigger the overflow.

How to fix

Mirror the write-side logic on the read path: accumulate bytes while stepping through rows and issue a new read_range call when the next row would push the accumulated size past MAX_BLOOMFILTER_ARRAY_LENGTH. Since the on-disk file already has the bytes-per-row shape available (the bloom_filter_data column length per written batch is known after write), a simpler option is to iterate over the written batches directly (one read_range per on-disk batch), or switch the schema to LargeBinary (i64 offsets), which eliminates the overflow class entirely.

Note that the new test test_bloomfilter_chunked_write_and_load uses number_of_items = 1, which produces tiny per-zone filters and thus doesn't exercise this failure mode on the read side — a regression test with larger filters or a lower effective max_array_length on the read path would surface it.

[Xuanwo]

I think the write-side batching direction is right, but this PR still does not fix the overflow invariant end-to-end.

The root constraint here is not row count. It is that bloom_filter_data is stored as Arrow Binary, whose offsets are i32. Therefore no read or write path should ever construct a BinaryArray whose concatenated bloom filter payload can exceed the i32 offset limit.

This PR fixes that on the write side by flushing batches based on serialized bloom filter bytes. However, the load path is still bounded by a fixed row count:

for start in (0..index_file.num_rows()).step_by(BLOOMFILTER_READ_BATCH_SIZE) {
    let end = (start + BLOOMFILTER_READ_BATCH_SIZE).min(index_file.num_rows());
    let bloom_data = index_file.read_range(start..end, None).await?;
    zones.extend(Self::try_from_serialized(bloom_data)?);
}

That is not a sufficient bound. With non-default bloom filter params, each serialized zone can be much larger than the default ~16 KiB. For example, 4096 zones at ~512 KiB each already crosses 2 GiB, so read_range can still reconstruct a BinaryArray that overflows even though each written batch was valid.

I think the fix should be:

1. Keep the write-side byte-bounded batching.
2. Change the load path to consume record batches without concatenating multiple physical batches into one BinaryArray, or add a byte-bounded streaming read path for this index.
3. Make the invariant explicit in tests: construct multiple written batches where each batch is below a small test byte limit, but merging two batches would exceed that limit; then verify BloomFilterIndex::load does not read them back through a path that concatenates them.
4. Optionally follow up later by making ZoneTrainer stream zone statistics into the writer, so large BloomFilter indexes do not materialize all zones in memory before writing.

So the invariant should be:

BloomFilter index serialization must be byte-bounded end-to-end; row count and logical index size are not valid bounds for Arrow BinaryArray.

Without the read-side change, this PR fixes the default 1.07B-row write failure but still allows validly-written BloomFilter indexes to fail on load.

[wojiaodoubao]

Hi @Xuanwo , thanks your detailed review! I should have thought through more carefully, I'll pay more attention next time.

I made a fix, now the read will use a safe batch size computed from number_of_items and probability.

[Xuanwo]

Thanks, this addresses my main concern.

One remaining request: could we add an end-to-end test for BloomFilterIndex::load? The new tests cover read_batch_size, but not that load actually avoids constructing an oversized BinaryArray. Ideally the test would use a small injected byte limit, write multiple batches, and verify load succeeds in a case where concatenating them would exceed that limit.

Minor: read_batch_size can use Sbbf::size_bytes() instead of to_bytes().len() to avoid allocating just to compute the size.

[Xuanwo]

This changes a public builder method signature to expose an internal Arrow byte limit. Downstream callers of the public API will fail to compile for an implementation detail that should remain internal.

[Xuanwo]

This test only injects the small byte limit into the writer, while load still uses the production limit. It can pass even if the load path regresses to a single oversized read, so CI would not protect the overflow invariant.

[zhangyue19921010]

talk with @wojiaodoubao , will take over this work through #7753

thanks for all your work @wojiaodoubao here!