avantgardnerio commented on code in PR #2211:
URL:
https://github.com/apache/datafusion-ballista/pull/2211#discussion_r3785982259
##########
ballista/core/proto/ballista.proto:
##########
@@ -656,6 +723,48 @@ message SuccessfulTask {
// executed `RuntimeStatsExec`; the scheduler groups by `order_by` tag
// to combine reports across tasks/executors.
repeated RuntimeStatsReport runtime_stats = 3;
+ // Finalized window-aggregate state captured during this task, one entry
+ // per (output partition, window expression, PARTITION BY group) that
+ // closed. Empty unless the plan contains an ever-expanding-frame window
+ // (`UNBOUNDED PRECEDING`), which is the only shape DataFusion will
+ // publish accumulator state for. The scheduler prefix-merges these across
+ // tasks and bakes the result into a downstream `PrefixMergeExec`.
+ //
+ // TODO: watch the size of this. Task completion is a hot, frequent
+ // message, and sketch-backed aggregates make the payload unbounded in a
+ // way row counts and quantile sketches are not — an HLL or KLL state is
+ // kilobytes per window expression per partition, and a task covering a
+ // wide partition slice carries one of each. If it stops being small,
+ // write the state as a sidecar next to the shuffle files instead, the way
+ // sort-shuffle already writes `<data>.arrow.index` beside its data
+ // (`sort_shuffle::get_index_path`), and send only a reference here. That
+ // keeps the completion message fixed-size regardless of aggregate.
Review Comment:
> HLL, KLL or TDigest state per task per window expression is not small
Agreed, and it is worth sizing. DataFusion's dense HLL is 16 KiB per sketch
([`approx_distinct.rs#L257`](https://github.com/apache/datafusion/blob/55.0.0-rc2/datafusion/functions-aggregate/src/approx_distinct.rs#L257)).
A task carries one state per (partition in its slice, aggregate window
expression), so `max_partitions_per_task` multiplies it: a 32 partition slice
with a single sketch aggregate puts 512 KiB on that task's completion message.
Two things make me think the current shape is defensible as a starting point
rather than a commitment. `RuntimeStatsExec` already ships quantile sketches
over this exact path, so this adds a second payload of a class task status
already carries rather than introducing one. And the escape hatch is cheap: the
state can be written as a sidecar beside the shuffle files, the way sort
shuffle already writes `<data>.arrow.index` next to its data, with only a
reference on the message. That keeps completion fixed size regardless of
aggregate, and it is a change to this one field rather than to the design
around it.
What I have not done is measure it. The e2e only exercises a Float64 SUM,
where the payload is a handful of bytes. If you would rather see a real number
for `approx_distinct` before this merges, I can add a test that reports the
encoded size and post it 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]
