lean_ctx_patterns/
dbt.rs

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

lean_ctx_patterns/
dbt.rs
