dre_core/
parse.rs

1//! The parse pass: each Binding rendered without a database, as dbt's parse finds `ref()` and
2//! `source()`, to learn every query's sources and the connection it runs on. Its results feed
3//! selection (`-s source:`), the manifest, `dre ls` and `dre validate`, none of which connect;
4//! a run repeats it with its own inputs and then holds rendering to what it found.
5//!
6//! A query's connection, by SPEC-011's rule:
7//! - explicit: the query's own `profile:` and the `profile:` of every source it uses. These
8//!   must agree (compared as rendered names), else it's an error naming both sides;
9//! - otherwise the inherited one: the Set's, the report's, the folder's `+profile`, then
10//!   `default_profile` (already merged into the Binding's `profile`).
11//!
12//! A source without a `profile:` never conflicts: it runs wherever its query runs.
13
14use std::collections::BTreeMap;
15use std::path::PathBuf;
16use std::sync::{Arc, Mutex};
17
18use chrono::{DateTime, NaiveDate, Utc};
19use serde_json::{Map as JsonMap, Value as Json};
20
21use crate::dates::Calendar;
22use crate::project::{Binding, Project, Report};
23use crate::render::{Limited, Mode, Renderer, RendererConfig, ResolvedSource, RunContext, SourceResolver};
24
25/// The code of a template the parse pass couldn't render.
26pub const PARSE_FAILED: &str = "parse-failed";
27
28/// What the parse pass renders `run.*`, `var()` and `target.name` with: the run's inputs.
29#[derive(Debug, Clone, Default)]
30pub struct Inputs {
31    pub target: String,
32    /// `--var`.
33    pub cli_vars: BTreeMap<String, String>,
34    /// `DRE_RUN_DATE`.
35    pub date: Option<NaiveDate>,
36    /// `DRE_RUN_AT`.
37    pub scheduled_at: Option<DateTime<Utc>>,
38    /// `--timezone`/`DRE_TIMEZONE`.
39    pub timezone: Option<String>,
40    /// `--schedule`.
41    pub schedule: Option<String>,
42    /// When the run started (`run.now` without `DRE_RUN_AT`); `None`: now.
43    pub started_at: Option<DateTime<Utc>>,
44}
45
46/// A problem found by the parse pass, at a file and line.
47#[derive(Debug, Clone, PartialEq, Eq)]
48pub struct Problem {
49    pub code: &'static str,
50    pub file: PathBuf,
51    pub line: Option<usize>,
52    pub message: String,
53}
54
55/// One query of a Binding after the parse pass.
56#[derive(Debug, Clone, Default, PartialEq, Eq)]
57pub struct ParsedQuery {
58    pub query: String,
59    /// Its own `profile:`, rendered.
60    pub profile: Option<String>,
61    /// The connection it runs on; `None` when none resolves (an error says why).
62    pub connection: Option<String>,
63    /// `source.table` of every `source()` it calls (directly, through `ref()` or a macro), in
64    /// order of first use.
65    pub sources: Vec<String>,
66}
67
68/// A Binding after the parse pass.
69#[derive(Debug, Clone, Default)]
70pub struct ParsedBinding {
71    /// The inherited connection, rendered.
72    pub inherited: Option<String>,
73    pub queries: Vec<ParsedQuery>,
74    /// Each `output.destination` entry's `profile`, rendered (`None` where that failed).
75    pub destinations: Vec<Option<String>>,
76    /// Every source the queries use, resolved, by `source.table`.
77    pub sources: BTreeMap<String, ResolvedSource>,
78    pub errors: Vec<Problem>,
79    pub warnings: Vec<Problem>,
80}
81
82impl ParsedBinding {
83    pub fn query(&self, name: &str) -> Option<&ParsedQuery> {
84        self.queries.iter().find(|q| q.query == name)
85    }
86
87    /// Every connection the queries use, in order of first use.
88    pub fn connections(&self) -> Vec<&str> {
89        let mut out: Vec<&str> = Vec::new();
90        for c in self.queries.iter().filter_map(|q| q.connection.as_deref()) {
91            if !out.contains(&c) {
92                out.push(c);
93            }
94        }
95        out
96    }
97
98    /// `source.table` of every source used, in order of first use.
99    pub fn source_keys(&self) -> Vec<&str> {
100        let mut out: Vec<&str> = Vec::new();
101        for s in self.queries.iter().flat_map(|q| q.sources.iter()) {
102            if !out.contains(&s.as_str()) {
103                out.push(s);
104            }
105        }
106        out
107    }
108}
109
110/// The run's calendar for a report: `--timezone`, else the schedule's, else the report's, else
111/// UTC.
112pub fn calendar(
113    project: &Project,
114    report: &Report,
115    timezone: Option<&str>,
116    schedule: Option<&str>,
117) -> Calendar {
118    let schedule_tz = schedule
119        .and_then(|n| project.schedules.iter().find(|e| e.name == n))
120        .and_then(|e| e.timezone.as_deref());
121    let tz = timezone
122        .or(schedule_tz)
123        .or(report.timezone.as_deref())
124        .and_then(|t| crate::dates::parse_tz(t).ok())
125        .unwrap_or(chrono_tz::Tz::UTC);
126    Calendar {
127        tz,
128        week_start: project.week_start,
129        numbering: project.week_numbering,
130    }
131}
132
133/// `run.date`: `DRE_RUN_DATE`, else `DRE_RUN_AT`'s date in the run's timezone, else today there.
134pub fn run_date(cal: &Calendar, date: Option<NaiveDate>, scheduled_at: Option<DateTime<Utc>>) -> NaiveDate {
135    date.or(scheduled_at.map(|t| t.with_timezone(&cal.tz).date_naive()))
136        .unwrap_or_else(|| cal.today())
137}
138
139/// The `run.*` context a Binding renders with.
140pub fn run_context(
141    project: &Project,
142    report: &Report,
143    b: &Binding,
144    inputs: &Inputs,
145    now: DateTime<Utc>,
146) -> RunContext {
147    let calendar = calendar(
148        project,
149        report,
150        inputs.timezone.as_deref(),
151        inputs.schedule.as_deref(),
152    );
153    RunContext {
154        report: report.name.clone(),
155        set: b.set.clone(),
156        target: inputs.target.clone(),
157        schedule: inputs.schedule.clone(),
158        date: run_date(&calendar, inputs.date, inputs.scheduled_at),
159        now: inputs.scheduled_at.unwrap_or(now),
160        scheduled_at: inputs.scheduled_at,
161        calendar,
162    }
163}
164
165/// Resolved sources by `(source, table)`, failures included.
166type Resolved = BTreeMap<(String, String), Result<ResolvedSource, String>>;
167
168/// Resolves `source()` for one Binding: fields rendered with `limited`, once each.
169pub fn resolver(project: &Project, limited: Arc<Limited>) -> Option<SourceResolver> {
170    if project.sources.is_empty() {
171        return None;
172    }
173    let sources = project.sources.clone();
174    let cache: Arc<Mutex<Resolved>> = Arc::default();
175    Some(Arc::new(move |source: &str, table: &str| {
176        let key = (source.to_string(), table.to_string());
177        if let Some(r) = cache.lock().unwrap().get(&key) {
178            return r.clone();
179        }
180        let r = (|| {
181            let Some(s) = sources.get(source) else {
182                return Err(format!(
183                    "no source `{source}` is declared (declared: {})",
184                    sources.keys().cloned().collect::<Vec<_>>().join(", ")
185                ));
186            };
187            let Some(t) = s.table(table) else {
188                return Err(format!(
189                    "source `{source}` has no table `{table}` (it has: {})",
190                    s.tables
191                        .iter()
192                        .map(|t| t.name.as_str())
193                        .collect::<Vec<_>>()
194                        .join(", ")
195                ));
196            };
197            let at = |key: &str| format!("source `{source}`: `{key}`");
198            let render = |key: &str, v: &str| -> Result<String, String> {
199                let out = limited.render(&at(key), v)?;
200                if out.is_empty() {
201                    return Err(format!("{} `{v}` renders empty", at(key)));
202                }
203                Ok(out)
204            };
205            Ok(ResolvedSource {
206                source: source.to_string(),
207                table: table.to_string(),
208                profile: s.profile.as_deref().map(|p| render("profile", p)).transpose()?,
209                database: s.database.as_deref().map(|d| render("database", d)).transpose()?,
210                schema: render("schema", s.schema.as_deref().unwrap_or(&s.name))?,
211                identifier: render("identifier", t.identifier.as_deref().unwrap_or(&t.name))?,
212                quoting: t.quoting.over(&s.quoting),
213            })
214        })();
215        cache.lock().unwrap().insert(key, r.clone());
216        r
217    }))
218}
219
220/// Parse one Binding: render each query without a database, then resolve its connection.
221/// `vars` are the Binding's vars as the run sees them (with a schedule's layered in).
222pub fn binding(
223    project: &Project,
224    report: &Report,
225    b: &Binding,
226    vars: &JsonMap<String, Json>,
227    inputs: &Inputs,
228) -> ParsedBinding {
229    let ctx_name = match &b.set {
230        Some(s) => format!("report `{}`, Set `{s}`", report.name),
231        None => format!("report `{}`", report.name),
232    };
233    let context = run_context(
234        project,
235        report,
236        b,
237        inputs,
238        inputs.started_at.unwrap_or_else(Utc::now),
239    );
240    let limited = Arc::new(Limited::new(
241        context.clone(),
242        vars.clone(),
243        inputs.cli_vars.clone(),
244    ));
245    let mut out = ParsedBinding::default();
246    let error =
247        |out: &mut ParsedBinding, code: &'static str, file: PathBuf, line: Option<usize>, msg: String| {
248            out.errors.push(Problem {
249                code,
250                file,
251                line,
252                message: msg,
253            })
254        };
255    // `ctx` is `None` for a value written once for many Bindings (an inherited profile): the same
256    // message at its own file and line is then reported once.
257    let render_profile = |out: &mut ParsedBinding,
258                          what: &str,
259                          raw: &str,
260                          at: (&PathBuf, Option<usize>),
261                          ctx: Option<&str>|
262     -> Option<String> {
263        let message = match limited.render(what, raw) {
264            Ok(v) if !v.is_empty() => return Some(v),
265            Ok(_) => format!("{what} `{raw}` renders empty"),
266            Err(e) => e,
267        };
268        out.errors.push(Problem {
269            code: "invalid-profile-value",
270            file: at.0.clone(),
271            line: at.1,
272            message: match ctx {
273                Some(c) => format!("{c}: {message}"),
274                None => message,
275            },
276        });
277        None
278    };
279    let inherited_at = b.profile_at.clone().unwrap_or(crate::project::ProfileAt {
280        file: report.file.clone(),
281        line: None,
282        key: "`profile`".into(),
283    });
284    if let Some(raw) = &b.profile {
285        out.inherited = render_profile(
286            &mut out,
287            &inherited_at.key,
288            raw,
289            (&inherited_at.file, inherited_at.line),
290            None,
291        );
292    }
293    let inherited_failed = b.profile.is_some() && out.inherited.is_none();
294    let sources = resolver(project, limited.clone());
295    let renderer = Renderer::new(RendererConfig {
296        root: &project.root,
297        macros: &project.macros,
298        context,
299        vars: vars.clone(),
300        cli_vars: inputs.cli_vars.clone(),
301        runner: None,
302        connections: None,
303        mode: Mode::Parse,
304        connection: None,
305        source_type: String::new(),
306        sources: sources.clone(),
307        run_query_max_rows: project.run_query_max_rows,
308        sql: project.sql.clone(),
309        lookups: project.lookups.clone(),
310        lookup_inline_max_rows: project.lookup_inline_max_rows,
311        packages: project.packages.clone(),
312        project_name: project.name.clone(),
313        dispatch: project.dispatch.clone(),
314    });
315    let renderer = match renderer {
316        Ok(r) => Some(r),
317        Err(e) => {
318            error(
319                &mut out,
320                PARSE_FAILED,
321                e.file.clone(),
322                e.line,
323                format!("{ctx_name}: {}", e.message),
324            );
325            None
326        }
327    };
328    for q in &b.queries {
329        let mut pq = ParsedQuery {
330            query: q.query.clone(),
331            ..Default::default()
332        };
333        let mut failed = inherited_failed;
334        if let Some(raw) = &q.profile {
335            pq.profile = render_profile(
336                &mut out,
337                &format!("`profile` of query `{}`", q.query),
338                raw,
339                (&report.file, None),
340                Some(&ctx_name),
341            );
342            failed |= pq.profile.is_none();
343        }
344        let Some(r) = &renderer else {
345            out.queries.push(pq);
346            continue;
347        };
348        let src = match std::fs::read_to_string(project.root.join(&q.path)) {
349            Ok(s) => s,
350            Err(e) => {
351                error(
352                    &mut out,
353                    PARSE_FAILED,
354                    q.path.clone(),
355                    None,
356                    format!("can't read it: {e}"),
357                );
358                out.queries.push(pq);
359                continue;
360            }
361        };
362        let _ = r.take_sources();
363        if let Err(e) = r.render(&q.path, &src) {
364            error(
365                &mut out,
366                PARSE_FAILED,
367                e.file.clone(),
368                e.line,
369                format!("{ctx_name}: query `{}` doesn't parse: {}", q.query, e.message),
370            );
371            out.queries.push(pq);
372            continue;
373        }
374        let mut explicit: Vec<(String, String)> = Vec::new();
375        if let Some(p) = &pq.profile {
376            explicit.push((format!("query `{}` sets `profile: {p}`", q.query), p.clone()));
377        }
378        for (s, t) in r.take_sources() {
379            let resolve = sources
380                .as_ref()
381                .expect("source() succeeded, so sources are declared");
382            let rs = resolve(&s, &t).expect("resolved during the render");
383            let key = rs.key();
384            if let Some(p) = &rs.profile {
385                explicit.push((format!("source `{key}` is on connection `{p}`"), p.clone()));
386            }
387            out.sources.entry(key.clone()).or_insert(rs);
388            pq.sources.push(key);
389        }
390        pq.connection = match explicit.split_first() {
391            Some(((first_why, first), rest)) => {
392                if let Some((why, _)) = rest.iter().find(|(_, p)| p != first) {
393                    error(
394                        &mut out,
395                        "connection-conflict",
396                        q.path.clone(),
397                        None,
398                        format!(
399                            "{ctx_name}: query `{}` can't run on one connection: {first_why}, but {why}; one query runs on one connection",
400                            q.query
401                        ),
402                    );
403                    None
404                } else {
405                    Some(first.clone())
406                }
407            }
408            None => out.inherited.clone(),
409        };
410        // A profile that didn't render, or a conflict, is already the reason.
411        let conflicted = out.errors.iter().any(|e| e.file == q.path);
412        if pq.connection.is_none() && !failed && !conflicted {
413            error(
414                &mut out,
415                "no-connection",
416                report.file.clone(),
417                None,
418                format!(
419                    "{ctx_name}: query `{}` has no connection: give it `profile:`, use a source with a `profile:`, or give the report one (`profile:` on the report or Set, a folder's `+profile`, or `default_profile` in dre_project.yml)",
420                    q.query
421                ),
422            );
423        }
424        out.queries.push(pq);
425    }
426    for d in &b.output.destinations {
427        let p = render_profile(
428            &mut out,
429            "destination `profile`",
430            &d.profile,
431            (&report.file, None),
432            Some(&ctx_name),
433        );
434        out.destinations.push(p);
435    }
436    setup_warnings(report, b, &mut out);
437    out
438}
439
440/// A `tab: false` query prepares what later queries on its connection read (temp tables,
441/// `SET`s). One whose connection no later tab uses, while later tabs run elsewhere, can't reach
442/// them.
443fn setup_warnings(report: &Report, b: &Binding, out: &mut ParsedBinding) {
444    let conn = |name: &str| out.query(name).and_then(|q| q.connection.clone());
445    let mut warnings = Vec::new();
446    for (i, q) in b.queries.iter().enumerate() {
447        if q.tab {
448            continue;
449        }
450        let Some(c) = conn(&q.query) else { continue };
451        let later: Vec<(String, String)> = b.queries[i + 1..]
452            .iter()
453            .filter(|l| l.tab)
454            .filter_map(|l| Some((l.query.clone(), conn(&l.query)?)))
455            .collect();
456        if later.is_empty() || later.iter().any(|(_, lc)| *lc == c) {
457            continue;
458        }
459        let (other, oc) = &later[0];
460        warnings.push(Problem {
461            code: "setup-on-other-connection",
462            file: report.file.clone(),
463            line: None,
464            message: format!(
465                "report `{}`: setup query `{}` (`tab: false`) runs on connection `{c}`, but the tabs after it run elsewhere (`{other}` on `{oc}`); its temp tables and `SET`s are only visible on `{c}`",
466                report.name, q.query
467            ),
468        });
469    }
470    out.warnings.extend(warnings);
471}
dre_core/parse.rs

dre_core/
parse.rs
