algeon-query-runtime 0.1.2

Native query runtime, lifecycle, ledger, and observation owners.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

use query_planning::correlation::QueryId;

use super::{JoinOperatorMetrics, QueryMetrics};

/// Generic operator metric row.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct OperatorMetrics {
    /// Query-wide diagnostic correlation id, when assigned.
    pub query_id: Option<QueryId>,
    /// Native node id.
    pub node_id: String,
    /// Operator-kind label.
    pub operator_kind: String,
    /// Input rows consumed by the operator.
    pub input_rows: u64,
    /// Input bytes consumed by the operator.
    pub input_bytes: u64,
    /// Output rows produced by the operator.
    pub output_rows: u64,
    /// Output bytes produced by the operator.
    pub output_bytes: u64,
    /// Join-specific metrics when this is a join row.
    pub join: Option<JoinOperatorMetrics>,
}

impl OperatorMetrics {
    /// Creates a join operator metric row.
    pub fn join(node_id: impl ToString, join: JoinOperatorMetrics) -> Self {
        Self { node_id: node_id.to_string(), operator_kind: "join".to_string(), join: Some(join), ..Default::default() }
    }
}

impl QueryMetrics {
    /// Appends one operator metric row.
    pub fn record_operator_metrics(&mut self, mut metrics: OperatorMetrics) {
        if metrics.query_id.is_none() {
            metrics.query_id = self.inherited_query_id();
        }
        if let Some(join) = &mut metrics.join {
            if join.query_id.is_none() {
                join.query_id = metrics.query_id.clone();
            }
            self.join_gather_map_bytes = self.join_gather_map_bytes.saturating_add(join.gather_map_bytes);
        }
        self.operators.push(metrics);
    }
}