avantgardnerio opened a new pull request, #2094: URL: https://github.com/apache/datafusion-ballista/pull/2094
## 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 - [x] `cargo test -p ballista-core` (all 123 tests + 2 doctests + 1 compile-fail pass). - [x] `cargo clippy --all-targets` clean. - [x] `cargo fmt --all`. 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
