lean_ctx/core/patterns/
dbt.rs

1//! dbt (data build tool) `run`/`test`/`build` output compression.
2//!
3//! dbt prefixes every line with an `HH:MM:SS` timestamp and emits a START/RUN
4//! line plus an OK/ERROR/SKIP line per node. We keep the final
5//! `Done. PASS=.. WARN=.. ERROR=.. SKIP=.. TOTAL=..` summary, the run duration
6//! and any failing node + its error detail, dropping timestamps and the
7//! per-node success noise.
8
9use crate::core::compressor::strip_ansi;
10
11pub fn compress(_cmd: &str, output: &str) -> Option<String> {
12    let trimmed = output.trim();
13    if trimmed.is_empty() {
14        return Some("dbt: ok".to_string());
15    }
16
17    let mut summary: Option<String> = None;
18    let mut found: Option<String> = None;
19    let mut duration: Option<String> = None;
20    let mut errors: Vec<String> = Vec::new();
21    let mut in_detail = false;
22
23    for raw in trimmed.lines() {
24        let stripped = strip_ansi(raw);
25        let line = strip_timestamp(&stripped);
26        let line = line.trim();
27        if line.is_empty() {
28            continue;
29        }
30
31        if let Some(rest) = line.strip_prefix("Done.") {
32            summary = Some(rest.trim().to_string());
33            continue;
34        }
35        if found.is_none() && line.starts_with("Found ") {
36            found = Some(line.to_string());
37            continue;
38        }
39        if line.starts_with("Finished running")
40            && let Some((_, dur)) = line.rsplit_once(" in ")
41        {
42            duration = Some(dur.trim_end_matches('.').trim().to_string());
43            continue;
44        }
45        if line.starts_with("Completed with") {
46            in_detail = true;
47            continue;
48        }
49        if is_failure_node(line) {
50            errors.push(node_label(line));
51            continue;
52        }
53        if in_detail && is_error_detail(line) {
54            errors.push(line.to_string());
55        }
56    }
57
58    let mut head = match (&summary, &found) {
59        (Some(s), _) => format!("dbt: {s}"),
60        (None, Some(f)) => format!("dbt: {f}"),
61        (None, None) => return Some(fallback(trimmed)),
62    };
63    if let Some(d) = duration {
64        head.push_str(&format!(" ({d})"));
65    }
66
67    let mut parts = vec![head];
68    let mut seen = std::collections::HashSet::new();
69    for e in errors {
70        if seen.insert(e.clone()) {
71            parts.push(format!("  {e}"));
72        }
73    }
74    Some(parts.join("\n"))
75}
76
77/// Strip dbt's leading `HH:MM:SS` timestamp (followed by spaces) from a line.
78fn strip_timestamp(line: &str) -> String {
79    let b = line.as_bytes();
80    if b.len() >= 8
81        && b[0].is_ascii_digit()
82        && b[1].is_ascii_digit()
83        && b[2] == b':'
84        && b[3].is_ascii_digit()
85        && b[4].is_ascii_digit()
86        && b[5] == b':'
87        && b[6].is_ascii_digit()
88        && b[7].is_ascii_digit()
89    {
90        line[8..].trim_start().to_string()
91    } else {
92        line.to_string()
93    }
94}
95
96/// A per-node failure line, e.g. `2 of 12 ERROR creating sql view model x ...`.
97fn is_failure_node(line: &str) -> bool {
98    line.contains(" of ") && (line.contains(" ERROR ") || line.contains(" FAIL "))
99}
100
101/// Compact a per-node failure to `ERROR <node>` without the `N of M` index,
102/// the trailing dotted padding or the `[ERROR in ..s]` status suffix.
103fn node_label(line: &str) -> String {
104    let start = line
105        .find(" ERROR ")
106        .or_else(|| line.find(" FAIL "))
107        .map_or(0, |i| i + 1);
108    let rest = &line[start..];
109    let rest = match rest.find("..") {
110        Some(i) => &rest[..i],
111        None => rest,
112    };
113    let rest = rest.split(" [").next().unwrap_or(rest);
114    rest.trim().to_string()
115}
116
117fn is_error_detail(line: &str) -> bool {
118    line.contains("Error in ")
119        || line.starts_with("Database Error")
120        || line.starts_with("Compilation Error")
121        || line.starts_with("Runtime Error")
122        || line.starts_with("Failure in ")
123}
124
125fn fallback(text: &str) -> String {
126    let lines: Vec<&str> = text.lines().filter(|l| !l.trim().is_empty()).collect();
127    let n = lines.len().min(8);
128    let mut s = lines[..n].join("\n");
129    if lines.len() > n {
130        s.push_str(&format!("\n... (+{} lines)", lines.len() - n));
131    }
132    s
133}
134
135#[cfg(test)]
136mod tests {
137    use super::*;
138
139    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";
140
141    #[test]
142    fn keeps_summary_duration_and_errors() {
143        let r = compress("dbt run", RUN_WITH_ERROR).unwrap();
144        assert!(r.contains("PASS=11"), "keeps pass count: {r}");
145        assert!(r.contains("ERROR=1"), "keeps error count: {r}");
146        assert!(r.contains("18.50s"), "keeps duration: {r}");
147        assert!(r.contains("dim_users"), "keeps failing node: {r}");
148        assert!(!r.contains("OK created"), "drops success noise: {r}");
149        assert!(!r.contains("20:14"), "drops timestamps: {r}");
150    }
151
152    #[test]
153    fn shorter_than_input() {
154        let r = compress("dbt run", RUN_WITH_ERROR).unwrap();
155        assert!(r.len() < RUN_WITH_ERROR.len());
156    }
157
158    #[test]
159    fn empty_is_ok() {
160        assert_eq!(compress("dbt run", "   ").unwrap(), "dbt: ok");
161    }
162
163    #[test]
164    fn falls_back_without_summary() {
165        let r = compress("dbt debug", "Connection test: OK\nAll checks passed").unwrap();
166        assert!(r.contains("Connection test"));
167    }
168}
lean_ctx/core/patterns/dbt.rs

lean_ctx/core/patterns/
dbt.rs
