feat(core): metrics() for OrderedRangeRepartitionExec and RuntimeStatsExec

[avantgardnerio]

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.

[avantgardnerio]

@phillipleblanc now that I'm using these for real, I needed metrics to debug performance.

[milenkovicm]

thanks @avantgardnerio

[milenkovicm]

it happened to me as well (not once) 😀