avantgardnerio opened a new pull request, #2250: URL: https://github.com/apache/datafusion-ballista/pull/2250
## Summary Both `OrderedRangeRepartitionExec` and `RuntimeStatsExec` were returning `None` from `metrics()`, so they were invisible in the scheduler's stage-metrics dump. Beyond just missing signal, this caused a display-walker misalignment inside `RuntimeStatsExec`: `collect_plan_metrics` skips `None` returns while the display visitor increments its index for every node, so metrics for later operators (SortExec, DataSourceExec) got labeled as the wrong op. Both ops are additive here — no behavior change, no execution-path change, just wiring in the standard `ExecutionPlanMetricsSet` and returning it from `metrics()`. ## `RuntimeStatsExec` Adds `ExecutionPlanMetricsSet` with: - `elapsed_compute` / `output_rows` via `BaselineMetrics.record_output` - `sketch_time` (subset time) — isolates the T-Digest `merge_unsorted_f64` cost from surrounding evaluate/downcast/flatten, so sketch-mode overhead is measurable separately. - `sketch_batches` (Count) — number of batches where the sketch was actually updated (excludes empty-after-nulls batches). Timer scoped post-child-poll so upstream shuffle IO / parquet reads aren't billed to `elapsed_compute`. ## `OrderedRangeRepartitionExec` `OrderedRangeRepartitionExec` was constructing `BaselineMetrics` for its `StreamingMerge` output side but never exposing them (no `metrics()` impl), and the scatter side — where the actual value-add of range-routing lives — had zero instrumentation. Adds `fn metrics()` so the merge-side baseline surfaces, plus per-input-partition scatter counters: - `scatter_elapsed_compute` — total scatter compute per input, timer scoped post child-poll. - `scatter_split_time` — subset time for `split_batch_by_range` alone. Isolates routing-expr evaluation cost from the surrounding channel work. - `scatter_send_time` — subset time for `senders[out].send().await`. High values relative to compute mean downstream merge is draining slower than we can scatter (backpressure). - `scatter_discover_cuts_time` — one-shot first-batch cost of the cut-discovery walk into `RuntimeStatsExec`. - `scatter_input_batches` / `scatter_input_rows` — input volume. - `scatter_output_sub_batches` — post-split fanout count. Under skew this can be up to K× input_batches. Compute timer is stopped around `send().await` so backpressure waits land in `send_time` alone, not double-counted into `elapsed_compute`. ## Context Extracted from #2223 (parallel BWAG for the range-window shape). Both ops already exist on main from prior slices; this is pure instrumentation on top of them, standalone. -- 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]
