reifydb_profiler/
spec.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::fmt::Write;
5
6use crate::{format::fmt_us, record::AggregateRecord};
7
8#[derive(Debug)]
9pub enum DimSource {
10	Text(&'static str),
11
12	Number {
13		field: &'static str,
14		prefix: &'static str,
15	},
16}
17
18impl DimSource {
19	pub fn field(&self) -> &'static str {
20		match self {
21			DimSource::Text(field) => field,
22			DimSource::Number {
23				field,
24				..
25			} => field,
26		}
27	}
28}
29
30#[derive(Debug)]
31pub struct SpanSpec {
32	pub name: &'static str,
33
34	pub duration_override: Option<&'static str>,
35	pub dims: &'static [DimSource],
36	pub extras: &'static [&'static str],
37	pub render: Option<fn(&AggregateRecord, &mut String)>,
38}
39
40static SPECS: &[SpanSpec] = &[
41	SpanSpec {
42		name: "flow::engine::apply",
43		duration_override: Some("apply_time_us"),
44		dims: &[
45			DimSource::Text("node_type"),
46			DimSource::Number {
47				field: "operator_id",
48				prefix: "op",
49			},
50		],
51		extras: &["input_rows", "output_rows", "lock_wait_us"],
52		render: Some(render_apply),
53	},
54	SpanSpec {
55		name: "flow::state::range_limited",
56		duration_override: None,
57		dims: &[
58			DimSource::Text("site"),
59			DimSource::Number {
60				field: "operator_id",
61				prefix: "op",
62			},
63		],
64		extras: &["rows_fetched", "rows_tombstoned"],
65		render: Some(render_state_range),
66	},
67	SpanSpec {
68		name: "flow::state::range",
69		duration_override: None,
70		dims: &[
71			DimSource::Text("site"),
72			DimSource::Number {
73				field: "operator_id",
74				prefix: "op",
75			},
76		],
77		extras: &["rows_fetched", "rows_tombstoned"],
78		render: Some(render_state_range),
79	},
80];
81
82pub fn spec_for(name: &str) -> Option<&'static SpanSpec> {
83	SPECS.iter().find(|spec| spec.name == name)
84}
85
86fn render_apply(record: &AggregateRecord, out: &mut String) {
87	let e = record.extras();
88	let _ = write!(out, " lock={} io={}->{}", fmt_us(e[2]), e[0], e[1]);
89}
90
91fn render_state_range(record: &AggregateRecord, out: &mut String) {
92	let e = record.extras();
93	let dead = match e[0] {
94		0 => 0.0,
95		fetched => e[1] as f64 * 100.0 / fetched as f64,
96	};
97	let per_call = match record.calls {
98		0 => 0,
99		calls => e[0] / calls,
100	};
101	let _ = write!(out, " fetched={} tomb={} dead={:.0}% rows/call={}", e[0], e[1], dead, per_call);
102}
103
104#[cfg(test)]
105mod tests {
106	use super::*;
107	use crate::record::{MAX_DIMENSIONS, MAX_EXTRAS};
108
109	#[test]
110	fn every_spec_fits_the_record_it_fills() {
111		// dims and extras are copied into fixed-size arrays on MinimalSpanRecord. A spec that
112		// declared more than the record holds would silently drop the overflow at the tail, so
113		// the widest column of a hot span would go missing rather than fail loudly.
114		for spec in SPECS {
115			assert!(
116				spec.dims.len() <= MAX_DIMENSIONS,
117				"{} declares {} dimensions but a record holds {MAX_DIMENSIONS}",
118				spec.name,
119				spec.dims.len()
120			);
121			assert!(
122				spec.extras.len() <= MAX_EXTRAS,
123				"{} declares {} extras but a record holds {MAX_EXTRAS}",
124				spec.name,
125				spec.extras.len()
126			);
127		}
128	}
129
130	#[test]
131	fn a_spec_name_resolves_only_itself() {
132		// spec_for keys off the span name, so a duplicated or prefix-colliding entry would hand
133		// the layer the wrong field wiring and mislabel every row of that span.
134		for spec in SPECS {
135			let found = spec_for(spec.name).expect("declared spec must resolve");
136			assert_eq!(found.name, spec.name);
137		}
138		assert!(spec_for("flow::engine::process_batch").is_none(), "an unlisted span must opt out");
139	}
140
141	#[test]
142	fn only_a_deliberate_span_overrides_its_own_duration() {
143		// A duration override makes the span report something narrower than its elapsed time,
144		// which hides every cost between the span boundary and the overridden measurement. It is
145		// load-bearing for flow::engine::apply and must not spread by copy-paste.
146		let overriding: Vec<&str> =
147			SPECS.iter().filter(|s| s.duration_override.is_some()).map(|s| s.name).collect();
148		assert_eq!(
149			overriding,
150			vec!["flow::engine::apply"],
151			"a new duration override needs an explicit decision, not a default"
152		);
153	}
154}
reifydb_profiler/spec.rs

reifydb_profiler/
spec.rs
