query_runtime/observability/
metrics_summary.rs

1//! Exact query metrics reduction kept by the observation owner.
2//!
3//! The owner updates these totals on every typed fact *before* deciding whether
4//! the fact's detail row is retained, so a total never depends on retention.
5//! Trace capture can be disabled or exhausted; the totals stay exact either way.
6//!
7//! Only the source-read family is reduced here so far. Families that have not
8//! moved yet are still written into `QueryMetrics` directly, and are listed in
9//! `docs/plan/2026-09-16-query-layer-boundaries/d4-observation-authorization.md`.
10
11use std::time::Duration;
12
13use crate::metrics::{QueryMetrics, SourceReadRuntimeMetrics};
14
15/// One source-read fact, recorded once at the boundary that owns it.
16///
17/// Each variant has exactly one recording site. That is what keeps retries,
18/// abandoned prefetches, and parallel workers from counting the same work
19/// twice: a worker records its physical read, the scheduler records the
20/// logical commit, and neither re-reports the other's facts.
21#[derive(Debug, Clone, PartialEq, Eq)]
22pub enum SourceReadFact {
23    /// A source chunk was planned.
24    ChunkPlanned { row_groups: u64 },
25    /// A source chunk read was attempted.
26    ChunkAttempted,
27    /// Source chunks were committed as logical output.
28    ChunkCommitted { chunk_count: u64, row_groups: u64 },
29    /// A source chunk read will be retried.
30    ChunkRetried,
31    /// A source-read prefetch was started.
32    PrefetchStarted,
33    /// A source-read prefetch completed.
34    PrefetchCompleted,
35    /// A source-read prefetch was skipped.
36    PrefetchSkipped,
37    /// A source-read prefetch was abandoned before its output was consumed.
38    PrefetchAbandoned,
39    /// Time one read spent between enqueue and a worker starting it.
40    QueueWait(Duration),
41    /// One input batch reached execution.
42    InputBatch { rows: u64 },
43    /// Parquet source work, as explicit increments. Recording sites add
44    /// different subsets of these; a re-read that only learns its row-group
45    /// count adds no file. Each field carries only what its recording site adds.
46    ParquetSource { files: u64, known_bytes: u64, selected_row_groups: u64, read_columns: u64 },
47    /// Time spent loading source metadata.
48    MetadataLoadTime(Duration),
49    /// Time spent inside the Parquet read call.
50    ParquetReadTime(Duration),
51    /// Time spent in source-read orchestration around the Parquet call.
52    SourceReadTime(Duration),
53}
54
55/// Exact source-read totals for one query.
56#[derive(Debug, Clone, Default, PartialEq, Eq)]
57pub struct SourceReadTotals {
58    pub(crate) input_batches: u64,
59    pub(crate) input_rows: u64,
60    /// Parquet file counts added by source recording sites.
61    pub parquet_files: u64,
62    /// Sum of known compressed Parquet byte counts reported by source sites.
63    pub parquet_bytes_known: u64,
64    /// Sum of selected Parquet row groups reported by source sites.
65    pub row_groups_selected: u64,
66    pub(crate) read_columns: u64,
67    pub(crate) source_chunks_planned: u64,
68    pub(crate) source_chunks_attempted: u64,
69    pub(crate) source_chunks_committed: u64,
70    pub(crate) source_chunks_retried: u64,
71    pub(crate) row_groups_planned: u64,
72    pub(crate) row_groups_committed: u64,
73    pub(crate) source_read_prefetches_started: u64,
74    pub(crate) source_read_prefetches_completed: u64,
75    pub(crate) source_read_prefetches_skipped: u64,
76    pub(crate) source_read_prefetches_abandoned: u64,
77    pub(crate) source_metadata_load_time: Duration,
78    /// Cumulative source-worker queue wait duration.
79    pub source_read_queue_wait_time: Duration,
80    pub(crate) parquet_read_time: Duration,
81    pub(crate) source_read_time: Duration,
82}
83
84impl SourceReadTotals {
85    /// Folds one fact into the totals. Returns `false` when a counter
86    /// saturated, so the caller can mark the totals as no longer exact.
87    pub(crate) fn apply(&mut self, fact: &SourceReadFact) -> bool {
88        match fact {
89            SourceReadFact::ChunkPlanned { row_groups } => {
90                add_u64(&mut self.source_chunks_planned, 1) & add_u64(&mut self.row_groups_planned, *row_groups)
91            },
92            SourceReadFact::ChunkAttempted => add_u64(&mut self.source_chunks_attempted, 1),
93            SourceReadFact::ChunkCommitted { chunk_count, row_groups } => {
94                add_u64(&mut self.source_chunks_committed, *chunk_count)
95                    & add_u64(&mut self.row_groups_committed, *row_groups)
96            },
97            SourceReadFact::ChunkRetried => add_u64(&mut self.source_chunks_retried, 1),
98            SourceReadFact::PrefetchStarted => add_u64(&mut self.source_read_prefetches_started, 1),
99            SourceReadFact::PrefetchCompleted => add_u64(&mut self.source_read_prefetches_completed, 1),
100            SourceReadFact::PrefetchSkipped => add_u64(&mut self.source_read_prefetches_skipped, 1),
101            SourceReadFact::PrefetchAbandoned => add_u64(&mut self.source_read_prefetches_abandoned, 1),
102            SourceReadFact::InputBatch { rows } => {
103                add_u64(&mut self.input_batches, 1) & add_u64(&mut self.input_rows, *rows)
104            },
105            SourceReadFact::ParquetSource { files, known_bytes, selected_row_groups, read_columns } => {
106                add_u64(&mut self.parquet_files, *files)
107                    & add_u64(&mut self.parquet_bytes_known, *known_bytes)
108                    & add_u64(&mut self.row_groups_selected, *selected_row_groups)
109                    & add_u64(&mut self.read_columns, *read_columns)
110            },
111            SourceReadFact::QueueWait(elapsed) => add_duration(&mut self.source_read_queue_wait_time, *elapsed),
112            SourceReadFact::MetadataLoadTime(elapsed) => add_duration(&mut self.source_metadata_load_time, *elapsed),
113            SourceReadFact::ParquetReadTime(elapsed) => add_duration(&mut self.parquet_read_time, *elapsed),
114            SourceReadFact::SourceReadTime(elapsed) => add_duration(&mut self.source_read_time, *elapsed),
115        }
116    }
117}
118
119// `apply` combines results with the non-short-circuit `&` operator. Every
120// counter in a fact must be updated even after an earlier one saturates.
121fn add_u64(counter: &mut u64, value: u64) -> bool {
122    match counter.checked_add(value) {
123        Some(next) => {
124            *counter = next;
125            true
126        },
127        None => {
128            *counter = u64::MAX;
129            false
130        },
131    }
132}
133
134fn add_duration(total: &mut Duration, value: Duration) -> bool {
135    match total.checked_add(value) {
136        Some(next) => {
137            *total = next;
138            true
139        },
140        None => {
141            *total = Duration::MAX;
142            false
143        },
144    }
145}
146
147/// Exact query metrics reduction kept by the observation owner.
148#[derive(Debug, Clone, Default, PartialEq, Eq)]
149pub struct QueryMetricsSummary {
150    /// Exact source totals reduced independently of bounded detail retention.
151    pub source_read: SourceReadTotals,
152}
153
154/// Whether detail rows were captured at all.
155///
156/// A disabled capture retains nothing on purpose, which is not the same as a
157/// bounded capture that happened to retain everything. Zero retained rows and
158/// zero losses under [`Self::Disabled`] say nothing about completeness.
159#[derive(Debug, Clone, Copy, PartialEq, Eq)]
160pub(crate) enum MetricsCaptureMode {
161    /// Trace capture was off; only totals exist.
162    Disabled,
163    /// Detail rows were retained up to the recorder's record and byte limits.
164    Bounded,
165}
166
167impl MetricsCaptureMode {
168    /// Stable report label.
169    pub const fn as_str(self) -> &'static str {
170        match self {
171            Self::Disabled => "disabled",
172            Self::Bounded => "bounded",
173        }
174    }
175}
176
177/// What a metrics checkpoint can and cannot claim.
178#[derive(Debug, Clone, Copy, PartialEq, Eq)]
179pub struct MetricsCompleteness {
180    pub(crate) capture_mode: MetricsCaptureMode,
181    /// Detail rows the recorder declined to retain.
182    pub(crate) detail_rows_dropped: u64,
183    /// Whether any detail row was dropped.
184    pub truncated: bool,
185    /// Whether a total saturated, so totals are no longer exact.
186    pub(crate) totals_saturated: bool,
187}
188
189impl MetricsCompleteness {
190    /// Whether the checkpoint's detail rows are every row that was recorded.
191    #[cfg(test)]
192    pub(crate) const fn detail_is_complete(&self) -> bool {
193        matches!(self.capture_mode, MetricsCaptureMode::Bounded) && !self.truncated
194    }
195
196    /// Stable report label for the retained detail: `not_captured` when trace
197    /// capture was off, `incomplete` when rows were declined, else `complete`.
198    pub(crate) const fn detail_status(&self) -> &'static str {
199        match (self.capture_mode, self.truncated) {
200            (MetricsCaptureMode::Disabled, _) => "not_captured",
201            (MetricsCaptureMode::Bounded, true) => "incomplete",
202            (MetricsCaptureMode::Bounded, false) => "complete",
203        }
204    }
205}
206
207/// One immutable, internally consistent view of the owner's metrics.
208///
209/// Totals, retained detail, and completeness come from a single lock
210/// acquisition and carry one `revision`, so a projection never mixes two
211/// moments. Revisions only increase; a projection that already applied a
212/// revision can skip it.
213#[derive(Debug, Clone, PartialEq, Eq)]
214pub struct QueryMetricsCheckpoint {
215    pub(crate) revision: u64,
216    /// Whether the owner has sealed, so no later revision will follow.
217    pub(crate) sealed: bool,
218    /// Exact reductions sampled under the same owner lock as retained details.
219    pub summary: QueryMetricsSummary,
220    /// Retained source-read detail rows, in recording order.
221    pub source_read_rows: Vec<SourceReadRuntimeMetrics>,
222    /// Reduction of every other metrics family, with its growing detail rows
223    /// already bounded. Its source-read fields are unused; see `summary`.
224    pub(crate) metrics: QueryMetrics,
225    pub(crate) completeness: MetricsCompleteness,
226}
227
228#[cfg(test)]
229mod tests;
query_runtime/observability/metrics_summary.rs

query_runtime/observability/
metrics_summary.rs
