This is an automated email from the ASF dual-hosted git repository.
avantgardnerio pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-ballista.git
The following commit(s) were added to refs/heads/main by this push:
new 3910a6ad0 feat(core): metrics() for OrderedRangeRepartitionExec and
RuntimeStatsExec (#2250)
3910a6ad0 is described below
commit 3910a6ad05a8076a7a842720c43e5e7c1e78b346
Author: Brent Gardner <[email protected]>
AuthorDate: Fri Aug 7 13:00:11 2026 -0600
feat(core): metrics() for OrderedRangeRepartitionExec and RuntimeStatsExec
(#2250)
* feat(core): RuntimeStatsExec metrics — baseline + sketch subset time
Was returning `None` from `metrics()`, so it was invisible in the
scheduler's stage-metrics dump AND caused display-walker misalignment:
metrics for later operators (SortExec, DataSourceExec) got labeled as
this op or ORRE, because `collect_plan_metrics` skips None returns
while the display visitor increments its index for every node.
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 (Time clone so the timer doesn't hold a
borrow across the mutable `ingest` call).
Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
* feat(core): ORRE metrics — scatter-side timers + fn metrics()
ORRE was constructing BaselineMetrics for its StreamingMerge output
side but never exposing them (no `fn metrics()` impl), and the scatter
side — where the actual value-add of range-routing lives — had zero
instrumentation.
Adds `fn metrics()` so both sides surface, plus per-input-partition
scatter counters:
- `scatter_elapsed_compute` — total scatter compute per input,
timer scoped post child-poll (upstream sort/read not billed).
- `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.
Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
* fixup: drop noisy Time-clone comment, allow too_many_arguments on scatter
fn
Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
---------
Co-authored-by: Claude Opus 4.7 (1M context) <[email protected]>
---
.../execution_plans/ordered_range_repartition.rs | 74 +++++++++++++++++++++-
ballista/core/src/execution_plans/runtime_stats.rs | 37 ++++++++++-
2 files changed, 107 insertions(+), 4 deletions(-)
diff --git a/ballista/core/src/execution_plans/ordered_range_repartition.rs
b/ballista/core/src/execution_plans/ordered_range_repartition.rs
index c040674da..3c0b42d8d 100644
--- a/ballista/core/src/execution_plans/ordered_range_repartition.rs
+++ b/ballista/core/src/execution_plans/ordered_range_repartition.rs
@@ -87,7 +87,9 @@ use datafusion::physical_expr::{
use datafusion::physical_plan::execution_plan::{
CardinalityEffect, EvaluationType, SchedulingType,
};
-use datafusion::physical_plan::metrics::{BaselineMetrics,
ExecutionPlanMetricsSet};
+use datafusion::physical_plan::metrics::{
+ BaselineMetrics, Count, ExecutionPlanMetricsSet, MetricBuilder,
MetricsSet, Time,
+};
use datafusion::physical_plan::sorts::streaming_merge::StreamingMergeBuilder;
use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
use datafusion::physical_plan::{
@@ -330,6 +332,10 @@ impl ExecutionPlan for OrderedRangeRepartitionExec {
}
/// Every input row is emitted exactly once. Overrides default `Unknown`.
+ fn metrics(&self) -> Option<MetricsSet> {
+ Some(self.metrics.clone_inner())
+ }
+
fn cardinality_effect(&self) -> CardinalityEffect {
CardinalityEffect::Equal
}
@@ -431,6 +437,22 @@ impl OrderedRangeRepartitionExec {
let scatter_senders: Arc<[mpsc::Sender<Result<RecordBatch>>]> =
senders.into();
let guard_senders = scatter_senders.clone();
+ let scatter_metrics = ScatterMetrics {
+ elapsed_compute: MetricBuilder::new(&self.metrics)
+ .subset_time("scatter_elapsed_compute", input_partition),
+ split_time: MetricBuilder::new(&self.metrics)
+ .subset_time("scatter_split_time", input_partition),
+ send_time: MetricBuilder::new(&self.metrics)
+ .subset_time("scatter_send_time", input_partition),
+ discover_cuts_time: MetricBuilder::new(&self.metrics)
+ .subset_time("scatter_discover_cuts_time",
input_partition),
+ input_batches: MetricBuilder::new(&self.metrics)
+ .counter("scatter_input_batches", input_partition),
+ input_rows: MetricBuilder::new(&self.metrics)
+ .counter("scatter_input_rows", input_partition),
+ scattered_batches: MetricBuilder::new(&self.metrics)
+ .counter("scatter_output_sub_batches", input_partition),
+ };
// `guarded_scatter` wraps the body in `catch_unwind`: a panic
// inside `scatter_input_partition` (or anything it calls) becomes
// a broadcast error rather than a silent sender-drop that
@@ -444,6 +466,7 @@ impl OrderedRangeRepartitionExec {
scatter_senders,
cuts_cell,
output_partitions,
+ scatter_metrics,
),
guard_senders,
)));
@@ -488,6 +511,37 @@ impl OrderedRangeRepartitionExec {
}
}
+/// Per-input-partition scatter counters. Labeled with `input_partition`;
+/// `aggregate_by_name` in the display collapses across all input
+/// partitions for a task-wide total.
+///
+/// Merge-side metrics (per output partition) are recorded separately
+/// by `StreamingMerge` via `BaselineMetrics` in `initialize_state`.
+struct ScatterMetrics {
+ /// Total scatter compute — timer scoped post child-poll so upstream
+ /// stream time (sort, parquet scan) isn't billed here.
+ elapsed_compute: Time,
+ /// `split_batch_by_range` cost isolated from the surrounding
+ /// send/backpressure work. Growing much faster than input_rows
+ /// suggests routing-expression evaluation is the hot loop.
+ split_time: Time,
+ /// `senders[output].send().await` cost. Includes backpressure
+ /// waits — high values mean downstream merge is slow to drain,
+ /// which starves the scatter path.
+ send_time: Time,
+ /// One-shot cost of the first-batch cut discovery walk. Reads the
+ /// upstream `RuntimeStatsExec` and computes K-1 quantiles.
+ discover_cuts_time: Time,
+ /// Batches read from the child. Zero-row batches short-circuit
+ /// so this counts only work-carrying batches.
+ input_batches: Count,
+ /// Rows read from the child.
+ input_rows: Count,
+ /// Sub-batches sent downstream. Under skew this can be much larger
+ /// than input_batches (each input batch splits into up to K subs).
+ scattered_batches: Count,
+}
+
/// One background task per input partition. Reads the sorted input; on the
/// first batch, the shared `OnceLock` learns the value cuts. Each batch is
/// then dispatched via `split_batch_by_range` to K per-input senders (one
@@ -501,6 +555,7 @@ impl OrderedRangeRepartitionExec {
/// 2. **Send zero-copy Arrow slices** (`batch.slice(start, len)`) instead
/// of materialising via `take_arrays`. Turns per-batch scatter into a
/// few Arc bumps rather than N × K allocations under skew.
+#[allow(clippy::too_many_arguments)]
async fn scatter_input_partition(
child: Arc<dyn ExecutionPlan>,
input_partition: usize,
@@ -509,6 +564,7 @@ async fn scatter_input_partition(
senders: Arc<[mpsc::Sender<Result<RecordBatch>>]>,
cuts_cell: Arc<OnceLock<Vec<f64>>>,
output_partitions: usize,
+ metrics: ScatterMetrics,
) -> Result<()> {
let mut stream = child.execute(input_partition, ctx)?;
while let Some(batch_result) = stream.next().await {
@@ -517,9 +573,15 @@ async fn scatter_input_partition(
if senders.iter().all(|s| s.is_closed()) {
return Ok(());
}
+ let compute_timer = metrics.elapsed_compute.timer();
let batch = batch_result?;
+ metrics.input_batches.add(1);
+ metrics.input_rows.add(batch.num_rows());
let cuts = cuts_cell.get_or_init(|| {
- discover_cuts(&child, routing_expr.as_ref(), output_partitions)
+ let discover_timer = metrics.discover_cuts_time.timer();
+ let cuts = discover_cuts(&child, routing_expr.as_ref(),
output_partitions);
+ discover_timer.done();
+ cuts
});
// TODO(perf): input is sorted — this per-row `split_batch_by_range`
// is legal but wasteful. Two follow-ups worth measuring:
@@ -528,16 +590,24 @@ async fn scatter_input_partition(
// 2. Send zero-copy `batch.slice(start, len)` instead of
// `take_arrays`-materialised sub-batches. Arc bumps replace
// allocations under skew.
+ let split_timer = metrics.split_time.timer();
let splits = split_batch_by_range(&batch, &routing_expr, cuts)?;
+ split_timer.done();
+ // Stop the compute timer around the send.await so backpressure
+ // waits get billed to `send_time` alone, not double-counted.
+ compute_timer.done();
for (output, sub) in splits.into_iter().enumerate() {
if sub.num_rows() == 0 {
continue;
}
+ let send_timer = metrics.send_time.timer();
// `send().await` provides backpressure: full channel suspends
// this task → suspends input read → propagates upstream. Error
// on send means downstream dropped its receiver; keep forwarding
// to other outputs.
let _ = senders[output].send(Ok(sub)).await;
+ send_timer.done();
+ metrics.scattered_batches.add(1);
}
}
// Senders drop with this task → each output's merger sees EOF on that
diff --git a/ballista/core/src/execution_plans/runtime_stats.rs
b/ballista/core/src/execution_plans/runtime_stats.rs
index 949241a16..eff45c6f1 100644
--- a/ballista/core/src/execution_plans/runtime_stats.rs
+++ b/ballista/core/src/execution_plans/runtime_stats.rs
@@ -61,6 +61,9 @@ use datafusion::physical_expr::{
Distribution, OrderingRequirements, PhysicalExpr, PhysicalSortExpr,
};
use datafusion::physical_plan::execution_plan::CardinalityEffect;
+use datafusion::physical_plan::metrics::{
+ BaselineMetrics, Count, ExecutionPlanMetricsSet, MetricBuilder,
MetricsSet, Time,
+};
use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
use datafusion::physical_plan::{
DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties,
PlanProperties,
@@ -96,6 +99,7 @@ pub struct RuntimeStatsExec {
/// keep writes off any shared lock.
sketches: Option<Arc<[Mutex<TDigest>]>>,
properties: Arc<PlanProperties>,
+ metrics: ExecutionPlanMetricsSet,
}
impl RuntimeStatsExec {
@@ -154,6 +158,7 @@ impl RuntimeStatsExec {
row_counts,
sketches,
properties,
+ metrics: ExecutionPlanMetricsSet::new(),
})
}
@@ -323,6 +328,10 @@ impl ExecutionPlan for RuntimeStatsExec {
CardinalityEffect::Equal
}
+ fn metrics(&self) -> Option<MetricsSet> {
+ Some(self.metrics.clone_inner())
+ }
+
fn with_new_children(
self: Arc<Self>,
children: Vec<Arc<dyn ExecutionPlan>>,
@@ -367,19 +376,37 @@ impl ExecutionPlan for RuntimeStatsExec {
.and_then(|exprs| exprs.first())
.map(|e| e.expr.clone());
+ let baseline = BaselineMetrics::new(&self.metrics, partition);
+ // Separates sketch cost from the rest so we can tell whether the
+ // hot path in sketching mode is the T-Digest merge_unsorted_f64
+ // or something else (evaluate/downcast/flatten).
+ let sketch_time =
+ MetricBuilder::new(&self.metrics).subset_time("sketch_time",
partition);
+ let sketch_batches =
+ MetricBuilder::new(&self.metrics).counter("sketch_batches",
partition);
+
let state = StreamState {
input: input_stream,
row_counts,
sketches,
routing_expr,
partition,
+ baseline,
+ sketch_time,
+ sketch_batches,
};
let out = futures::stream::unfold(state, |mut state| async move {
- let batch = state.input.next().await?;
- let forwarded = batch.and_then(|batch| {
+ // Await the child first so upstream shuffle IO / parquet
+ // reads aren't billed to our elapsed_compute.
+ let next = state.input.next().await?;
+ let elapsed = state.baseline.elapsed_compute().clone();
+ let timer = elapsed.timer();
+ let forwarded = next.and_then(|batch| {
state.ingest(&batch)?;
+ state.baseline.record_output(batch.num_rows());
Ok(batch)
});
+ timer.done();
Some((forwarded, state))
});
@@ -396,6 +423,9 @@ struct StreamState {
sketches: Option<Arc<[Mutex<TDigest>]>>,
routing_expr: Option<Arc<dyn datafusion::physical_plan::PhysicalExpr>>,
partition: usize,
+ baseline: BaselineMetrics,
+ sketch_time: Time,
+ sketch_batches: Count,
}
impl StreamState {
@@ -457,7 +487,10 @@ impl StreamState {
self.partition
)
})?;
+ let sketch_timer = self.sketch_time.timer();
*sketch = sketch.merge_unsorted_f64(values);
+ sketch_timer.done();
+ self.sketch_batches.add(1);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
