Skip to content

metrics: update plan_bytes_sent to be a bytes metric #506

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jayshrivastava merged 3 commits into from
Jun 18, 2026
Merged

metrics: update plan_bytes_sent to be a bytes metric #506

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a567b4b
metrics: update plan_bytes_sent to be a bytes metric
jayshrivastava Jun 18, 2026
40eba83
nuke test
jayshrivastava Jun 18, 2026
8822f6d
newline
jayshrivastava Jun 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/source/user-guide/metrics.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ This produces an EXPLAIN ANALYZE that spans the whole cluster — every stage an
runtime metrics, including network-level metrics on the boundaries:

```
┌───── DistributedExec ── Tasks: t0:[p0] plan_bytes_sent_0=8.07 K, plan_send_latency_avg_0=22.63ms, ...
┌───── DistributedExec ── Tasks: t0:[p0] plan_bytes_sent_0=8.07 KB, plan_send_latency_avg_0=22.63ms, ...
│ SortPreservingMergeExec: [count(*)@0 DESC], fetch=5, metrics=[output_rows=5, elapsed_compute=391.83µs, ...]
│ [Stage 2] => NetworkCoalesceExec: output_partitions=32, input_tasks=2, metrics=[elapsed_compute=5.86ms, bytes_transferred=20.1 KB, network_latency_p50=366.00µs, network_latency_p95=603.43µs, ...]
└──────────────────────────────────────────────────
Expand Down
16 changes: 7 additions & 9 deletions src/coordinator/query_coordinator.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,19 +11,17 @@ use crate::worker::generated::worker as pb;
use crate::worker::generated::worker::coordinator_to_worker_msg::Inner;
use crate::worker::generated::worker::set_plan_request::WorkUnitFeedDeclaration;
use crate::{
DISTRIBUTED_DATAFUSION_TASK_ID_LABEL, DistributedCodec, DistributedConfig,
DistributedTaskContext, DistributedWorkUnitFeedContext, TaskEstimator, TaskKey,
TaskRoutingContext, get_distributed_channel_resolver, get_distributed_worker_resolver,
BytesCounterMetric, BytesMetricExt, DISTRIBUTED_DATAFUSION_TASK_ID_LABEL, DistributedCodec,
DistributedConfig, DistributedTaskContext, DistributedWorkUnitFeedContext, TaskEstimator,
TaskKey, TaskRoutingContext, get_distributed_channel_resolver, get_distributed_worker_resolver,
};
use datafusion::common::instant::Instant;
use datafusion::common::runtime::JoinSet;
use datafusion::common::tree_node::{Transformed, TreeNodeRecursion};
use datafusion::common::{DataFusionError, exec_datafusion_err};
use datafusion::common::{Result, exec_err};
use datafusion::execution::TaskContext;
use datafusion::physical_expr_common::metrics::{
Count, ExecutionPlanMetricsSet, Label, MetricBuilder,
};
use datafusion::physical_expr_common::metrics::{ExecutionPlanMetricsSet, Label, MetricBuilder};
use datafusion::physical_plan::ExecutionPlan;
use datafusion_proto::physical_plan::AsExecutionPlan;
use datafusion_proto::protobuf::PhysicalPlanNode;
Expand Down Expand Up @@ -198,7 +196,7 @@ impl<'a> StageCoordinator<'a> {
})
})?;
metrics.plan_send_latency.record(&start);
metrics.plan_bytes_sent.add(plan_size);
metrics.plan_bytes_sent.add_bytes(plan_size);
let mut worker_to_coordinator_stream = response.into_inner();
while let Some(msg_or_err) = worker_to_coordinator_stream.next().await {
let msg = msg_or_err.map_err(|err| {
Expand Down Expand Up @@ -420,7 +418,7 @@ impl Drop for NotifyGuard {
/// Metrics that measure network details about communications between [DistributedExec] and a worker.
#[derive(Clone)]
pub(super) struct CoordinatorToWorkerMetrics {
pub(super) plan_bytes_sent: Count,
pub(super) plan_bytes_sent: BytesCounterMetric,
pub(super) plan_send_latency: Arc<LatencyMetric>,
pub(super) instantiation_time: u64,
}
Expand All @@ -431,7 +429,7 @@ impl CoordinatorToWorkerMetrics {
// Metric that measures to total sum of bytes worth of subplans sent.
plan_bytes_sent: MetricBuilder::new(metrics)
.with_label(Label::new(DISTRIBUTED_DATAFUSION_TASK_ID_LABEL, "0"))
.global_counter("plan_bytes_sent"),
.bytes_counter("plan_bytes_sent"),
// Latency statistics about the network calls issued to the workers for feeding subplans.
plan_send_latency: Arc::new(LatencyMetric::new(
"plan_send_latency",
Expand Down
Loading