Repository navigation
fix: bound WindowAgg and grouped TopK output batches - #26788
koriyoshi2041 wants to merge 3 commits into
Conversation
|
Revalidated this against current
The |
kosiew
left a comment
kosiew
left a comment
There was a problem hiding this comment.
Have you considered async_try_stream + into_stream() pattern used by the other aggregate streams (e.g. single_stream.rs), which would remove the output-cursor fields and completion guards added here.
|
Yes. I traced the For I will revise the branch in that form and rerun the two focused suites. |
Closes #26772.
Problem
WindowAggExecand grouped TopK materialize their final result and currently emit it as oneRecordBatch. Large results therefore bypassdatafusion.execution.batch_sizeand make one stream poll hand an oversized batch to downstream operators.Fix
Carry the configured batch size into both streams, retain the materialized result, and emit zero-copy
RecordBatch::slicechunks across successive polls. Output metrics are recorded per emitted chunk, andWindowAggExecreleases its buffered input batches after computing the result.Tests
cargo test -p datafusion-physical-plan grouped_topk_stream::tests --libcargo test -p datafusion-physical-plan window_agg_exec::tests --libcargo clippy -p datafusion-physical-plan --lib --tests -- -D warningscargo fmt --all -- --checkgit diff --checkThe regressions assert
4/4/2and2/2/1output shapes, unchanged values, and per-chunk row/batch metrics.Risk
Low. Result computation, ordering, and values are unchanged. Slicing is zero-copy; the behavioral change is limited to output batch boundaries and the matching output-batch metric count.