feat(core): RuntimeStatsExec — passthrough tap for row counts + quantile sketches

[avantgardnerio]

Summary

Adds RuntimeStatsExec: a passthrough physical operator that observes input batches as they stream past and accumulates runtime state, without touching the data. Two accessor families:

• row count — per input partition, AtomicUsize::fetch_add on the hot path (no cross-partition contention, no lock overhead). Always tracked.
• quantile sketch — per input partition, Mutex<TDigest> (writes never contend across partitions). Only allocated when the caller passes Some(order_by) at construction.

Both accessors are readable at any point, so callers get a mid-stream snapshot when the sample is accurate enough or a post-drain snapshot when they want the fully-observed state (typical after a blocking downstream like SortExec).

Why this exists — supporting the forthcoming range-repartition operators

RuntimeStatsExec is the shared substrate for two upcoming ops (OrderedRangeRepartitionExec and UnorderedRangeRepartitionExec) that will land the distributed parallel-window path. Two modes matched to two use sites:

Sketching mode (Some(order_by)) → drives UnorderedRangeRepartitionExec.
Placed above the scan, RuntimeStatsExec streams a T-Digest over the routing column while the data is scanned. Each executor's sketch is merged (TDigest::merge_digests) to choose global cut points at a scheduler barrier, then UnorderedRangeRepartitionExec uses those cuts to route rows to sub-partitions by value range — no upfront sort required.

Row-count-only mode (None) → drives OrderedRangeRepartitionExec.
Placed above a SortExec, it exposes exact runtime row counts per sub-partition. DataFusion's partition_statistics() only returns plan-time estimates, which are too coarse for the ordered stage's bin-packer. The tap fixes that at operator level without touching DF upstream.

The full parallel-window path (Stage 1 sketch → cut selection → Stage 2 slice) is described in the parallel-window-kll-adaptive design doc on the source branch and will land alongside the range-repartition operators.

Scope of this PR

Just the operator, its proto/serde plumbing, and tests. Nothing consumes the accumulated state yet — the executor still ships SuccessfulTask unchanged, and no rule inserts RuntimeStatsExec into any plan. Those pieces land with the range-repartition operators.

Included:

• datafusion-functions-aggregate-common = \"54\" workspace dep (for TDigest).
• Proto: RuntimeStatsExecNode on BallistaPhysicalPlanNode.oneof (tag 6), and QuantileSketchState for the wire T-Digest (fixed 6-element Vec<ScalarValue> per TDigest::to_scalar_state).
• sketch_to_proto / sketch_from_proto helpers with a shape guard against corrupted wire input (would otherwise panic in from_scalar_state).
• Codec encode/decode registration on BallistaPhysicalExtensionCodec.
• TODO in module docs: swap TDigest (Float64-only, single-column) for a generic-over-Ord KLL sketch that can cover the full Vec<PhysicalSortExpr>.

Tests (9)

• Three wire round-trips: populated sketch, empty sketch, wrong-element-count rejection.
• Three streaming: execute_populates_sketch_and_row_count, execute_skips_null_routing_keys_from_sketch, execute_row_count_only_no_sketch.
• Three plan-node serde: sketching-mode roundtrip, row-count-only-mode roundtrip, empty-order_by rejection.

Test plan

• cargo test -p ballista-core (all 123 tests + 2 doctests + 1 compile-fail pass).
• cargo clippy --all-targets clean.
• cargo fmt --all.

🤖 Generated with Claude Code

[avantgardnerio]

@phillipleblanc FYI

[phillipleblanc]

I have a few comments, but this is looking good!

[avantgardnerio]

Thanks for your feedback @phillipleblanc . As usual it was very helpful 😄