query_runtime/metrics/
operator.rs

1use query_planning::correlation::QueryId;
2
3use super::{JoinOperatorMetrics, QueryMetrics};
4
5/// Generic operator metric row.
6#[derive(Debug, Clone, Default, PartialEq, Eq)]
7pub struct OperatorMetrics {
8    /// Query-wide diagnostic correlation id, when assigned.
9    pub query_id: Option<QueryId>,
10    /// Native node id.
11    pub node_id: String,
12    /// Operator-kind label.
13    pub operator_kind: String,
14    /// Input rows consumed by the operator.
15    pub input_rows: u64,
16    /// Input bytes consumed by the operator.
17    pub input_bytes: u64,
18    /// Output rows produced by the operator.
19    pub output_rows: u64,
20    /// Output bytes produced by the operator.
21    pub output_bytes: u64,
22    /// Join-specific metrics when this is a join row.
23    pub join: Option<JoinOperatorMetrics>,
24}
25
26impl OperatorMetrics {
27    /// Creates a join operator metric row.
28    pub fn join(node_id: impl ToString, join: JoinOperatorMetrics) -> Self {
29        Self { node_id: node_id.to_string(), operator_kind: "join".to_string(), join: Some(join), ..Default::default() }
30    }
31}
32
33impl QueryMetrics {
34    /// Appends one operator metric row.
35    pub fn record_operator_metrics(&mut self, mut metrics: OperatorMetrics) {
36        if metrics.query_id.is_none() {
37            metrics.query_id = self.inherited_query_id();
38        }
39        if let Some(join) = &mut metrics.join {
40            if join.query_id.is_none() {
41                join.query_id = metrics.query_id.clone();
42            }
43            self.join_gather_map_bytes = self.join_gather_map_bytes.saturating_add(join.gather_map_bytes);
44        }
45        self.operators.push(metrics);
46    }
47}
query_runtime/metrics/operator.rs

query_runtime/metrics/
operator.rs
