adriangb commented on code in PR #24354:
URL: https://github.com/apache/datafusion/pull/24354#discussion_r3781477822
##########
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:
Heads up that after this fix nothing exercises the rebuild below. I put an
`eprintln!` on the `into_builder().with_row_groups(...)` branch and ran all
nine tests in `dynamic_row_group_pruning.rs`: it fires zero times. Every prune
drains `rg_plan` and takes the `return None` early exit. The new test is what
changes this: pre-fix the stale plan left a survivor to rebuild with, post-fix
it does not.
`cargo-mutants` on this file agrees. Mutating `pruned_count += 1` (line 365)
to `pruned_count *= 1` pins the count at 0, so the rebuild and the early exit
never run, and all nine tests still pass, because the metric increment on the
next line is a separate statement. `row_groups_pruned_dynamic_filter >= 1` is
therefore satisfied by the counter, not by any skipping having happened.
`at_boundary && ...` to `at_boundary || ...` also survives, despite
`dynamic_rg_pruner_does_not_call_into_builder_mid_row_group` existing for it.
Worth adding one test that prunes the middle and keeps the tail, asserting
on `bytes_scanned` (or the arrow reader's records-read counters) rather than on
the prune counter.
##########
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()`
Review Comment:
`peek_next_row_group()` clones the frontier including its
`Option<RowSelection>`. arrow-rs documents the cost as "O(remaining row groups
+ selectors)". With a page-index-derived selection the selector term dominates
and is paid at every boundary, so "O(remaining row groups)" understates it.
##########
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:
`Result<(), DataFusionError>` here vs `Result<()>` on `advance_rg_plan_to`
below; `datafusion_common::Result` already defaults the error type.
Also: the invariant that actually broke is stated as a comment in the `Data`
arm ("`rg_plan.front()` is the RG the decoder is about to read") and is still
only enforced at boundaries where a pruner happens to exist. A
`debug_assert_eq!` on that pop would have caught all three desyncs in tests
that already existed, without anyone having to construct the pathological
fixture, which is what @hhhizzz asked for in #24352. The cheapest upstream
version is `DecodeResult::Data { row_group_idx, reader }`, so the caller pops
to an index instead of inferring it.
##########
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:
Two passes over `rg_plan` (`iter().any()` then the pop loop) where one would
do. Minor.
Also worth knowing this guard has no coverage: mutating `e.rg_index ==
target` to `!=` here survives the suite.
##########
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()
Review Comment:
This makes the correctness of `rg_plan` conditional on a field that has
nothing to do with `rg_plan`'s definition. Today `row_group_pruner` is the only
consumer; the next one (a metric, a second rebuild trigger, per-RG reporting)
silently reintroduces #24352. Given the perf justification above is mis-costed,
I would sync unconditionally.
Bigger picture: `RgPlanEntry` is `{ rg_index: usize }`, so `rg_plan` is a
verbatim duplicate of arrow-rs's `RowGroupFrontier::row_groups`, and
`builder_from_remaining` already writes that list into the rebuilt builder as
`row_groups: Some(row_groups)` before we overwrite it with our copy. Three bugs
so far have all been "the duplicate drifted": the `reorder_by_statistics`
ordering bug, #24352, and #24355. One accessor upstream,
`remaining_row_groups() -> impl ExactSizeIterator<Item = usize>` next to the
existing `row_groups_remaining() -> usize`, would let `rg_plan`, `RgPlanEntry`,
`sync_rg_plan_to_decoder_frontier`, `advance_rg_plan_to` and the pop in the
`Data` arm all be deleted. That is @hhhizzz's suggested direction 2, and a
smaller diff than this one. Worth a 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]
