query_runtime/metrics/
aggregate.rs

1use std::collections::BTreeMap;
2
3use query_planning::correlation::QueryId;
4
5use super::QueryMetrics;
6
7/// Aggregate-strategy decision facts captured for diagnostics.
8#[derive(Debug, Clone, Default, PartialEq, Eq)]
9pub struct AggregateStrategyDecisionMetrics {
10    /// Query-wide diagnostic correlation id, when assigned.
11    pub(crate) query_id: Option<QueryId>,
12    /// Native aggregate node id, or a stage-local fallback label when unavailable.
13    pub node_id: String,
14    /// Stable decision fact keys and values.
15    pub facts: BTreeMap<String, String>,
16}
17
18impl QueryMetrics {
19    /// Records aggregate-strategy diagnostics for a node or execution stage.
20    pub fn record_aggregate_strategy_decision(
21        &mut self,
22        node_id: impl ToString,
23        facts: &BTreeMap<&'static str, String>,
24    ) {
25        self.aggregate_strategy_decisions.push(AggregateStrategyDecisionMetrics {
26            query_id: self.inherited_query_id(),
27            node_id: node_id.to_string(),
28            facts: facts.iter().map(|(key, value)| ((*key).to_string(), value.clone())).collect(),
29        });
30    }
31
32    /// Records one partial aggregate task summary.
33    pub fn record_partial_aggregate(&mut self, input_rows: u64, output_rows: u64, retained_bytes: u64) {
34        self.partial_aggregate_count = self.partial_aggregate_count.saturating_add(1);
35        self.partial_aggregate_input_rows = self.partial_aggregate_input_rows.saturating_add(input_rows);
36        self.partial_aggregate_rows = self.partial_aggregate_rows.saturating_add(output_rows);
37        self.partial_aggregate_retained_bytes = self.partial_aggregate_retained_bytes.max(retained_bytes);
38    }
39
40    /// Records one final merge attempt and fan-in.
41    pub fn record_final_merge_attempt(&mut self, merge_fan_in: usize) {
42        self.final_merge_attempts = self.final_merge_attempts.saturating_add(1);
43        self.final_merge_max_fan_in = self.final_merge_max_fan_in.max(merge_fan_in as u64);
44    }
45
46    /// Records one final merge level.
47    pub fn record_final_merge_level(&mut self) {
48        self.final_merge_levels = self.final_merge_levels.saturating_add(1);
49    }
50}
query_runtime/metrics/aggregate.rs

query_runtime/metrics/
aggregate.rs
