2010YOUY01 commented on code in PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#discussion_r4139961214
##########
datafusion/physical-plan/src/aggregates/spill.rs:
##########
@@ -280,6 +300,7 @@ impl AggregateSpill {
.with_expressions(&spill_expr)
.with_metrics(baseline_metrics.intermediate())
.with_batch_size(batch_size)
+ .with_max_batch_bytes(Some(SPILL_BATCH_TARGET_BYTES))
.with_reservation(merge_reservation)
.with_replay_headroom()
Review Comment:
Just a rough intuition: we’ve accumulated quite a few special cases to make
memory-limited execution work. There may be an opportunity to unify and
simplify some of them.
##########
datafusion/physical-plan/src/aggregates/spill.rs:
##########
@@ -290,11 +311,141 @@ impl AggregateSpill {
partition,
merged,
&InputOrderMode::Sorted,
- baseline_metrics.clone(),
+ baseline_metrics.intermediate(),
metrics,
None,
reservation,
)?;
- Ok(replay.into_stream())
+ // Replay emits the groups of each merged batch, which may be few
+ let output = coalesce(
+ replay.into_stream(),
+ batch_size,
+ SPILL_BATCH_TARGET_BYTES,
+ baseline_metrics.elapsed_compute().clone(),
+ );
+ Ok(Box::pin(ObservedStream::new(
+ output,
+ baseline_metrics.clone(),
+ None,
+ )))
+ }
+}
+
+/// Combines consecutive small batches of `input` into batches of up to
+/// `batch_size` rows and `max_bytes`. Batches of at least half that size are
+/// passed on as they are.
+fn coalesce(
Review Comment:
IIRC, producing smaller batches is an internal issue for ordered
aggregation, and it would be better solved within the operator rather than by
adding a wrapper here.
--
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]
