alamb commented on code in PR #11165:
URL: https://github.com/apache/arrow-rs/pull/11165#discussion_r4072315377
##########
parquet/src/arrow/async_reader/mod.rs:
##########
@@ -580,6 +580,15 @@ impl<T: AsyncFileReader + Send + 'static>
ParquetRecordBatchStreamBuilder<T> {
Self::new_builder(AsyncReader(input), metadata)
}
+ /// Consume this builder and return its underlying async reader.
+ ///
+ /// The reader retains any state established by metadata or bloom filter
reads.
Review Comment:
This API seems fine to me (give back the inner reader)
However, I found this comment quite confusing as the reader doesn't know
anything about metadata or bloom filters 🤔 Is this comment trying to say that
the inner reader may have been modified by the builder (e.g. to read metadata
or bloom filters)? 🤔
##########
parquet/src/arrow/push_decoder/mod.rs:
##########
@@ -605,6 +619,36 @@ impl ParquetPushDecoder {
self.state.peek_next_row_group()
}
+ /// Preview at most two filter-free row groups using the demand range
planner.
Review Comment:
I found the limit of 2 row groups confusing here -- why can't it do more?
Also, What is the "demand range planner"? that maybe is describing some
internal detail
I think this documentation should focus on the API this function is
implementing. Perhaps something like
```rust
/// Returns an estimate of the data that will be needed for the next
`max_row_groups` row groups
///
/// Note this estimate may be larger than the range actually required, as it
does not include
/// the effects of any filters that may be applied.
///
/// (here spell out what this returns based on the current reader state)
```
##########
parquet/src/arrow/push_decoder/remaining.rs:
##########
@@ -446,6 +447,36 @@ impl RemainingRowGroups {
self.frontier.peek_next_row_group()
}
+ /// Preview only at a filter-free boundary; the cloned frontier preserves
+ /// selection and offset/limit accounting without advancing ordered demand.
+ pub fn preview_row_group_ranges(
+ &self,
+ max_row_groups: usize,
+ ) -> Result<Option<Vec<RowGroupRangePreview>>, ParquetError> {
+ if self.row_group_reader_builder.has_active_row_group() ||
self.frontier.has_predicates {
+ return Ok(None);
+ }
+ let mut frontier = self.frontier.clone();
+ let mut previews = Vec::with_capacity(max_row_groups);
+ while previews.len() < max_row_groups {
Review Comment:
this look seems to work for more than 2 next row groups 🤔
##########
parquet/src/arrow/push_decoder/mod.rs:
##########
@@ -605,6 +619,36 @@ impl ParquetPushDecoder {
self.state.peek_next_row_group()
}
+ /// Preview at most two filter-free row groups using the demand range
planner.
+ ///
+ /// Returns `None` outside a row-group boundary or when row predicates are
+ /// present; returns an empty vector when there is no selected work. This
+ /// never advances the decoder, evaluates predicates, reads data, or
creates
+ /// batch readers. Ranges fully covered by an input buffer are excluded;
Review Comment:
> Relatedly, did you consider returning all required ranges independently of
buffer coverage? The current behavior matches NeedsData, which is useful, but I
would like to understand why it is the preferred contract for a caller managing
its own read-ahead cache.
One usecase is that the caller may already have an existing caching
mechanism (e.g. that caches object_store requests)
##########
parquet/src/arrow/push_decoder/mod.rs:
##########
@@ -605,6 +619,36 @@ impl ParquetPushDecoder {
self.state.peek_next_row_group()
}
+ /// Preview at most two filter-free row groups using the demand range
planner.
+ ///
+ /// Returns `None` outside a row-group boundary or when row predicates are
+ /// present; returns an empty vector when there is no selected work. This
+ /// never advances the decoder, evaluates predicates, reads data, or
creates
+ /// batch readers. Ranges fully covered by an input buffer are excluded;
Review Comment:
> Could we make explicit that all preview entries use the same current
buffer state, without simulating buffer consumption by earlier entries?
I don't understand this comment -- are you saying that if a previous entry
consumed a buffer then it doesn't show up in this API 🤔
##########
parquet/src/arrow/push_decoder/remaining.rs:
##########
@@ -446,6 +447,36 @@ impl RemainingRowGroups {
self.frontier.peek_next_row_group()
}
+ /// Preview only at a filter-free boundary; the cloned frontier preserves
+ /// selection and offset/limit accounting without advancing ordered demand.
+ pub fn preview_row_group_ranges(
+ &self,
+ max_row_groups: usize,
+ ) -> Result<Option<Vec<RowGroupRangePreview>>, ParquetError> {
+ if self.row_group_reader_builder.has_active_row_group() ||
self.frontier.has_predicates {
+ return Ok(None);
Review Comment:
rather than returning None, I think it would be more user friendly to return
a conservative estimate of all the pages for the columns of that row group
(subject to project and selection)
##########
parquet/src/arrow/push_decoder/mod.rs:
##########
@@ -427,6 +427,20 @@ pub struct ParquetPushDecoder {
state: ParquetDecoderState,
}
+/// A metadata-only snapshot of one row group's next required byte ranges.
+///
+/// No payload or decoded arrays are retained. A caller must compare these
ranges
+/// with ordered `NeedsData` demand before consuming speculative I/O: pushing,
+/// consuming, or clearing buffered data, or rebuilding the decoder, can
+/// invalidate a snapshot.
Review Comment:
I found this documentation confusing as it refers to a bunch of internal
implementation details and doesn't seem to really explain the end user effect
or how to use this
Maybe something more like this would be better
```suggestion
/// An estimate of the input data ranges that will be needed for reading a
row group.
///
/// This structure is returned from
[`ParquetPushDecoder::preview_row_group_ranges`]
```
And then we can add details on `preview_row_group_ranges` for caveats for
how to interpret this data
##########
parquet/src/arrow/async_reader/mod.rs:
##########
@@ -1034,6 +1043,55 @@ mod tests {
}
}
+ #[tokio::test]
+ async fn test_builder_into_inner_preserves_reader_after_bloom_filter() {
+ let schema = Arc::new(Schema::new(vec![Field::new("a",
DataType::Int32, false)]));
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![Arc::new(Int32Array::from(vec![1, 2, 3]))],
+ )
+ .unwrap();
+ let mut data = Vec::new();
+ let properties = WriterProperties::builder()
+ .set_bloom_filter_enabled(true)
+ .build();
+ let mut writer = ArrowWriter::try_new(&mut data, schema,
Some(properties)).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+
+ // Cover recovery after metadata alone and after additional bloom I/O.
Review Comment:
As above I don't understand what this test is verifying. The type system
verifies that the reader is returned I thought. I feel like i am missing
something fundamental
##########
parquet/src/arrow/push_decoder/mod.rs:
##########
@@ -605,6 +619,36 @@ impl ParquetPushDecoder {
self.state.peek_next_row_group()
}
+ /// Preview at most two filter-free row groups using the demand range
planner.
+ ///
+ /// Returns `None` outside a row-group boundary or when row predicates are
Review Comment:
I agree that an example here is critical-- it would both document how to use
the API and would help the review process.
--
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]
