alamb commented on code in PR #11953:
URL: https://github.com/apache/arrow-rs/pull/11953#discussion_r4431521988
##########
parquet/src/bloom_filter/mod.rs:
##########
@@ -520,6 +522,82 @@ impl Sbbf {
Ok(Some(Self::new(&bitset)))
}
+ /// Read a Bloom filter asynchronously from existing column-chunk metadata.
+ ///
+ /// The metadata and reader must refer to the same file.
+ ///
+ /// A missing Bloom filter offset returns `None` without performing I/O.
Invalid
+ /// metadata, malformed filters, and read failures return an error.
+ ///
+ /// When the total filter length is present, one range is fetched.
Otherwise,
+ /// this may perform additional fetches.
+ #[cfg(all(feature = "arrow", feature = "async"))]
+ pub async fn read_from_column_chunk_async<R: AsyncFileReader + ?Sized>(
Review Comment:
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
```rust
/// 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:
```rust
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:
https://github.com/apache/arrow-rs/blob/bf0cde345b14e5abdf3e8abb57f55d43da48b49f/parquet/src/bloom_filter/mod.rs#L476
────────────────────────────────────────
Copy: ParquetRecordBatchReaderBuilder::get_row_group_column_bloom_filter
(sync builder)
On main: yes
In PR: yes, unchanged
Link:
https://github.com/apache/arrow-rs/blob/bf0cde345b14e5abdf3e8abb57f55d43da48b49f/parquet/src/arrow/arrow_reader/mod.rs#L1348
────────────────────────────────────────
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:
https://github.com/apache/arrow-rs/blob/bf0cde345b14e5abdf3e8abb57f55d43da48b49f/parquet/src/arrow/async_reader/mod.rs#L588
────────────────────────────────────────
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
</details>
##########
parquet/src/arrow/async_reader/mod.rs:
##########
@@ -593,63 +590,7 @@ impl<T: AsyncFileReader + Send + 'static>
ParquetRecordBatchStreamBuilder<T> {
let metadata = self.metadata.row_group(row_group_idx);
let column_metadata = metadata.column(column_idx);
- let offset: u64 = if let Some(offset) =
column_metadata.bloom_filter_offset() {
Review Comment:
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
##########
parquet/tests/arrow_reader/bloom_filter/async.rs:
##########
@@ -79,4 +154,256 @@ async fn test_get_row_group_column_bloom_filter(data:
Bytes, with_length: bool)
.unwrap();
assert!(sbbf.check(&"Hello"));
assert!(!sbbf.check(&"Hello_Not_Exists"));
+ assert_eq!(sbbf.num_blocks(), standalone.num_blocks());
+ // Delegation must preserve the same Bloom-only ranges, with or without a
length.
+ assert_eq!(
+ *builder_requests.lock().unwrap(),
+ *reader.requests().lock().unwrap()
+ );
+ assert_eq!(
+ builder_requests.lock().unwrap().len(),
+ if with_length { 1 } else { 2 }
+ );
+}
+
+#[derive(Default)]
+struct RecordingReader {
Review Comment:
With a push decoder, you would get the ranges directly rather than having to
back it out like this
--
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]
