[etseidl]

Which issue does this PR close?

• Closes Expose public async bloom filter reader (metadata + AsyncFileReader) #9067

Rationale for this change

Applications with existing Parquet column metadata and an AsyncFileReader cannot currently load a Bloom filter through a public API without constructing a ParquetRecordBatchStreamBuilder or duplicating internal header parsing.

Expose the async equivalent of Sbbf::read_from_column_chunk so metadata-only pruning code can read a selected physical column's Bloom filter directly.

What changes are included in this PR?

Add Sbbf::read_from_column_chunk_async<R: AsyncFileReader + ?Sized> and delegate ParquetRecordBatchStreamBuilder::get_row_group_column_bloom_filter to it.

Are these changes tested?

Yes, new tests are added.

Are there any significant user-facing changes?

Adds to the public API, but changes no public signatures. Malformed lengths or incomplete bitsets will now return an error.

AI disclosure

AI assistance was used to author the implementation and tests.

[etseidl]

Note: this PR only creates the new async API, which does more validation than the sync version. A follow up to this will consolidate bloom filter parsing and make validation uniform.

[etseidl]

cc @alamb @adriangb, first baby step towards #6514. (Although maybe I should wait to see how #11681 plays out)

[adriangb]

Yeah there's a couple bloom filter changes in flight (I'm thinking of #11741 / #11741), there will be some conflicts. In general I would like to land:

• This change (looks good / makes sense to me)
• Add ArrowSbbf to check Arrow arrays against Parquet bloom filters #11549
• Make bloom filters / ArrowSbbf it's own crate #11681

I'm happy to do this work in any order, I guess it doesn't matter since it will be easy to resolve mechanical conflicts.

[etseidl]

Thanks @adriangb, yeah I noticed the uptick in Bloom filter interest 😉. I'm happy to help with landing the Arrow Sbbf work first and then fix any conflicts here and in #11974 before moving on to the crate pull-out.

[adriangb]

Sounds like a plan, let's do it!

[adriangb]

Token approval. I didn't find any issues besides noting the behavior changes already disclosed (which are a good thing).

[etseidl]

Ok, I'm going to mark #11549 draft so I can get #11741 ready to merge. Then we'll see how #11549 looks.

[alamb]

I plan to review this tomorrow

[alamb]

Thank you @etseidl -- I think this new API looks good to me.

What do you think about consolidating the parsing even more?

[alamb]

This looks very similar to the read_from_column_chunk method right above it According to the bots, there are still multiple copies (see details below). Apparently there are also some slight differences with validation in the copies.

What would you think about pulling the parsing logic into "PushDecoder" (and then using that same decoder for the sync and async APIs)?

Something like this

/// Decodes a Bloom filter from bytes pushed by the caller
pub struct SbbfPushDecoder { /* state machine */ }

impl SbbfPushDecoder {
    /// Returns `None` if the column chunk has no Bloom filter
    pub fn try_new(column_metadata: &ColumnChunkMetaData) -> Result<Option<Self>>;
    pub fn push_range(&mut self, range: Range<u64>, buffer: Bytes) -> Result<()>;
    pub fn push_ranges(&mut self, ranges: Vec<Range<u64>>, buffers: Vec<Bytes>) -> Result<()>;
    /// `NeedsData` once if `bloom_filter_length` is known, twice otherwise
    /// (header estimate, then the bitset)
    pub fn try_decode(&mut self) -> Result<DecodeResult<Sbbf>>;
}

Then the new method would just look like this:

let Some(mut decoder) = SbbfPushDecoder::try_new(column_metadata)? else {
    return Ok(None);
};
loop {
    match decoder.try_decode()? {
        DecodeResult::NeedsData(ranges) => {
            let buffers = reader.get_byte_ranges(ranges.clone()).await?;
            decoder.push_ranges(ranges, buffers)?;
        }
        DecodeResult::Data(sbbf) => return Ok(Some(sbbf)),
        DecodeResult::Finished => return Err(general_err!("...")),
    }
}

And we could consolidate testing as we wouldn't have multiple independent copies to test.

This would also give us an extensible API if we wanted to add more controls over what bloom filters were actually read.

Locations of bloom filter loading copies

Details

Copy: Sbbf::read_from_column_chunk (sync standalone)
On main: yes
In PR: yes, unchanged
Link:

arrow-rs/parquet/src/bloom_filter/mod.rs

Line 476 in bf0cde3

reader: &R,

────────────────────────────────────────
Copy: ParquetRecordBatchReaderBuilder::get_row_group_column_bloom_filter (sync builder)
On main: yes
In PR: yes, unchanged
Link:

arrow-rs/parquet/src/arrow/arrow_reader/mod.rs

Line 1348 in bf0cde3

pub fn get_row_group_column_bloom_filter(

────────────────────────────────────────
Copy: ParquetRecordBatchStreamBuilder::get_row_group_column_bloom_filter (async builder, inline copy)
On main: yes
In PR: no, replaced by a one-line delegation
Link:

arrow-rs/parquet/src/arrow/async_reader/mod.rs

Line 588 in bf0cde3

pub async fn get_row_group_column_bloom_filter(

────────────────────────────────────────
Copy: Sbbf::read_from_column_chunk_async (async standalone, new)
On main: no
In PR: yes
Link: https://github.com/apache/arrow-rs/blob/948390caa2/parquet/src/bloom_filter/mod.rs#L535

[alamb]

The PR basically pulls this code into a new public API for reuse which I think a good step forward. However, I have another idea for consolidation that I think could be even nicer

[alamb]

With a push decoder, you would get the ranges directly rather than having to back it out like this

[etseidl]

Thanks @alamb! I also have #11975 in flight to address many of your duplication concerns. I'll look into the push design as part of that. I imagine that will be a relatively minor tweak. Thanks for the suggestion and the review.

If you'd prefer to get it all done in one PR I'll just close #11975 and change tack here.