1use 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 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 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 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}