src/warehouse/fn_trace.rs
//! **Traced runs and profiles, filed** — the warehouse half of a Runa `Trace`
//! (LLVM XRay) or `Profile` (`perf record -g`) step, read back by the 3D code
//! map's replay (`Viz.CodeMap` lods `trace_runs` / `trace` / `profile`).
//!
//! | table | one row per | partitioned by |
//! |---|---|---|
//! | `fn_trace_events` | function entry / exit of a run: `(repo, commit, run_id, thread, seq_no, t_ns, stable_id, kind)` + `symbol` | `run_id` (a replay reads one run) |
//! | `fn_trace_runs` | filed chunk of a run: what ran, its events, threads, span, how many of its functions the room knew | `repo` |
//! | `fn_profile` | function of a profile: self / total time and samples | `repo` |
//!
//! `stable_id` is the code room function the event's function is (its moniker
//! with the version as `*`, `mimir_schema::moniker::stable`), NULL when the room
//! knew none (std, a registry crate, a function the room was not read at). The
//! mapping is the filer's: it holds the room ([`crate::code_room`]).
//!
//! A run is filed in self-standing chunks (`nornir_trace::wire::ChunkFiling`);
//! each chunk is idempotent on `(run_id, chunk)`.
use std::collections::HashMap;
use std::sync::Arc;
use anyhow::{Context, Result};
use arrow::array::{Array, BooleanArray, Int64Array, RecordBatch, StringArray, TimestampMicrosecondArray};
use futures::TryStreamExt;
use iceberg::Catalog;
use iceberg::arrow::schema_to_arrow_schema;
use iceberg::expr::Reference;
use iceberg::spec::{Datum, NestedField, PrimitiveType, Schema, Type};
use nornir_trace::EventKind;
use super::iceberg::{IcebergWarehouse, append_batch, ensure_table_schema};
pub(crate) const TABLE_FN_TRACE_EVENTS: &str = "fn_trace_events";
pub(crate) const TABLE_FN_TRACE_RUNS: &str = "fn_trace_runs";
pub(crate) const TABLE_FN_PROFILE: &str = "fn_profile";
fn req(id: i32, name: &str, t: PrimitiveType) -> Arc<NestedField> {
Arc::new(NestedField::required(id, name, Type::Primitive(t)))
}
fn opt(id: i32, name: &str, t: PrimitiveType) -> Arc<NestedField> {
Arc::new(NestedField::optional(id, name, Type::Primitive(t)))
}
/// `fn_trace_events`: one function entry or exit of a traced run.
pub fn events_schema() -> Result<Schema> {
use PrimitiveType::*;
Ok(Schema::builder()
.with_schema_id(0)
.with_fields(vec![
req(1, "repo", String),
opt(2, "commit", String),
req(3, "run_id", String),
req(4, "thread", Long),
req(5, "seq_no", Long),
req(6, "t_ns", Long),
opt(7, "stable_id", String),
req(8, "kind", String),
req(9, "symbol", String),
])
.build()?)
}
/// `fn_trace_runs`: one filed chunk of a traced run.
pub fn runs_schema() -> Result<Schema> {
use PrimitiveType::*;
Ok(Schema::builder()
.with_schema_id(0)
.with_fields(vec![
req(1, "run_id", String),
req(2, "ts_micros", Timestamptz),
req(3, "repo", String),
opt(4, "commit", String),
req(5, "target", String),
req(6, "source", String),
req(7, "chunk", Long),
req(8, "chunks", Long),
req(9, "events", Long),
req(10, "mapped_events", Long),
req(11, "functions", Long),
req(12, "mapped_functions", Long),
req(13, "local_functions", Long),
req(14, "mapped_local_functions", Long),
req(15, "threads", Long),
req(16, "first_ns", Long),
req(17, "last_ns", Long),
req(18, "total_events", Long),
req(19, "duration_ns", Long),
req(20, "truncated", Boolean),
req(21, "binaries", String),
])
.build()?)
}
/// `fn_profile`: one function's share of a sampled profile.
pub fn profile_schema() -> Result<Schema> {
use PrimitiveType::*;
Ok(Schema::builder()
.with_schema_id(0)
.with_fields(vec![
req(1, "run_id", String),
req(2, "ts_micros", Timestamptz),
req(3, "repo", String),
opt(4, "commit", String),
req(5, "target", String),
opt(6, "stable_id", String),
req(7, "symbol", String),
req(8, "self_ns", Long),
req(9, "total_ns", Long),
req(10, "self_samples", Long),
req(11, "total_samples", Long),
req(12, "samples", Long),
req(13, "profile_ns", Long),
])
.build()?)
}
/// One event row as filed / read.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EventRow {
pub thread: u32,
pub seq_no: u64,
pub t_ns: u64,
/// `""` = the room knew no function for it.
pub stable_id: String,
pub kind: EventKind,
pub symbol: String,
}
/// One chunk's summary row.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct RunRow {
pub run_id: String,
pub ts_micros: i64,
pub repo: String,
pub commit: String,
pub target: String,
pub source: String,
pub chunk: u32,
pub chunks: u32,
pub events: u64,
pub mapped_events: u64,
pub functions: u64,
pub mapped_functions: u64,
pub local_functions: u64,
pub mapped_local_functions: u64,
pub threads: u64,
pub first_ns: u64,
pub last_ns: u64,
pub total_events: u64,
pub duration_ns: u64,
pub truncated: bool,
/// The traced binaries (`nornir_trace::TracedBinary` list) as JSON.
pub binaries: String,
}
/// One profile row.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ProfileRow {
pub stable_id: String,
pub symbol: String,
pub self_ns: u64,
pub total_ns: u64,
pub self_samples: u64,
pub total_samples: u64,
}
fn strs<T>(rows: &[T], f: impl Fn(&T) -> String) -> Arc<dyn Array> {
Arc::new(StringArray::from(rows.iter().map(f).collect::<Vec<_>>()))
}
fn opt_strs<T>(rows: &[T], f: impl Fn(&T) -> String) -> Arc<dyn Array> {
Arc::new(StringArray::from(rows.iter().map(|r| Some(f(r)).filter(|s| !s.is_empty())).collect::<Vec<_>>()))
}
fn longs<T>(rows: &[T], f: impl Fn(&T) -> u64) -> Arc<dyn Array> {
Arc::new(Int64Array::from(rows.iter().map(|r| f(r) as i64).collect::<Vec<_>>()))
}
fn ts<T>(rows: &[T], v: i64) -> Arc<dyn Array> {
Arc::new(TimestampMicrosecondArray::from(vec![v; rows.len()]).with_timezone("+00:00"))
}
async fn append(wh: &IcebergWarehouse, table: &str, schema: Schema, cols: Vec<Arc<dyn Array>>) -> Result<()> {
let ident = wh.table_ident(table);
let t = wh.catalog().load_table(&ident).await?;
let t = ensure_table_schema(wh.catalog(), &ident, t, &schema).await?;
let arrow_schema = Arc::new(schema_to_arrow_schema(t.metadata().current_schema())?);
let batch = RecordBatch::try_new(arrow_schema, cols)?;
append_batch(wh.catalog(), t, batch).await
}
/// Has chunk `chunk` of `run_id` been filed? (A retried filing writes nothing.)
pub async fn chunk_filed(wh: &IcebergWarehouse, run_id: &str, chunk: u32) -> Result<bool> {
let t = wh.catalog().load_table(&wh.table_ident(TABLE_FN_TRACE_RUNS)).await?;
let scan = t
.scan()
.with_filter(Reference::new("run_id").equal_to(Datum::string(run_id)))
.select(["run_id", "chunk"])
.build()?;
let batches: Vec<RecordBatch> = scan.to_arrow().await?.try_collect().await?;
for b in &batches {
let (ids, chunks) = (str_col(b, "run_id")?, i64_col(b, "chunk")?);
if (0..b.num_rows()).any(|i| ids.value(i) == run_id && chunks.value(i) == chunk as i64) {
return Ok(true);
}
}
Ok(false)
}
/// **File one chunk**: its events (one snapshot) and its summary row.
pub async fn append_chunk(wh: &IcebergWarehouse, repo: &str, commit: &str, run: &RunRow, events: &[EventRow]) -> Result<()> {
if !events.is_empty() {
let cols: Vec<Arc<dyn Array>> = vec![
strs(events, |_| repo.to_string()),
opt_strs(events, |_| commit.to_string()),
strs(events, |_| run.run_id.clone()),
longs(events, |e| e.thread as u64),
longs(events, |e| e.seq_no),
longs(events, |e| e.t_ns),
opt_strs(events, |e| e.stable_id.clone()),
strs(events, |e| e.kind.as_str().to_string()),
strs(events, |e| e.symbol.clone()),
];
append(wh, TABLE_FN_TRACE_EVENTS, events_schema()?, cols).await?;
}
let r = std::slice::from_ref(run);
let cols: Vec<Arc<dyn Array>> = vec![
strs(r, |r| r.run_id.clone()),
ts(r, run.ts_micros),
strs(r, |_| repo.to_string()),
opt_strs(r, |_| commit.to_string()),
strs(r, |r| r.target.clone()),
strs(r, |r| r.source.clone()),
longs(r, |r| r.chunk as u64),
longs(r, |r| r.chunks as u64),
longs(r, |r| r.events),
longs(r, |r| r.mapped_events),
longs(r, |r| r.functions),
longs(r, |r| r.mapped_functions),
longs(r, |r| r.local_functions),
longs(r, |r| r.mapped_local_functions),
longs(r, |r| r.threads),
longs(r, |r| r.first_ns),
longs(r, |r| r.last_ns),
longs(r, |r| r.total_events),
longs(r, |r| r.duration_ns),
Arc::new(BooleanArray::from(vec![run.truncated])),
strs(r, |r| r.binaries.clone()),
];
append(wh, TABLE_FN_TRACE_RUNS, runs_schema()?, cols).await
}
/// Is a profile of `run_id` filed?
pub async fn profile_filed(wh: &IcebergWarehouse, run_id: &str) -> Result<bool> {
let t = wh.catalog().load_table(&wh.table_ident(TABLE_FN_PROFILE)).await?;
let scan = t.scan().with_filter(Reference::new("run_id").equal_to(Datum::string(run_id))).select(["run_id"]).build()?;
let batches: Vec<RecordBatch> = scan.to_arrow().await?.try_collect().await?;
Ok(batches.iter().any(|b| str_col(b, "run_id").is_ok_and(|c| (0..b.num_rows()).any(|i| c.value(i) == run_id))))
}
/// **File a profile** as one snapshot.
#[allow(clippy::too_many_arguments)]
pub async fn append_profile(
wh: &IcebergWarehouse,
repo: &str,
commit: &str,
run_id: &str,
ts_micros: i64,
target: &str,
samples: u64,
profile_ns: u64,
rows: &[ProfileRow],
) -> Result<()> {
if rows.is_empty() {
return Ok(());
}
let cols: Vec<Arc<dyn Array>> = vec![
strs(rows, |_| run_id.to_string()),
ts(rows, ts_micros),
strs(rows, |_| repo.to_string()),
opt_strs(rows, |_| commit.to_string()),
strs(rows, |_| target.to_string()),
opt_strs(rows, |r| r.stable_id.clone()),
strs(rows, |r| r.symbol.clone()),
longs(rows, |r| r.self_ns),
longs(rows, |r| r.total_ns),
longs(rows, |r| r.self_samples),
longs(rows, |r| r.total_samples),
longs(rows, |_| samples),
longs(rows, |_| profile_ns),
];
append(wh, TABLE_FN_PROFILE, profile_schema()?, cols).await
}
// ── read ─────────────────────────────────────────────────────────────────────
fn str_col<'a>(b: &'a RecordBatch, name: &str) -> Result<&'a StringArray> {
b.column_by_name(name)
.and_then(|c| c.as_any().downcast_ref::<StringArray>())
.with_context(|| format!("column `{name}` is no string column"))
}
fn i64_col<'a>(b: &'a RecordBatch, name: &str) -> Result<&'a Int64Array> {
b.column_by_name(name)
.and_then(|c| c.as_any().downcast_ref::<Int64Array>())
.with_context(|| format!("column `{name}` is no long column"))
}
fn opt_str(c: &StringArray, i: usize) -> String {
if c.is_null(i) { String::new() } else { c.value(i).to_string() }
}
/// A run as the replay lists it: its chunks folded into one.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct TraceRunSummary {
pub run_id: String,
pub ts_micros: i64,
pub repo: String,
pub commit: String,
pub target: String,
pub source: String,
/// Chunks filed of how many.
pub chunks_filed: u32,
pub chunks: u32,
pub events: u64,
pub mapped_events: u64,
pub total_events: u64,
pub threads: u64,
pub duration_ns: u64,
pub truncated: bool,
/// Distinct functions summed over the chunks (a function in two chunks
/// counts twice; the share is what matters).
pub functions: u64,
pub mapped_functions: u64,
pub local_functions: u64,
pub mapped_local_functions: u64,
pub binaries: String,
}
/// **The traced runs of `repo`** (every repo when empty), newest first.
pub async fn query_runs(wh: &IcebergWarehouse, repo: &str) -> Result<Vec<TraceRunSummary>> {
let t = wh.catalog().load_table(&wh.table_ident(TABLE_FN_TRACE_RUNS)).await?;
let mut scan = t.scan();
if !repo.is_empty() {
scan = scan.with_filter(Reference::new("repo").equal_to(Datum::string(repo)));
}
let batches: Vec<RecordBatch> = scan.build()?.to_arrow().await?.try_collect().await?;
let mut by: HashMap<String, TraceRunSummary> = HashMap::new();
for b in &batches {
let tsc = b
.column_by_name("ts_micros")
.and_then(|c| c.as_any().downcast_ref::<TimestampMicrosecondArray>())
.context("ts_micros")?;
let trunc = b.column_by_name("truncated").and_then(|c| c.as_any().downcast_ref::<BooleanArray>()).context("truncated")?;
let (ids, repos, commits, targets, sources, bins) =
(str_col(b, "run_id")?, str_col(b, "repo")?, str_col(b, "commit")?, str_col(b, "target")?, str_col(b, "source")?, str_col(b, "binaries")?);
let l = |n: &str| i64_col(b, n);
let (chunks, events, mapped, total, threads, dur, fns, mfns, lfns, mlfns) = (
l("chunks")?,
l("events")?,
l("mapped_events")?,
l("total_events")?,
l("threads")?,
l("duration_ns")?,
l("functions")?,
l("mapped_functions")?,
l("local_functions")?,
l("mapped_local_functions")?,
);
for i in 0..b.num_rows() {
if !repo.is_empty() && repos.value(i) != repo {
continue;
}
let s = by.entry(ids.value(i).to_string()).or_insert_with(|| TraceRunSummary {
run_id: ids.value(i).to_string(),
ts_micros: tsc.value(i),
repo: repos.value(i).to_string(),
commit: opt_str(commits, i),
target: targets.value(i).to_string(),
source: sources.value(i).to_string(),
chunks: chunks.value(i) as u32,
total_events: total.value(i) as u64,
threads: threads.value(i) as u64,
duration_ns: dur.value(i) as u64,
truncated: trunc.value(i),
binaries: bins.value(i).to_string(),
..Default::default()
});
s.ts_micros = s.ts_micros.min(tsc.value(i));
s.chunks_filed += 1;
s.events += events.value(i) as u64;
s.mapped_events += mapped.value(i) as u64;
s.functions += fns.value(i) as u64;
s.mapped_functions += mfns.value(i) as u64;
s.local_functions += lfns.value(i) as u64;
s.mapped_local_functions += mlfns.value(i) as u64;
}
}
let mut out: Vec<TraceRunSummary> = by.into_values().collect();
out.sort_by(|a, b| b.ts_micros.cmp(&a.ts_micros).then(a.run_id.cmp(&b.run_id)));
Ok(out)
}
/// **A window of one run's events**: `from_ns <= t_ns <= to_ns`, after
/// `after_seq` (exclusive; `None` = from the window's start), in sequence
/// order, at most `limit`. The second value says whether more follow.
pub async fn query_window(
wh: &IcebergWarehouse,
run_id: &str,
from_ns: u64,
to_ns: u64,
after_seq: Option<u64>,
limit: usize,
) -> Result<(Vec<EventRow>, bool)> {
let t = wh.catalog().load_table(&wh.table_ident(TABLE_FN_TRACE_EVENTS)).await?;
let mut pred = Reference::new("run_id")
.equal_to(Datum::string(run_id))
.and(Reference::new("t_ns").greater_than_or_equal_to(Datum::long(from_ns as i64)))
.and(Reference::new("t_ns").less_than_or_equal_to(Datum::long(to_ns.min(i64::MAX as u64) as i64)));
if let Some(a) = after_seq {
pred = pred.and(Reference::new("seq_no").greater_than(Datum::long(a as i64)));
}
let scan = t
.scan()
.with_filter(pred)
.select(["run_id", "thread", "seq_no", "t_ns", "stable_id", "kind", "symbol"])
.build()?;
let batches: Vec<RecordBatch> = scan.to_arrow().await?.try_collect().await?;
let mut out = read_events(&batches, run_id, |e| e.t_ns >= from_ns && e.t_ns <= to_ns && after_seq.is_none_or(|a| e.seq_no > a))?;
out.sort_by_key(|e| e.seq_no);
let more = out.len() > limit;
out.truncate(limit);
Ok((out, more))
}
/// Every event of a run (for its heat), in sequence order.
pub async fn query_all_events(wh: &IcebergWarehouse, run_id: &str) -> Result<Vec<EventRow>> {
let t = wh.catalog().load_table(&wh.table_ident(TABLE_FN_TRACE_EVENTS)).await?;
let scan = t
.scan()
.with_filter(Reference::new("run_id").equal_to(Datum::string(run_id)))
.select(["run_id", "thread", "seq_no", "t_ns", "stable_id", "kind", "symbol"])
.build()?;
let batches: Vec<RecordBatch> = scan.to_arrow().await?.try_collect().await?;
let mut out = read_events(&batches, run_id, |_| true)?;
out.sort_by_key(|e| e.seq_no);
Ok(out)
}
fn read_events(batches: &[RecordBatch], run_id: &str, keep: impl Fn(&EventRow) -> bool) -> Result<Vec<EventRow>> {
let mut out = Vec::new();
for b in batches {
let (ids, sid, kind, sym) = (str_col(b, "run_id")?, str_col(b, "stable_id")?, str_col(b, "kind")?, str_col(b, "symbol")?);
let (thread, seq, t) = (i64_col(b, "thread")?, i64_col(b, "seq_no")?, i64_col(b, "t_ns")?);
for i in 0..b.num_rows() {
// the pushed-down filter prunes files; rows are checked here again
if ids.value(i) != run_id {
continue;
}
let e = EventRow {
thread: thread.value(i) as u32,
seq_no: seq.value(i) as u64,
t_ns: t.value(i) as u64,
stable_id: opt_str(sid, i),
kind: EventKind::parse(kind.value(i)).unwrap_or(EventKind::Exit),
symbol: sym.value(i).to_string(),
};
if keep(&e) {
out.push(e);
}
}
}
Ok(out)
}
/// **Total time per function of a run** (`stable_id` → ns; unmapped events
/// are keyed by their symbol behind a `?`): each entry paired with its exit on
/// its thread, a recursive function counted once — `nornir_trace`'s
/// inclusive time over the stored rows.
pub fn heat_of(events: &[EventRow]) -> HashMap<String, u64> {
let mut keys: Vec<String> = Vec::new();
let mut idx: HashMap<String, u32> = HashMap::new();
let mut run = nornir_trace::TraceRun::default();
for e in events {
let key = if e.stable_id.is_empty() { format!("?{}", e.symbol) } else { e.stable_id.clone() };
let f = *idx.entry(key.clone()).or_insert_with(|| {
keys.push(key);
(keys.len() - 1) as u32
});
run.events.push(nornir_trace::TraceEvent { thread: e.thread, seq_no: e.seq_no, t_ns: e.t_ns, func: f, kind: e.kind });
}
run.functions = keys.iter().map(|k| nornir_trace::TracedFn { symbol: k.clone(), ..Default::default() }).collect();
keys.into_iter().zip(run.inclusive_ns()).collect()
}
/// The profile rows of `repo`'s newest profile (or of `run_id` when named):
/// `(run_id, target, samples, profile_ns, rows)`.
pub async fn query_profile(wh: &IcebergWarehouse, repo: &str, run_id: &str) -> Result<Option<(String, String, u64, u64, Vec<ProfileRow>)>> {
let t = wh.catalog().load_table(&wh.table_ident(TABLE_FN_PROFILE)).await?;
let mut scan = t.scan();
if !run_id.is_empty() {
scan = scan.with_filter(Reference::new("run_id").equal_to(Datum::string(run_id)));
} else if !repo.is_empty() {
scan = scan.with_filter(Reference::new("repo").equal_to(Datum::string(repo)));
}
let batches: Vec<RecordBatch> = scan.build()?.to_arrow().await?.try_collect().await?;
// the newest run first
let mut newest: Option<(i64, String)> = None;
for b in &batches {
let tsc = b.column_by_name("ts_micros").and_then(|c| c.as_any().downcast_ref::<TimestampMicrosecondArray>()).context("ts_micros")?;
let (ids, repos) = (str_col(b, "run_id")?, str_col(b, "repo")?);
for i in 0..b.num_rows() {
let ok = if !run_id.is_empty() { ids.value(i) == run_id } else { repo.is_empty() || repos.value(i) == repo };
if ok && newest.as_ref().is_none_or(|(t, _)| tsc.value(i) > *t) {
newest = Some((tsc.value(i), ids.value(i).to_string()));
}
}
}
let Some((_, id)) = newest else { return Ok(None) };
let (mut target, mut samples, mut pns, mut rows) = (String::new(), 0, 0, Vec::new());
for b in &batches {
let (ids, targets, sid, sym) = (str_col(b, "run_id")?, str_col(b, "target")?, str_col(b, "stable_id")?, str_col(b, "symbol")?);
let l = |n: &str| i64_col(b, n);
let (sn, tn, ss, ts_, all, pn) = (l("self_ns")?, l("total_ns")?, l("self_samples")?, l("total_samples")?, l("samples")?, l("profile_ns")?);
for i in 0..b.num_rows() {
if ids.value(i) != id {
continue;
}
target = targets.value(i).to_string();
samples = all.value(i) as u64;
pns = pn.value(i) as u64;
rows.push(ProfileRow {
stable_id: opt_str(sid, i),
symbol: sym.value(i).to_string(),
self_ns: sn.value(i) as u64,
total_ns: tn.value(i) as u64,
self_samples: ss.value(i) as u64,
total_samples: ts_.value(i) as u64,
});
}
}
rows.sort_by(|a, b| b.total_ns.cmp(&a.total_ns).then(a.symbol.cmp(&b.symbol)));
Ok(Some((id, target, samples, pns, rows)))
}