//! dbt (data build tool) `run`/`test`/`build` output compression.
//!
//! dbt prefixes every line with an `HH:MM:SS` timestamp and emits a START/RUN
//! line plus an OK/ERROR/SKIP line per node. We keep the final
//! `Done. PASS=.. WARN=.. ERROR=.. SKIP=.. TOTAL=..` summary, the run duration
//! and any failing node + its error detail, dropping timestamps and the
//! per-node success noise.
use crate::core::compressor::strip_ansi;
pub fn compress(_cmd: &str, output: &str) -> Option<String> {
let trimmed = output.trim();
if trimmed.is_empty() {
return Some("dbt: ok".to_string());
}
let mut summary: Option<String> = None;
let mut found: Option<String> = None;
let mut duration: Option<String> = None;
let mut errors: Vec<String> = Vec::new();
let mut in_detail = false;
for raw in trimmed.lines() {
let stripped = strip_ansi(raw);
let line = strip_timestamp(&stripped);
let line = line.trim();
if line.is_empty() {
continue;
}
if let Some(rest) = line.strip_prefix("Done.") {
summary = Some(rest.trim().to_string());
continue;
}
if found.is_none() && line.starts_with("Found ") {
found = Some(line.to_string());
continue;
}
if line.starts_with("Finished running")
&& let Some((_, dur)) = line.rsplit_once(" in ")
{
duration = Some(dur.trim_end_matches('.').trim().to_string());
continue;
}
if line.starts_with("Completed with") {
in_detail = true;
continue;
}
if is_failure_node(line) {
errors.push(node_label(line));
continue;
}
if in_detail && is_error_detail(line) {
errors.push(line.to_string());
}
}
let mut head = match (&summary, &found) {
(Some(s), _) => format!("dbt: {s}"),
(None, Some(f)) => format!("dbt: {f}"),
(None, None) => return Some(fallback(trimmed)),
};
if let Some(d) = duration {
head.push_str(&format!(" ({d})"));
}
let mut parts = vec![head];
let mut seen = std::collections::HashSet::new();
for e in errors {
if seen.insert(e.clone()) {
parts.push(format!(" {e}"));
}
}
Some(parts.join("\n"))
}
/// Strip dbt's leading `HH:MM:SS` timestamp (followed by spaces) from a line.
fn strip_timestamp(line: &str) -> String {
let b = line.as_bytes();
if b.len() >= 8
&& b[0].is_ascii_digit()
&& b[1].is_ascii_digit()
&& b[2] == b':'
&& b[3].is_ascii_digit()
&& b[4].is_ascii_digit()
&& b[5] == b':'
&& b[6].is_ascii_digit()
&& b[7].is_ascii_digit()
{
line[8..].trim_start().to_string()
} else {
line.to_string()
}
}
/// A per-node failure line, e.g. `2 of 12 ERROR creating sql view model x ...`.
fn is_failure_node(line: &str) -> bool {
line.contains(" of ") && (line.contains(" ERROR ") || line.contains(" FAIL "))
}
/// Compact a per-node failure to `ERROR <node>` without the `N of M` index,
/// the trailing dotted padding or the `[ERROR in ..s]` status suffix.
fn node_label(line: &str) -> String {
let start = line
.find(" ERROR ")
.or_else(|| line.find(" FAIL "))
.map_or(0, |i| i + 1);
let rest = &line[start..];
let rest = match rest.find("..") {
Some(i) => &rest[..i],
None => rest,
};
let rest = rest.split(" [").next().unwrap_or(rest);
rest.trim().to_string()
}
fn is_error_detail(line: &str) -> bool {
line.contains("Error in ")
|| line.starts_with("Database Error")
|| line.starts_with("Compilation Error")
|| line.starts_with("Runtime Error")
|| line.starts_with("Failure in ")
}
fn fallback(text: &str) -> String {
let lines: Vec<&str> = text.lines().filter(|l| !l.trim().is_empty()).collect();
let n = lines.len().min(8);
let mut s = lines[..n].join("\n");
if lines.len() > n {
s.push_str(&format!("\n... (+{} lines)", lines.len() - n));
}
s
}
#[cfg(test)]
mod tests {
use super::*;
const RUN_WITH_ERROR: &str = "20:14:01 Running with dbt=1.7.3\n20:14:02 Found 12 models, 4 tests, 2 sources\n20:14:03 Concurrency: 4 threads (target='dev')\n20:14:03 1 of 12 START sql table model public.stg_users ........ [RUN]\n20:14:04 1 of 12 OK created sql table model public.stg_users ... [SELECT 100 in 0.50s]\n20:14:05 2 of 12 ERROR creating sql view model public.dim_users . [ERROR in 0.30s]\n20:14:20 Finished running 11 table models, 1 view model in 18.50s.\n20:14:20 Completed with 1 error and 0 warnings:\n20:14:20 Database Error in model dim_users (models/dim_users.sql)\n20:14:20 column \"foo\" does not exist\n20:14:20 Done. PASS=11 WARN=0 ERROR=1 SKIP=0 TOTAL=12\n";
#[test]
fn keeps_summary_duration_and_errors() {
let r = compress("dbt run", RUN_WITH_ERROR).unwrap();
assert!(r.contains("PASS=11"), "keeps pass count: {r}");
assert!(r.contains("ERROR=1"), "keeps error count: {r}");
assert!(r.contains("18.50s"), "keeps duration: {r}");
assert!(r.contains("dim_users"), "keeps failing node: {r}");
assert!(!r.contains("OK created"), "drops success noise: {r}");
assert!(!r.contains("20:14"), "drops timestamps: {r}");
}
#[test]
fn shorter_than_input() {
let r = compress("dbt run", RUN_WITH_ERROR).unwrap();
assert!(r.len() < RUN_WITH_ERROR.len());
}
#[test]
fn empty_is_ok() {
assert_eq!(compress("dbt run", " ").unwrap(), "dbt: ok");
}
#[test]
fn falls_back_without_summary() {
let r = compress("dbt debug", "Connection test: OK\nAll checks passed").unwrap();
assert!(r.contains("Connection test"));
}
}