andygrove commented on PR #2211: URL: https://github.com/apache/datafusion-ballista/pull/2211#issuecomment-5226665162
Design feedback, separate from the CI note above. The overall shape reads well to me, and the module header is genuinely good documentation. Splitting the global prefix merge onto the scheduler and leaving a row wise apply on the executor is the right decomposition, and the APPROX_DISTINCT case makes a convincing argument for why the `Aggregate` path has to exist. A few things I'd want settled before this grows more code on top of it. **The per row accumulator replay looks like a performance trap.** `AggregateApply::apply` calls `update_batch` on a one row slice and then `evaluate()` once per row. For SUM that's just wasteful, but the motivating cases are sketches, and that's where it gets expensive. `evaluate()` on an HLL scans every register to produce a cardinality estimate, and on TDigest or KLL it runs a quantile computation. Doing that once per row turns a linear pass into something quite a bit worse, and it re derives work the upstream BWAG already did. The APPROX_DISTINCT test proves correctness on three rows, which is exactly the size that won't surface this. Could you run it over a realistic partition before we commit to the shape? If the numbers are bad there may be a middle path where the upstream emits partial state columns and the correction stays batch at a time. **Serde is deferred, and it's the hard part.** #2255 landed its proto message and codec arm in the same PR, and I'd like this one to end up there too, since `PrefixMergeExec` can't reach an executor without them. No objection to a scaffold that defers it, I just want to flag that the remaining work isn't mechanical. `Arc<AggregateUDF>`, `Vec<Arc<dyn PhysicalExpr>>`, and `Vec<ScalarValue>` sketch state all have to cross the wire. The related question is the transport the design picks. The description says upstream state reaches the scheduler over task status. That's a hot and frequent message, and HLL, KLL or TDigest state per task per window expression is not small. Since you describe the division of labor as fixed at design time, I'd rather pressure test that choice now than after the scheduler side is built on top of it. **Partition index coupling has no guard.** Both `per_partition_state[k]` and `Scalar.offset[k]` are keyed by partition index, but the operator declares `UnspecifiedDistribution`, no required input ordering, and `maintains_input_order: true`, and `with_new_children` only re validates counts. So any rule that repartitions the input to the same partition count would silently attach each partition's offsets to the wrong rows. Wrong answers, no error. In practice the scheduler hands over a finished plan so it may never happen, but this is the "correct on one node, silently wrong once split across stages" shape that [user-personas.md](https://github.com/apache/datafusion-ballista/blob/main/docs/source/contributors-guide/user-personas.md) calls out for Persona 1, and I'd want at least a loud invariant comment on it. **Smaller things:** - `ScalarOp::Overwrite` is documented as fitting `first_value` and `last_value`. `first_value` I follow. For the cumulative frame `last_value` is just the current row's value and needs no correction at all, so overwriting every row with a single scalar would be wrong. Which frame is that aimed at? - No metrics. `PrefixMergeExec` doesn't implement `metrics()` and `ApplyStream` has no `BaselineMetrics`. Given the first point above this is the operator you'd most want timings from, and Spark shaped users lean on per operator timings for skew debugging. - Type drift is only caught by accident. If `numeric::add` promotes, or `evaluate()` returns a different type than the column it replaces, `RecordBatch::try_new` fails with an opaque arrow error rather than something naming the offending `applies[i]`. - `ScalarOp` and `WindowApply` are public enums that you say will grow. Marking both `#[non_exhaustive]` now costs nothing and saves a breaking change on the first new variant. Similarly, `FinalizedPartitionState` as a transparent `pub type` alias means swapping it for the real DataFusion type once apache/datafusion#24007 lands is a silent public API change. A newtype now would keep that swap internal. - Minor housekeeping, the PR description has the "Generated with Claude Code" footer, and `CLAUDE.md` in the repo asks us to keep that out of PRs. -- 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]
