linkrs_query/executor/streaming/runtime/
columnar_stats.rs

1//! Query-level columnar fast-path counters (observability).
2//!
3//! Shared via `Arc` with the chunks produced by source operators; every
4//! `DataChunk::evaluate_expression` call records a hit (columnar batch path
5//! succeeded) or a miss (fell back to per-row evaluation). The miss rate
6//! exposes how much of the flat-column promise is actually kept.
7
8use std::collections::HashMap;
9use std::sync::atomic::{AtomicU64, Ordering};
10
11/// Constants for typed-column analysis.
12///
13/// These are exported so that benchmark harnesses can emit machine-readable
14/// status lines that mirror the selection logic.
15pub const D1_EVAL_THRESHOLD: u64 = 1_000_000;
16pub const D1_TYPED_RATE_THRESHOLD: f64 = 0.5;
17
18/// Closed enumeration of opaque operator boundaries that materialize a
19/// selection vector.  Each entry pairs a display name (used by callers via
20/// `materialize_selection_by("Name")`) with a per-query counter slot.
21///
22/// Keeping this list static lets the hot path attribute a materialization
23/// with a cheap index lookup instead of a `Mutex<HashMap>` write, while the
24/// snapshot routine still emits the same `HashMap<&'static str, u64>` shape
25/// that PROFILE tooling already consumes.
26///
27/// New callers of [`DataChunk::materialize_selection_by`] must append their
28/// label here (the lookup falls back to the unattributed path for unknown
29/// names, but adding the label keeps per-operator visibility).
30pub const SELECTION_BOUNDARY_OPS: &[&str] = &[
31    "Filter",
32    "Dedup",
33    "Assign",
34    "Remove",
35    "Unwind",
36    "AppendVertices",
37    "Sample",
38    "Root",
39    "Engine",
40    "Sort",
41    "WindowFunction",
42    "Window",
43    "TopN",
44    "Distinct",
45    "Aggregate",
46    "GroupBy",
47    "Materialize",
48    "DataCollect",
49    "RollUpApply",
50    "Sink",
51    "CopyFrom",
52    "CopyTo",
53    "Exchange",
54    "CrossJoin",
55    "SemiJoin",
56    "NestedLoopJoin",
57    "HashJoin",
58    "MergeJoin",
59    "Apply",
60    "CorrelatedApply",
61    "ShuffleJoin",
62    "Set",
63    "VectorSearch",
64    "WcoBuild",
65    "RecursiveFragment",
66    "Fulltext",
67    "Subgraph",
68    "Gather",
69    "Subquery",
70];
71
72/// Number of distinct opaque-boundary operator labels.
73const N_BOUNDARY_OPS: usize = SELECTION_BOUNDARY_OPS.len();
74
75/// Index of an opaque-boundary label inside [`SELECTION_BOUNDARY_OPS`], or
76/// `None` for ad-hoc strings not part of the closed enumeration.
77fn boundary_op_index(op: &'static str) -> Option<usize> {
78    SELECTION_BOUNDARY_OPS.iter().position(|n| *n == op)
79}
80
81/// Query-level columnar fast-path counters (observability).
82///
83/// Shared via `Arc` with the chunks produced by source operators; every
84/// `DataChunk::evaluate_expression` call records a hit (columnar batch path
85/// succeeded) or a miss (fell back to per-row evaluation). The miss rate
86/// exposes how much of the flat-column promise is actually kept.
87///
88/// `columnar_typed_hits` additionally counts evaluations served by the typed
89/// batch fast path (raw `Vec<i64>`/`Vec<f64>`/`Vec<i32>` buffers).
90/// Selection-vector counters record how often chunks are handed downstream
91/// with a selection attached vs. materialized at a boundary.
92#[derive(Debug)]
93pub struct ColumnarStats {
94    pub columnar_hits: AtomicU64,
95    pub columnar_misses: AtomicU64,
96    /// Evaluations served by the typed batch fast path.
97    pub columnar_typed_hits: AtomicU64,
98    /// Chunks handed downstream with a selection vector attached.
99    pub selection_attached: AtomicU64,
100    /// Chunks materialized at a selection boundary.
101    pub selection_materialized: AtomicU64,
102    /// evaluations served by the selection-aware visible-row fast path
103    /// (selection vector consumed without materializing).
104    pub selection_pushed: AtomicU64,
105    /// per-operator materialization counters, indexed by the position of
106    /// the label inside [`SELECTION_BOUNDARY_OPS`].  Lock-free on the hot
107    /// path; only the snapshot iterates the array (no per-call Mutex).
108    selection_materialized_by_op: [AtomicU64; N_BOUNDARY_OPS],
109    /// Chunks produced by a source via the storage column-block path (A1).
110    pub column_block_hits: AtomicU64,
111    /// Logical rows produced by expanding `DataChunk::multiplicity` at opaque
112    /// boundaries (`normalize_for_opaque`). Compared against stored rows it
113    /// yields the expanded/stored ratio for PROFILE observability.
114    pub multiplicity_expanded: AtomicU64,
115    /// Rows written to spill runs (all operators, incl. terminal collector).
116    pub spill_rows: AtomicU64,
117    /// Bytes written to spill runs (on-disk file sizes).
118    pub spill_bytes: AtomicU64,
119    /// Number of spill runs finalized.
120    pub spill_runs: AtomicU64,
121    /// Chunks emitted through the batch-slice outlet path.
122    pub batch_outlets: AtomicU64,
123    /// Expressions served by the typed columnar fast path on the compiled
124    /// (project/filter) hot path — i.e. the compiled-over-rows path was
125    /// bypassed because the chunk carried a usable typed layout. Distinct
126    /// from `columnar_typed_hits`, which counts the scalar chunk evaluator.
127    pub columnar_b_path_hits: AtomicU64,
128    /// Typed layouts built by a producer that were dropped before any
129    /// downstream consumer read them (`take_rows_for_reuse`, join build
130    /// consumption). A high count means typed columns were built in vain.
131    pub columnar_wasted_builds: AtomicU64,
132}
133
134impl Default for ColumnarStats {
135    fn default() -> Self {
136        Self {
137            columnar_hits: AtomicU64::new(0),
138            columnar_misses: AtomicU64::new(0),
139            columnar_typed_hits: AtomicU64::new(0),
140            selection_attached: AtomicU64::new(0),
141            selection_materialized: AtomicU64::new(0),
142            selection_pushed: AtomicU64::new(0),
143            selection_materialized_by_op: [const { AtomicU64::new(0) }; N_BOUNDARY_OPS],
144            column_block_hits: AtomicU64::new(0),
145            multiplicity_expanded: AtomicU64::new(0),
146            spill_rows: AtomicU64::new(0),
147            spill_bytes: AtomicU64::new(0),
148            spill_runs: AtomicU64::new(0),
149            batch_outlets: AtomicU64::new(0),
150            columnar_b_path_hits: AtomicU64::new(0),
151            columnar_wasted_builds: AtomicU64::new(0),
152        }
153    }
154}
155
156impl ColumnarStats {
157    pub fn new() -> Self {
158        Self::default()
159    }
160
161    pub fn record_hit(&self) {
162        self.columnar_hits.fetch_add(1, Ordering::Relaxed);
163    }
164
165    pub fn record_miss(&self) {
166        self.columnar_misses.fetch_add(1, Ordering::Relaxed);
167    }
168
169    /// Record an evaluation served by the typed batch fast path.
170    pub fn record_typed_hit(&self) {
171        self.columnar_typed_hits.fetch_add(1, Ordering::Relaxed);
172    }
173
174    /// Record an expression served by the typed columnar fast path on the
175    /// compiled (project/filter) hot path.
176    pub fn record_b_path_hit(&self) {
177        self.columnar_b_path_hits.fetch_add(1, Ordering::Relaxed);
178    }
179
180    /// Record a typed layout dropped before any downstream consumer read it.
181    pub fn record_wasted_build(&self) {
182        self.columnar_wasted_builds.fetch_add(1, Ordering::Relaxed);
183    }
184
185    /// Record a chunk handed downstream with a selection vector.
186    pub fn record_selection_attached(&self) {
187        self.selection_attached.fetch_add(1, Ordering::Relaxed);
188    }
189
190    /// Record a chunk materialized at a selection boundary.
191    pub fn record_selection_materialized(&self) {
192        self.selection_materialized.fetch_add(1, Ordering::Relaxed);
193    }
194
195    /// Record a chunk materialized at a named operator boundary.
196    ///
197    /// The hot path is lock-free: it bumps the global counter and, if `op`
198    /// belongs to the closed [`SELECTION_BOUNDARY_OPS`] enumeration, the
199    /// matching per-operator atomic slot.  Unknown labels still count toward
200    /// the global materialization total but are not attributed per operator.
201    pub fn record_selection_materialized_by(&self, op: &'static str) {
202        self.selection_materialized.fetch_add(1, Ordering::Relaxed);
203        if let Some(idx) = boundary_op_index(op) {
204            self.selection_materialized_by_op[idx].fetch_add(1, Ordering::Relaxed);
205        }
206    }
207
208    /// record an evaluation served by the selection-aware visible-row
209    /// fast path (selection consumed in place, upstream chunk untouched).
210    pub fn record_selection_pushed(&self) {
211        self.selection_pushed.fetch_add(1, Ordering::Relaxed);
212    }
213
214    /// materialization sites, attributed per operator.
215    pub fn materialized_by_operator(&self) -> HashMap<&'static str, u64> {
216        SELECTION_BOUNDARY_OPS
217            .iter()
218            .zip(self.selection_materialized_by_op.iter())
219            .filter(|(_, c)| c.load(Ordering::Relaxed) > 0)
220            .map(|(name, c)| (*name, c.load(Ordering::Relaxed)))
221            .collect()
222    }
223
224    /// Record a chunk produced via the storage column-block path (A1).
225    pub fn record_column_block_hit(&self) {
226        self.column_block_hits.fetch_add(1, Ordering::Relaxed);
227    }
228
229    /// Record `n` logical rows produced by multiplicity expansion.
230    pub fn record_multiplicity_expanded(&self, n: u64) {
231        if n > 0 {
232            self.multiplicity_expanded.fetch_add(n, Ordering::Relaxed);
233        }
234    }
235
236    /// Record one finalized spill run.
237    pub fn record_spill(&self, rows: u64, bytes: u64) {
238        self.record_spill_with_runs(rows, bytes, 1);
239    }
240
241    /// Record one chunk emitted through the batch-slice outlet path.
242    pub fn record_batch_outlet(&self) {
243        self.batch_outlets.fetch_add(1, Ordering::Relaxed);
244    }
245
246    /// Record aggregated spill output covering `runs` finalized runs.
247    ///
248    /// Terminal collector paths finalize N runs but report once; passing the
249    /// run count keeps `spill_runs` exact instead of undercounting to 1.
250    pub fn record_spill_with_runs(&self, rows: u64, bytes: u64, runs: u64) {
251        self.spill_rows.fetch_add(rows, Ordering::Relaxed);
252        self.spill_bytes.fetch_add(bytes, Ordering::Relaxed);
253        self.spill_runs.fetch_add(runs, Ordering::Relaxed);
254    }
255
256    /// Fraction of evaluation calls that hit the columnar fast path.
257    /// Returns 1.0 when nothing was evaluated (vacuous).
258    pub fn hit_rate(&self) -> f64 {
259        let hits = self.columnar_hits.load(Ordering::Relaxed);
260        let misses = self.columnar_misses.load(Ordering::Relaxed);
261        let total = hits + misses;
262        if total == 0 {
263            1.0
264        } else {
265            hits as f64 / total as f64
266        }
267    }
268
269    /// Fraction of evaluations served by the typed batch fast path.
270    pub fn typed_hit_rate(&self) -> f64 {
271        let hits = self.columnar_hits.load(Ordering::Relaxed);
272        let typed = self.columnar_typed_hits.load(Ordering::Relaxed);
273        if hits == 0 {
274            0.0
275        } else {
276            typed as f64 / hits as f64
277        }
278    }
279
280    /// Selection passthrough rate: attached / (attached + materialized).
281    /// 1.0 means every selected chunk reached its consumer without materializing.
282    pub fn selection_passthrough_rate(&self) -> f64 {
283        let attached = self.selection_attached.load(Ordering::Relaxed);
284        let materialized = self.selection_materialized.load(Ordering::Relaxed);
285        let total = attached + materialized;
286        if total == 0 {
287            0.0
288        } else {
289            attached as f64 / total as f64
290        }
291    }
292}
293
294/// Point-in-time snapshot of the columnar fast-path counters for PROFILE.
295///
296/// Surfaces how much of the evaluation work ran through the columnar batch
297/// fast path vs. row-wise fallback, and how often storage column blocks /
298/// selection vectors were used.  When `hit_rate` is high for typical
299/// workloads, a typed column representation becomes unnecessary; a low
300/// `typed_hit_rate` with sustained `columnar_misses` motivates revisiting it.
301#[derive(Debug, Clone, Copy, Default)]
302pub struct ColumnarStatsSnapshot {
303    pub columnar_hits: u64,
304    pub columnar_misses: u64,
305    pub columnar_typed_hits: u64,
306    pub selection_attached: u64,
307    pub selection_materialized: u64,
308    pub selection_pushed: u64,
309    pub column_block_hits: u64,
310    pub multiplicity_expanded: u64,
311    pub spill_rows: u64,
312    pub spill_bytes: u64,
313    pub spill_runs: u64,
314    pub batch_outlets: u64,
315    pub columnar_b_path_hits: u64,
316    pub columnar_wasted_builds: u64,
317}
318
319impl ColumnarStatsSnapshot {
320    pub fn from_stats(stats: &ColumnarStats) -> Self {
321        Self {
322            columnar_hits: stats.columnar_hits.load(Ordering::Relaxed),
323            columnar_misses: stats.columnar_misses.load(Ordering::Relaxed),
324            columnar_typed_hits: stats.columnar_typed_hits.load(Ordering::Relaxed),
325            selection_attached: stats.selection_attached.load(Ordering::Relaxed),
326            selection_materialized: stats.selection_materialized.load(Ordering::Relaxed),
327            selection_pushed: stats.selection_pushed.load(Ordering::Relaxed),
328            column_block_hits: stats.column_block_hits.load(Ordering::Relaxed),
329            multiplicity_expanded: stats.multiplicity_expanded.load(Ordering::Relaxed),
330            spill_rows: stats.spill_rows.load(Ordering::Relaxed),
331            spill_bytes: stats.spill_bytes.load(Ordering::Relaxed),
332            spill_runs: stats.spill_runs.load(Ordering::Relaxed),
333            batch_outlets: stats.batch_outlets.load(Ordering::Relaxed),
334            columnar_b_path_hits: stats.columnar_b_path_hits.load(Ordering::Relaxed),
335            columnar_wasted_builds: stats.columnar_wasted_builds.load(Ordering::Relaxed),
336        }
337    }
338
339    /// Fraction of evaluations served by the columnar batch fast path.
340    pub fn hit_rate(&self) -> f64 {
341        let total = self.columnar_hits + self.columnar_misses;
342        if total == 0 {
343            1.0
344        } else {
345            self.columnar_hits as f64 / total as f64
346        }
347    }
348
349    /// Fraction of evaluations served by the typed batch fast path.
350    pub fn typed_hit_rate(&self) -> f64 {
351        if self.columnar_hits == 0 {
352            0.0
353        } else {
354            self.columnar_typed_hits as f64 / self.columnar_hits as f64
355        }
356    }
357
358    /// Spilled rows per stored row for a caller-supplied stored-row count.
359    ///
360    /// Placed next to the expanded/stored ratio in PROFILE output so a large
361    /// query shows spilled/stored at a glance. Returns 0.0 when there are no
362    /// stored rows rather than dividing by zero.
363    pub fn spilled_stored_ratio(&self, stored_rows: u64) -> f64 {
364        if stored_rows == 0 {
365            0.0
366        } else {
367            self.spill_rows as f64 / stored_rows as f64
368        }
369    }
370
371    /// Human-readable one-line summary for PROFILE output.
372    pub fn summary(&self) -> String {
373        format!(
374            "columnar_hits={}, misses={}, hit_rate={:.3}, typed_hit_rate={:.3}, b_path_hits={}, wasted_builds={}, selection_attached={}, selection_materialized={}, selection_pushed={}, column_block_hits={}, multiplicity_expanded={}, spill_rows={}, spill_bytes={}, spill_runs={}, batch_outlets={}",
375            self.columnar_hits,
376            self.columnar_misses,
377            self.hit_rate(),
378            self.typed_hit_rate(),
379            self.columnar_b_path_hits,
380            self.columnar_wasted_builds,
381            self.selection_attached,
382            self.selection_materialized,
383            self.selection_pushed,
384            self.column_block_hits,
385            self.multiplicity_expanded,
386            self.spill_rows,
387            self.spill_bytes,
388            self.spill_runs,
389            self.batch_outlets,
390        )
391    }
392}
393
394#[cfg(test)]
395mod tests {
396    use super::*;
397    use std::sync::Arc;
398
399    #[test]
400    fn test_spilled_stored_ratio() {
401        let stats = ColumnarStats::new();
402        stats.record_spill_with_runs(200, 4096, 2);
403        let snapshot = ColumnarStatsSnapshot::from_stats(&stats);
404        assert_eq!(snapshot.spill_rows, 200);
405        assert_eq!(snapshot.spill_bytes, 4096);
406        assert_eq!(snapshot.spill_runs, 2);
407        assert!((snapshot.spilled_stored_ratio(100) - 2.0).abs() < f64::EPSILON);
408        assert_eq!(snapshot.spilled_stored_ratio(0), 0.0);
409        assert!(snapshot.summary().contains("spill_rows=200"));
410    }
411
412    #[test]
413    fn test_columnar_policy_flush_merges_into_shared_policy() {
414        use crate::executor::streaming::chunk::ColumnarPolicy;
415        use crate::executor::streaming::runtime::ExecutionRuntime;
416
417        let policy = Arc::new(ColumnarPolicy::new(0.8, 100));
418        let mut rt = ExecutionRuntime::default_budget();
419        rt.set_columnar_policy(Some(policy.clone()));
420
421        // Simulate a query: 10 columnar hits, 5 misses recorded on the
422        // per-query counters, plus 4 compiled-hot-path (B path) hits.
423        rt.columnar_stats().record_hit();
424        for _ in 0..9 {
425            rt.columnar_stats().record_hit();
426        }
427        for _ in 0..5 {
428            rt.columnar_stats().record_miss();
429        }
430        for _ in 0..4 {
431            rt.columnar_stats().record_b_path_hit();
432        }
433        assert_eq!(rt.columnar_stats().hit_rate(), 10.0 / 15.0);
434
435        // Flushing merges the deltas into the shared policy, with B path
436        // hits counted as effective hits.
437        rt.flush_columnar_stats_to_policy();
438        assert_eq!(policy.snapshot(), (14, 5));
439
440        // Without an injected policy the flush is a no-op.
441        let rt2 = ExecutionRuntime::default_budget();
442        rt2.flush_columnar_stats_to_policy();
443    }
444
445    #[test]
446    fn test_columnar_policy_gates_typed_columns() {
447        use crate::executor::streaming::chunk::ColumnarPolicy;
448        use crate::executor::streaming::runtime::ExecutionRuntime;
449
450        let policy = Arc::new(ColumnarPolicy::new(0.8, 100));
451        let mut rt = ExecutionRuntime::default_budget();
452        rt.set_columnar_policy(Some(policy.clone()));
453
454        // Below the sample floor the columnar path is chosen.
455        let decide = || rt.columnar_policy().is_none_or(|p| p.should_use_columnar());
456        assert!(decide());
457
458        // Simulate a fallback-heavy workload: merge 90% misses across
459        // queries; once the sample floor is crossed the decision flips.
460        for _ in 0..4 {
461            rt.flush_columnar_stats_to_policy();
462            policy.merge(3, 30);
463        }
464        assert!(!decide());
465
466        // Recovery: sustained hits flip the decision back.
467        for _ in 0..10 {
468            policy.merge(100, 0);
469        }
470        assert!(decide());
471    }
472}
linkrs_query/executor/streaming/runtime/columnar_stats.rs

linkrs_query/executor/streaming/runtime/
columnar_stats.rs
