zhuqi-lucas commented on code in PR #24354:
URL: https://github.com/apache/datafusion/pull/24354#discussion_r3781581371
##########
datafusion/datasource-parquet/src/push_decoder.rs:
##########
@@ -414,6 +428,49 @@ impl PushDecoderStreamState {
}
}
+ /// Keep `rg_plan.front()` aligned with the row group the decoder will emit
+ /// next. `try_next_reader` silently finishes row groups whose
post-predicate
+ /// selection is empty (no reader handed back), which would otherwise leave
+ /// `rg_plan` trailing the decoder by one — a later prune/rebuild would
then
+ /// re-include an already-delivered row group (#24352).
+ fn sync_rg_plan_to_decoder_frontier(&mut self) -> Result<(),
DataFusionError> {
Review Comment:
Unified to `Result<()>` in c4f56335. I tried the pop `debug_assert_eq!` too,
but it false-positived: `rg_plan.front()` is the *current* RG while
`peek_next_row_group()` returns the *next* one, so they only line up at the
boundary where `sync` runs, not at every pop. Dropped it rather than keep an
assert I could not time correctly — which really argues for your
`remaining_row_groups()` idea.
##########
datafusion/datasource-parquet/src/push_decoder.rs:
##########
@@ -414,6 +428,49 @@ impl PushDecoderStreamState {
}
}
+ /// Keep `rg_plan.front()` aligned with the row group the decoder will emit
+ /// next. `try_next_reader` silently finishes row groups whose
post-predicate
+ /// selection is empty (no reader handed back), which would otherwise leave
+ /// `rg_plan` trailing the decoder by one — a later prune/rebuild would
then
+ /// re-include an already-delivered row group (#24352).
+ fn sync_rg_plan_to_decoder_frontier(&mut self) -> Result<(),
DataFusionError> {
+ match self
+ .decoder
+ .as_ref()
+ .expect("decoder present")
+ .peek_next_row_group()
+ .map_err(DataFusionError::from)?
+ {
+ Some(actual) => self.advance_rg_plan_to(actual)?,
+ // Decoder has nothing left to emit — drain our plan so the stream
+ // finishes cleanly.
+ None => self.rg_plan.clear(),
+ }
+ Ok(())
+ }
+
+ /// Pop `rg_plan` entries until its front is `target`.
+ ///
+ /// `target` is the RG the decoder will emit next and must still be in our
+ /// plan. A missing `target` means the decoder's frontier and `rg_plan`
+ /// have diverged; we surface that as an internal error rather than
+ /// silently draining the plan, which would truncate the scan.
+ fn advance_rg_plan_to(&mut self, target: usize) -> Result<()> {
Review Comment:
Folded into a single pass in c4f56335 (pop until `target`, error if the plan
drains without finding it). And fair point that the guard itself has no
coverage.
##########
datafusion/datasource-parquet/src/push_decoder.rs:
##########
@@ -342,6 +342,20 @@ impl PushDecoderStreamState {
.as_ref()
.expect("decoder present")
.is_at_row_group_boundary();
+ // Only the runtime pruner rebuilds the decoder from `rg_plan`, so
+ // only it needs `rg_plan` kept in sync with the decoder frontier.
+ // arrow-rs silently finishes row groups whose post-predicate
+ // selection is empty without handing back a reader, so without
this
+ // sync `rg_plan` trails the decoder by one and a rebuild re-reads
an
+ // already-delivered row group (#24352). Gating on the pruner also
+ // avoids the O(remaining row groups) cost of
`peek_next_row_group()`
+ // on ordinary scans that never rebuild.
+ if at_boundary
+ && self.row_group_pruner.is_some()
+ && let Err(e) = self.sync_rg_plan_to_decoder_frontier()
+ {
+ return Some((Err(e), self));
+ }
if at_boundary && !self.rg_plan.is_empty() {
Review Comment:
Agreed — the new integration test is the first thing to actually enter that
rebuild branch (pre-fix a stale survivor was left to rebuild with; post-fix
not). The rebuild path's broader zero-coverage is worth its own follow-up.
--
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]
