Skip to main content

reifydb_profiler/
format.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{cmp::Reverse, collections::HashMap, fmt::Write};
5
6use crate::{
7	category::{ALL_CATEGORIES, ProfilerCategory},
8	intern::DimInterner,
9	record::{AggregateRecord, DimIdx},
10	spec::spec_for,
11	summary::ProfilerSummary,
12};
13
14pub fn summary(summary: &ProfilerSummary, top_n: usize) -> String {
15	let (totals, mut hot) = flow_aggregates(summary);
16	let mut out = summary_header(&totals);
17
18	if hot.is_empty() {
19		out.push_str(" hot=[]");
20		return out;
21	}
22
23	render_hot_rows(&mut out, summary, &mut hot, top_n);
24	out
25}
26
27#[inline]
28fn summary_header(totals: &FlowTotals) -> String {
29	format!(
30		"tick(proc={}us/{}c, apply_sum={}us/{}ops, lock_sum={}us)",
31		totals.process_wall_us,
32		totals.process_calls,
33		totals.apply_total_us,
34		totals.op_calls,
35		totals.lock_total_us,
36	)
37}
38
39#[inline]
40fn render_hot_rows(out: &mut String, summary: &ProfilerSummary, hot: &mut [(FlowKey, HotEntry)], top_n: usize) {
41	hot.sort_by(|a, b| b.1.apply_us.cmp(&a.1.apply_us));
42	out.push_str(" hot=[");
43	for (i, (key, entry)) in hot.iter().take(top_n).enumerate() {
44		if i > 0 {
45			out.push_str(", ");
46		}
47		let label = resolve_label(summary.interner.as_deref(), key.0);
48		let id = resolve_id(summary.interner.as_deref(), key.1);
49		let _ = write!(
50			out,
51			"{}@{}={}us/{}lk/{}c/{}in/{}out",
52			label, id, entry.apply_us, entry.lock_us, entry.calls, entry.input_rows, entry.output_rows,
53		);
54	}
55	out.push(']');
56}
57
58pub fn summary_table(summary: &ProfilerSummary, top_n: usize) -> String {
59	let mut out = String::new();
60	let _ = writeln!(
61		out,
62		"profile scope={} total={}",
63		summary.scope_name,
64		fmt_us(summary.total_duration.microseconds().unwrap_or(0) as u64)
65	);
66
67	for cat in ALL_CATEGORIES {
68		let cat_summary = summary.category(cat);
69		if cat_summary.calls == 0 {
70			continue;
71		}
72
73		match cat {
74			ProfilerCategory::Flow => {
75				let _ = writeln!(
76					out,
77					"  {}: {} calls, apply={}, lock={}",
78					category_label(cat),
79					cat_summary.calls,
80					fmt_us(cat_summary.total_us),
81					fmt_us(cat_summary.extras_sum[2]),
82				);
83				render_flow_rows(&mut out, summary, top_n);
84			}
85			_ => {
86				let _ = writeln!(
87					out,
88					"  {}: {} calls, total={}",
89					category_label(cat),
90					cat_summary.calls,
91					fmt_us(cat_summary.total_us),
92				);
93				render_non_flow_rows(&mut out, summary, cat, top_n);
94			}
95		}
96	}
97
98	out
99}
100
101pub fn aggregates_table(records: &[AggregateRecord], top_n: usize) -> String {
102	let mut out = String::new();
103	if records.is_empty() {
104		out.push_str("profile (accumulator) empty\n");
105		return out;
106	}
107	write_accumulator_header(&mut out, records);
108
109	for cat in ALL_CATEGORIES {
110		render_category(&mut out, records, cat, top_n);
111	}
112
113	out
114}
115
116#[inline]
117fn write_accumulator_header(out: &mut String, records: &[AggregateRecord]) {
118	let total_calls: u64 = records.iter().map(|r| r.calls).sum();
119	let self_us: u64 = records.iter().map(|r| r.self_us).sum();
120	let _ = writeln!(
121		out,
122		"profile (accumulator) {} records, {} calls, self={}",
123		records.len(),
124		total_calls,
125		fmt_us(self_us)
126	);
127}
128
129#[inline]
130fn render_category(out: &mut String, records: &[AggregateRecord], cat: ProfilerCategory, top_n: usize) {
131	let cat_records: Vec<&AggregateRecord> = records.iter().filter(|r| r.category == cat).collect();
132	if cat_records.is_empty() {
133		return;
134	}
135	let cat_calls: u64 = cat_records.iter().map(|r| r.calls).sum();
136	let cat_self: u64 = cat_records.iter().map(|r| r.self_us).sum();
137	let _ = writeln!(
138		out,
139		"  {}: {} records, {} calls, self={}",
140		category_label(cat),
141		cat_records.len(),
142		cat_calls,
143		fmt_us(cat_self)
144	);
145
146	let mut by_name: HashMap<&str, Vec<&AggregateRecord>> = HashMap::new();
147	for r in &cat_records {
148		by_name.entry(r.span_name.as_str()).or_default().push(*r);
149	}
150	let mut groups: Vec<(&str, Vec<&AggregateRecord>)> = by_name.into_iter().collect();
151	groups.sort_by_key(|(_, recs)| Reverse(recs.iter().map(|r| r.total_us).sum::<u64>()));
152
153	for (span_name, group) in groups {
154		render_group(out, span_name, group, top_n);
155	}
156}
157
158#[inline]
159fn render_group(out: &mut String, span_name: &str, mut group: Vec<&AggregateRecord>, top_n: usize) {
160	let group_total: u64 = group.iter().map(|r| r.total_us).sum();
161	let group_self: u64 = group.iter().map(|r| r.self_us).sum();
162	let group_calls: u64 = group.iter().map(|r| r.calls).sum();
163
164	if group.len() == 1 && group[0].dimensions.is_empty() {
165		let r = group[0];
166		let p = r.histogram.percentiles();
167		let _ = writeln!(
168			out,
169			"    {}  total={} self={} calls={} p50={} p75={} p90={} p95={} p99={}",
170			span_name,
171			fmt_us(r.total_us),
172			fmt_us(r.self_us),
173			r.calls,
174			fmt_us(p.p50 as u64),
175			fmt_us(p.p75 as u64),
176			fmt_us(p.p90 as u64),
177			fmt_us(p.p95 as u64),
178			fmt_us(p.p99 as u64),
179		);
180		return;
181	}
182
183	let _ = writeln!(
184		out,
185		"    {} [{} ops, total={}, self={}, calls={}]",
186		span_name,
187		group.len(),
188		fmt_us(group_total),
189		fmt_us(group_self),
190		group_calls,
191	);
192
193	group.sort_by(|a, b| b.total_us.cmp(&a.total_us));
194	group.truncate(top_n);
195
196	let labels: Vec<String> = group
197		.iter()
198		.map(|r| {
199			if r.dimensions.is_empty() {
200				"<no-dims>".to_string()
201			} else {
202				r.dimensions.join("@")
203			}
204		})
205		.collect();
206	let max_label_width = labels.iter().map(|s| s.len()).max().unwrap_or(0);
207
208	let render_extras = spec_for(span_name).and_then(|spec| spec.render);
209	for (i, r) in group.iter().enumerate() {
210		let p = r.histogram.percentiles();
211		let _ = write!(
212			out,
213			"      {:<width$}  total={} self={} calls={} p50={} p75={} p90={} p95={} p99={}",
214			labels[i],
215			fmt_us(r.total_us),
216			fmt_us(r.self_us),
217			r.calls,
218			fmt_us(p.p50 as u64),
219			fmt_us(p.p75 as u64),
220			fmt_us(p.p90 as u64),
221			fmt_us(p.p95 as u64),
222			fmt_us(p.p99 as u64),
223			width = max_label_width,
224		);
225		if let Some(render) = render_extras {
226			render(r, out);
227		}
228		let _ = writeln!(out);
229	}
230}
231
232pub fn fmt_us(us: u64) -> String {
233	if us < 1_000 {
234		format!("{}us", us)
235	} else if us < 1_000_000 {
236		format!("{:.1}ms", us as f64 / 1_000.0)
237	} else {
238		format!("{:.1}s", us as f64 / 1_000_000.0)
239	}
240}
241
242#[derive(Default)]
243struct FlowTotals {
244	apply_total_us: u64,
245	op_calls: u32,
246	process_wall_us: u64,
247	process_calls: u32,
248	lock_total_us: u64,
249}
250
251#[derive(Default, Clone)]
252struct HotEntry {
253	apply_us: u64,
254	lock_us: u64,
255	calls: u32,
256	input_rows: u64,
257	output_rows: u64,
258}
259
260type FlowKey = (DimIdx, DimIdx);
261
262fn flow_aggregates(summary: &ProfilerSummary) -> (FlowTotals, Vec<(FlowKey, HotEntry)>) {
263	let mut totals = FlowTotals::default();
264	let mut aggregates: HashMap<FlowKey, HotEntry> = HashMap::new();
265
266	for r in &summary.records {
267		if r.category_id != ProfilerCategory::Flow as u8 {
268			continue;
269		}
270		let is_apply = r.dim_indices[0] != 0 || r.dim_indices[1] != 0;
271		if is_apply {
272			totals.apply_total_us = totals.apply_total_us.saturating_add(r.duration_us as u64);
273			totals.op_calls = totals.op_calls.saturating_add(1);
274			totals.lock_total_us = totals.lock_total_us.saturating_add(r.extras[2]);
275			let entry = aggregates.entry((r.dim_indices[0], r.dim_indices[1])).or_default();
276			entry.apply_us = entry.apply_us.saturating_add(r.duration_us as u64);
277			entry.lock_us = entry.lock_us.saturating_add(r.extras[2]);
278			entry.calls = entry.calls.saturating_add(1);
279			entry.input_rows = entry.input_rows.saturating_add(r.extras[0]);
280			entry.output_rows = entry.output_rows.saturating_add(r.extras[1]);
281		} else {
282			totals.process_wall_us = totals.process_wall_us.saturating_add(r.duration_us as u64);
283			totals.process_calls = totals.process_calls.saturating_add(1);
284		}
285	}
286
287	(totals, aggregates.into_iter().collect())
288}
289
290fn render_flow_rows(out: &mut String, summary: &ProfilerSummary, top_n: usize) {
291	let (_, mut hot) = flow_aggregates(summary);
292	hot.sort_by(|a, b| b.1.apply_us.cmp(&a.1.apply_us));
293	hot.truncate(top_n);
294
295	if hot.is_empty() {
296		return;
297	}
298
299	let labels: Vec<String> = hot
300		.iter()
301		.map(|(key, _)| {
302			let label = resolve_label(summary.interner.as_deref(), key.0);
303			let id = resolve_id(summary.interner.as_deref(), key.1);
304			format!("{}@{}", label, id)
305		})
306		.collect();
307	let max_label_width = labels.iter().map(|s| s.len()).max().unwrap_or(0);
308
309	for (i, (_, entry)) in hot.iter().enumerate() {
310		let _ = writeln!(
311			out,
312			"    {:<width$}  apply={} calls={} lock={} io={}->{}",
313			labels[i],
314			fmt_us(entry.apply_us),
315			entry.calls,
316			fmt_us(entry.lock_us),
317			entry.input_rows,
318			entry.output_rows,
319			width = max_label_width,
320		);
321	}
322}
323
324fn render_non_flow_rows(out: &mut String, summary: &ProfilerSummary, cat: ProfilerCategory, top_n: usize) {
325	let mut agg: HashMap<u64, (u64, u64)> = HashMap::new();
326	for r in &summary.records {
327		if r.category_id != cat as u8 {
328			continue;
329		}
330		let entry = agg.entry(r.callsite_id).or_insert((0, 0));
331		entry.0 = entry.0.saturating_add(r.duration_us as u64);
332		entry.1 = entry.1.saturating_add(1);
333	}
334	let mut sorted: Vec<(u64, (u64, u64))> = agg.into_iter().collect();
335	sorted.sort_by(|a, b| b.1.0.cmp(&a.1.0));
336	sorted.truncate(top_n);
337
338	if sorted.is_empty() {
339		return;
340	}
341
342	let labels: Vec<String> = sorted.iter().map(|(callsite, _)| format!("span#{}", callsite)).collect();
343	let max_label_width = labels.iter().map(|s| s.len()).max().unwrap_or(0);
344
345	for (i, (_, (total, calls))) in sorted.iter().enumerate() {
346		let _ = writeln!(
347			out,
348			"    {:<width$}  total={} calls={}",
349			labels[i],
350			fmt_us(*total),
351			calls,
352			width = max_label_width,
353		);
354	}
355}
356
357fn resolve_label(interner: Option<&DimInterner>, idx: DimIdx) -> String {
358	let resolved = interner.and_then(|i| i.resolve(idx));
359	match resolved {
360		Some(s) if !s.is_empty() => s,
361		_ => "?".to_string(),
362	}
363}
364
365fn resolve_id(interner: Option<&DimInterner>, idx: DimIdx) -> String {
366	interner.and_then(|i| i.resolve(idx)).filter(|s| !s.is_empty()).unwrap_or_else(|| idx.to_string())
367}
368
369fn category_label(c: ProfilerCategory) -> &'static str {
370	match c {
371		ProfilerCategory::Query => "Query",
372		ProfilerCategory::Txn => "Txn",
373		ProfilerCategory::Storage => "Storage",
374		ProfilerCategory::Plan => "Plan",
375		ProfilerCategory::Cdc => "Cdc",
376		ProfilerCategory::Flow => "Flow",
377		ProfilerCategory::Subscription => "Subscription",
378		ProfilerCategory::Server => "Server",
379		ProfilerCategory::Wire => "Wire",
380		ProfilerCategory::Auth => "Auth",
381		ProfilerCategory::Catalog => "Catalog",
382		ProfilerCategory::Engine => "Engine",
383		ProfilerCategory::Mutate => "Mutate",
384		ProfilerCategory::Transport => "Transport",
385		ProfilerCategory::Task => "Task",
386		ProfilerCategory::Policy => "Policy",
387		ProfilerCategory::ExternC => "ExternC",
388		ProfilerCategory::Cache => "Cache",
389		ProfilerCategory::RowShape => "RowShape",
390		ProfilerCategory::Api => "Api",
391		ProfilerCategory::Actor => "Actor",
392		ProfilerCategory::Lifecycle => "Lifecycle",
393	}
394}
395
396#[cfg(test)]
397mod tests {
398	use std::sync::Arc;
399
400	use reifydb_value::value::duration::Duration;
401
402	use super::*;
403	use crate::{
404		category::{CATEGORY_COUNT, ProfilerCategory},
405		intern::DimInterner,
406		percentile::PercentileHistogram,
407		record::{AggregateRecord, DIM_UNSET, MAX_EXTRAS, MinimalSpanRecord},
408		scope::ScopeId,
409		summary::CategorySummary,
410	};
411
412	fn empty_summary() -> ProfilerSummary {
413		ProfilerSummary {
414			scope_id: ScopeId(1),
415			scope_name: "x",
416			started_at_nanos: 0,
417			total_duration: Duration::zero(),
418			records: Vec::new(),
419			per_category: [CategorySummary::default(); CATEGORY_COUNT],
420			interner: None,
421		}
422	}
423
424	fn summary_with(records: Vec<MinimalSpanRecord>, interner: Option<Arc<DimInterner>>) -> ProfilerSummary {
425		ProfilerSummary::from_records(
426			ScopeId(1),
427			"chaindex.batch_commit",
428			0,
429			Duration::from_microseconds(12_345).unwrap(),
430			records,
431			interner,
432		)
433	}
434
435	#[test]
436	fn empty_summary_renders_hot_empty() {
437		let s = empty_summary();
438		assert!(summary(&s, 5).ends_with(" hot=[]"));
439	}
440
441	#[test]
442	fn summary_resolves_labels_when_interner_present() {
443		let interner = Arc::new(DimInterner::new());
444		let type_idx = interner.intern("map");
445		let id_idx = interner.intern("n1");
446
447		let apply_rec = MinimalSpanRecord::new(ProfilerCategory::Flow, 100, 500)
448			.with_dimensions([type_idx, id_idx])
449			.with_extras([10, 7, 50, 0]);
450		let s = summary_with(vec![apply_rec], Some(Arc::clone(&interner)));
451
452		let line = summary(&s, 5);
453		assert!(line.contains("map@n1=500us/50lk/1c/10in/7out"), "got {}", line);
454	}
455
456	#[test]
457	fn summary_falls_back_to_placeholder_when_interner_missing() {
458		let apply_rec = MinimalSpanRecord::new(ProfilerCategory::Flow, 100, 500)
459			.with_dimensions([42, 43])
460			.with_extras([1, 1, 1, 0]);
461		let s = summary_with(vec![apply_rec], None);
462
463		let line = summary(&s, 5);
464		assert!(line.contains("?@43=500us/1lk/1c/1in/1out"), "got {}", line);
465	}
466
467	#[test]
468	fn summary_separates_process_and_apply() {
469		let process_rec = MinimalSpanRecord::new(ProfilerCategory::Flow, 200, 1000);
470		let apply_rec = MinimalSpanRecord::new(ProfilerCategory::Flow, 100, 400)
471			.with_dimensions([1, 2])
472			.with_extras([5, 3, 25, 0]);
473		let s = summary_with(vec![process_rec, apply_rec], None);
474
475		let line = summary(&s, 5);
476		assert!(line.starts_with("tick(proc=1000us/1c, apply_sum=400us/1ops, lock_sum=25us)"), "got {}", line);
477	}
478
479	#[test]
480	fn summary_aggregates_repeated_apply_per_operator() {
481		let interner = Arc::new(DimInterner::new());
482		let t = interner.intern("filter");
483		let i = interner.intern("n2");
484
485		let recs = vec![
486			MinimalSpanRecord::new(ProfilerCategory::Flow, 100, 100)
487				.with_dimensions([t, i])
488				.with_extras([5, 3, 10, 0]),
489			MinimalSpanRecord::new(ProfilerCategory::Flow, 100, 200)
490				.with_dimensions([t, i])
491				.with_extras([7, 5, 15, 0]),
492		];
493		let s = summary_with(recs, Some(Arc::clone(&interner)));
494
495		let line = summary(&s, 5);
496		assert!(line.contains("filter@n2=300us/25lk/2c/12in/8out"), "got {}", line);
497	}
498
499	#[test]
500	fn summary_table_renders_multi_line_with_categories() {
501		let interner = Arc::new(DimInterner::new());
502		let map_t = interner.intern("map");
503		let map_id = interner.intern("n1");
504		let filter_t = interner.intern("filter");
505		let filter_id = interner.intern("n2");
506
507		let recs = vec![
508			MinimalSpanRecord::new(ProfilerCategory::Flow, 100, 5000)
509				.with_dimensions([map_t, map_id])
510				.with_extras([10, 7, 100, 0]),
511			MinimalSpanRecord::new(ProfilerCategory::Flow, 100, 3000)
512				.with_dimensions([filter_t, filter_id])
513				.with_extras([5, 3, 50, 0]),
514			MinimalSpanRecord::new(ProfilerCategory::Storage, 200, 1500),
515			MinimalSpanRecord::new(ProfilerCategory::Storage, 201, 600),
516		];
517		let s = summary_with(recs, Some(Arc::clone(&interner)));
518		let table = summary_table(&s, 5);
519
520		assert!(table.starts_with("profile scope=chaindex.batch_commit total="), "first line: {}", table);
521		assert!(table.contains("Flow: 2 calls, apply="), "flow header missing: {}", table);
522		assert!(table.contains("map@n1"), "map@n1 missing: {}", table);
523		assert!(table.contains("filter@n2"), "filter@n2 missing: {}", table);
524		assert!(table.contains("io=10->7"), "io rendering missing: {}", table);
525		assert!(table.contains("Storage: 2 calls, total="), "storage header missing: {}", table);
526		assert!(!table.contains('\u{2192}'), "unicode arrow leaked into ASCII output");
527	}
528
529	#[test]
530	fn summary_table_aligns_labels_within_category() {
531		let interner = Arc::new(DimInterner::new());
532		let short = interner.intern("a");
533		let long = interner.intern("very_long_type");
534		let short_id = interner.intern("1");
535		let long_id = interner.intern("z");
536
537		let recs = vec![
538			MinimalSpanRecord::new(ProfilerCategory::Flow, 100, 100)
539				.with_dimensions([short, short_id])
540				.with_extras([0, 0, 0, 0]),
541			MinimalSpanRecord::new(ProfilerCategory::Flow, 100, 200)
542				.with_dimensions([long, long_id])
543				.with_extras([0, 0, 0, 0]),
544		];
545		let s = summary_with(recs, Some(Arc::clone(&interner)));
546		let table = summary_table(&s, 5);
547
548		let lines: Vec<&str> = table.lines().collect();
549		let short_line = lines.iter().find(|l| l.contains("a@1 ")).expect("short label line");
550		let long_line = lines.iter().find(|l| l.contains("very_long_type@z")).expect("long label line");
551		let short_apply_pos = short_line.find("apply=").unwrap();
552		let long_apply_pos = long_line.find("apply=").unwrap();
553		assert_eq!(
554			short_apply_pos, long_apply_pos,
555			"apply= columns are not aligned:\n{}\n{}",
556			short_line, long_line
557		);
558	}
559
560	#[test]
561	fn fmt_us_unit_promotion() {
562		assert_eq!(fmt_us(0), "0us");
563		assert_eq!(fmt_us(500), "500us");
564		assert_eq!(fmt_us(1_500), "1.5ms");
565		assert_eq!(fmt_us(12_345), "12.3ms");
566		assert_eq!(fmt_us(1_500_000), "1.5s");
567	}
568
569	#[test]
570	fn aggregates_table_renders_per_category() {
571		let records = vec![
572			AggregateRecord {
573				category: ProfilerCategory::Flow,
574				span_name: "flow::engine::process_batch".to_string(),
575				dimensions: Vec::new(),
576				calls: 6,
577				total_us: 1_000,
578				self_us: 1_000,
579				histogram: PercentileHistogram::new(),
580				extras_sum: [0; MAX_EXTRAS],
581			},
582			AggregateRecord {
583				category: ProfilerCategory::Flow,
584				span_name: "flow::engine::apply".to_string(),
585				dimensions: vec!["map".to_string(), "n1".to_string()],
586				calls: 3,
587				total_us: 5_000,
588				self_us: 5_000,
589				histogram: PercentileHistogram::new(),
590				extras_sum: [0; MAX_EXTRAS],
591			},
592			AggregateRecord {
593				category: ProfilerCategory::Flow,
594				span_name: "flow::engine::apply".to_string(),
595				dimensions: vec!["filter".to_string(), "n2".to_string()],
596				calls: 2,
597				total_us: 3_000,
598				self_us: 3_000,
599				histogram: PercentileHistogram::new(),
600				extras_sum: [0; MAX_EXTRAS],
601			},
602			AggregateRecord {
603				category: ProfilerCategory::Storage,
604				span_name: "store::multi::write".to_string(),
605				dimensions: Vec::new(),
606				calls: 30,
607				total_us: 1_500,
608				self_us: 1_500,
609				histogram: PercentileHistogram::new(),
610				extras_sum: [0; MAX_EXTRAS],
611			},
612		];
613		let table = aggregates_table(&records, 10);
614		assert!(table.starts_with("profile (accumulator) 4 records, 41 calls, self="));
615		assert!(table.contains("Flow: 3 records, 11 calls, self="));
616		assert!(table.contains("flow::engine::apply [2 ops, total="));
617		assert!(table.contains("\n      map@n1  "), "expected nested map@n1 row, got:\n{}", table);
618		assert!(table.contains("\n      filter@n2  "), "expected nested filter@n2 row, got:\n{}", table);
619		assert!(table.contains("\n    flow::engine::process_batch  total="));
620		assert!(table.contains("Storage: 1 records, 30 calls, self="));
621		assert!(table.contains("\n    store::multi::write  total="));
622	}
623
624	#[test]
625	fn aggregates_table_groups_flow_apply_by_operator() {
626		let mk = |op: &str, total: u64| AggregateRecord {
627			category: ProfilerCategory::Flow,
628			span_name: "flow::engine::apply".to_string(),
629			dimensions: vec![op.to_string()],
630			calls: 1,
631			total_us: total,
632			self_us: total,
633			histogram: PercentileHistogram::new(),
634			extras_sum: [0; MAX_EXTRAS],
635		};
636		let records = vec![mk("op_a", 4_000), mk("op_b", 3_000), mk("op_c", 2_000), mk("op_d", 1_000)];
637		let table = aggregates_table(&records, 2);
638		assert!(table.contains("flow::engine::apply [4 ops, total="));
639		assert!(table.contains("\n      op_a  "));
640		assert!(table.contains("\n      op_b  "));
641		assert!(!table.contains("\n      op_c  "), "op_c should be truncated by top_n=2: {}", table);
642		assert!(!table.contains("\n      op_d  "), "op_d should be truncated by top_n=2: {}", table);
643	}
644
645	#[test]
646	fn an_inline_row_reports_self_apart_from_its_inclusive_total() {
647		// On a saturated box the actionable number is self, not the inclusive total: a span that
648		// spends all of its time inside children burns no CPU of its own. Rendering total into both
649		// slots would read as "range_limited burns 1.0ms" when it burns 200us, so they must differ.
650		let record = AggregateRecord {
651			category: ProfilerCategory::Flow,
652			span_name: "flow::state::range_limited".to_string(),
653			dimensions: Vec::new(),
654			calls: 1,
655			total_us: 1_000,
656			self_us: 200,
657			histogram: PercentileHistogram::new(),
658			extras_sum: [0; MAX_EXTRAS],
659		};
660
661		let table = aggregates_table(&[record], 10);
662
663		assert!(
664			table.contains("flow::state::range_limited  total=1.0ms self=200us"),
665			"an inline row must carry its own self time beside the inclusive total:\n{}",
666			table
667		);
668	}
669
670	#[test]
671	fn a_dimensioned_group_reports_self_per_row_and_in_its_header() {
672		// Grouped rows are where operator attribution is read off, so each dimension needs its own
673		// self and the header needs the group's sum. Summing totals into the header self would say
674		// 7.0ms of CPU across two operators that together burn 1.5ms.
675		let mk = |op: &str, total_us: u64, self_us: u64| AggregateRecord {
676			category: ProfilerCategory::Flow,
677			span_name: "flow::engine::apply".to_string(),
678			dimensions: vec![op.to_string()],
679			calls: 1,
680			total_us,
681			self_us,
682			histogram: PercentileHistogram::new(),
683			extras_sum: [0; MAX_EXTRAS],
684		};
685
686		let table = aggregates_table(&[mk("op_a", 4_000, 1_000), mk("op_b", 3_000, 500)], 10);
687
688		assert!(
689			table.contains("flow::engine::apply [2 ops, total=7.0ms, self=1.5ms, calls=2]"),
690			"the group header must sum self across its rows:\n{}",
691			table
692		);
693		assert!(table.contains("op_a  total=4.0ms self=1.0ms"), "op_a row lost its self time:\n{}", table);
694		assert!(table.contains("op_b  total=3.0ms self=500us"), "op_b row lost its self time:\n{}", table);
695	}
696
697	#[test]
698	fn a_category_total_counts_each_span_once_across_nesting() {
699		// A parent's inclusive total already contains its children, so summing totals counts every
700		// leaf once per ancestor above it. That is what reported 2.3s of Flow work inside a 10s
701		// window and made Join look like a larger share of the engine than it is.
702		let parent = AggregateRecord {
703			category: ProfilerCategory::Flow,
704			span_name: "flow::operator::join::insert".to_string(),
705			dimensions: Vec::new(),
706			calls: 1,
707			total_us: 1_000,
708			self_us: 200,
709			histogram: PercentileHistogram::new(),
710			extras_sum: [0; MAX_EXTRAS],
711		};
712		let child = AggregateRecord {
713			span_name: "flow::operator::join::store::put_row".to_string(),
714			total_us: 800,
715			self_us: 800,
716			..parent.clone()
717		};
718
719		let table = aggregates_table(&[parent, child], 10);
720
721		assert!(
722			table.contains("Flow: 2 records, 2 calls, self=1.0ms"),
723			"the category must report 200us of parent work plus 800us of child work, not 1.8ms:\n{}",
724			table
725		);
726		assert!(
727			table.contains("flow::operator::join::insert  total=1.0ms"),
728			"per-span rows stay inclusive so the nesting is still readable:\n{}",
729			table
730		);
731	}
732
733	#[test]
734	fn aggregates_table_single_no_dim_record_renders_inline() {
735		let records = vec![AggregateRecord {
736			category: ProfilerCategory::Flow,
737			span_name: "flow::engine::process_batch".to_string(),
738			dimensions: Vec::new(),
739			calls: 5,
740			total_us: 800,
741			self_us: 800,
742			histogram: PercentileHistogram::new(),
743			extras_sum: [0; MAX_EXTRAS],
744		}];
745		let table = aggregates_table(&records, 10);
746		assert!(!table.contains("[1 ops"), "single no-dim record must render inline, got:\n{}", table);
747		assert!(table.contains("\n    flow::engine::process_batch  total="));
748	}
749
750	#[test]
751	fn aggregates_table_handles_empty() {
752		let table = aggregates_table(&[], 10);
753		assert!(table.contains("empty"));
754	}
755
756	#[test]
757	fn summary_table_skips_empty_categories() {
758		let recs =
759			vec![MinimalSpanRecord::new(ProfilerCategory::Flow, 100, 0)
760				.with_dimensions([DIM_UNSET, DIM_UNSET])];
761		let s = summary_with(recs, None);
762		let table = summary_table(&s, 5);
763		assert!(table.contains("Flow:"));
764		assert!(!table.contains("Query:"));
765		assert!(!table.contains("Storage:"));
766	}
767}