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        // Checked when the project was loaded.
163        locale: b
164            .locale
165            .as_deref()
166            .and_then(|l| crate::numbers::Locale::parse(l).ok())
167            .unwrap_or_default(),
168    }
169}
170
171/// Resolved sources by `(source, table)`, failures included.
172type Resolved = BTreeMap<(String, String), Result<ResolvedSource, String>>;
173
174/// Resolves `source()` for one Binding: fields rendered with `limited`, once each.
175pub fn resolver(project: &Project, limited: Arc<Limited>) -> Option<SourceResolver> {
176    if project.sources.is_empty() {
177        return None;
178    }
179    let sources = project.sources.clone();
180    let cache: Arc<Mutex<Resolved>> = Arc::default();
181    Some(Arc::new(move |source: &str, table: &str| {
182        let key = (source.to_string(), table.to_string());
183        if let Some(r) = cache.lock().unwrap().get(&key) {
184            return r.clone();
185        }
186        let r = (|| {
187            let Some(s) = sources.get(source) else {
188                return Err(format!(
189                    "no source `{source}` is declared (declared: {})",
190                    sources.keys().cloned().collect::<Vec<_>>().join(", ")
191                ));
192            };
193            let Some(t) = s.table(table) else {
194                return Err(format!(
195                    "source `{source}` has no table `{table}` (it has: {})",
196                    s.tables
197                        .iter()
198                        .map(|t| t.name.as_str())
199                        .collect::<Vec<_>>()
200                        .join(", ")
201                ));
202            };
203            let at = |key: &str| format!("source `{source}`: `{key}`");
204            let render = |key: &str, v: &str| -> Result<String, String> {
205                let out = limited.render(&at(key), v)?;
206                if out.is_empty() {
207                    return Err(format!("{} `{v}` renders empty", at(key)));
208                }
209                Ok(out)
210            };
211            Ok(ResolvedSource {
212                source: source.to_string(),
213                table: table.to_string(),
214                profile: s.profile.as_deref().map(|p| render("profile", p)).transpose()?,
215                database: s.database.as_deref().map(|d| render("database", d)).transpose()?,
216                schema: render("schema", s.schema.as_deref().unwrap_or(&s.name))?,
217                identifier: render("identifier", t.identifier.as_deref().unwrap_or(&t.name))?,
218                quoting: t.quoting.over(&s.quoting),
219            })
220        })();
221        cache.lock().unwrap().insert(key, r.clone());
222        r
223    }))
224}
225
226/// Parse one Binding: render each query without a database, then resolve its connection.
227/// `vars` are the Binding's vars as the run sees them (with a schedule's layered in).
228pub fn binding(
229    project: &Project,
230    report: &Report,
231    b: &Binding,
232    vars: &JsonMap<String, Json>,
233    inputs: &Inputs,
234) -> ParsedBinding {
235    let ctx_name = match &b.set {
236        Some(s) => format!("report `{}`, Set `{s}`", report.name),
237        None => format!("report `{}`", report.name),
238    };
239    let context = run_context(
240        project,
241        report,
242        b,
243        inputs,
244        inputs.started_at.unwrap_or_else(Utc::now),
245    );
246    let limited = Arc::new(Limited::new(
247        context.clone(),
248        vars.clone(),
249        inputs.cli_vars.clone(),
250    ));
251    let mut out = ParsedBinding::default();
252    let error =
253        |out: &mut ParsedBinding, code: &'static str, file: PathBuf, line: Option<usize>, msg: String| {
254            out.errors.push(Problem {
255                code,
256                file,
257                line,
258                message: msg,
259            })
260        };
261    // `ctx` is `None` for a value written once for many Bindings (an inherited profile): the same
262    // message at its own file and line is then reported once.
263    let render_profile = |out: &mut ParsedBinding,
264                          what: &str,
265                          raw: &str,
266                          at: (&PathBuf, Option<usize>),
267                          ctx: Option<&str>|
268     -> Option<String> {
269        let message = match limited.render(what, raw) {
270            Ok(v) if !v.is_empty() => return Some(v),
271            Ok(_) => format!("{what} `{raw}` renders empty"),
272            Err(e) => e,
273        };
274        out.errors.push(Problem {
275            code: "invalid-profile-value",
276            file: at.0.clone(),
277            line: at.1,
278            message: match ctx {
279                Some(c) => format!("{c}: {message}"),
280                None => message,
281            },
282        });
283        None
284    };
285    let inherited_at = b.profile_at.clone().unwrap_or(crate::project::ProfileAt {
286        file: report.file.clone(),
287        line: None,
288        key: "`profile`".into(),
289    });
290    if let Some(raw) = &b.profile {
291        out.inherited = render_profile(
292            &mut out,
293            &inherited_at.key,
294            raw,
295            (&inherited_at.file, inherited_at.line),
296            None,
297        );
298    }
299    let inherited_failed = b.profile.is_some() && out.inherited.is_none();
300    let sources = resolver(project, limited.clone());
301    let renderer = Renderer::new(RendererConfig {
302        root: &project.root,
303        macros: &project.macros,
304        context,
305        vars: vars.clone(),
306        cli_vars: inputs.cli_vars.clone(),
307        runner: None,
308        connections: None,
309        mode: Mode::Parse,
310        connection: None,
311        source_type: String::new(),
312        sources: sources.clone(),
313        run_query_max_rows: project.run_query_max_rows,
314        sql: project.sql.clone(),
315        lookups: project.lookups.clone(),
316        lookup_inline_max_rows: project.lookup_inline_max_rows,
317        packages: project.packages.clone(),
318        project_name: project.name.clone(),
319        dispatch: project.dispatch.clone(),
320    });
321    let renderer = match renderer {
322        Ok(r) => Some(r),
323        Err(e) => {
324            error(
325                &mut out,
326                PARSE_FAILED,
327                e.file.clone(),
328                e.line,
329                format!("{ctx_name}: {}", e.message),
330            );
331            None
332        }
333    };
334    for q in &b.queries {
335        let mut pq = ParsedQuery {
336            query: q.query.clone(),
337            ..Default::default()
338        };
339        let mut failed = inherited_failed;
340        if let Some(raw) = &q.profile {
341            pq.profile = render_profile(
342                &mut out,
343                &format!("`profile` of query `{}`", q.query),
344                raw,
345                (&report.file, None),
346                Some(&ctx_name),
347            );
348            failed |= pq.profile.is_none();
349        }
350        let Some(r) = &renderer else {
351            out.queries.push(pq);
352            continue;
353        };
354        let src = match std::fs::read_to_string(project.root.join(&q.path)) {
355            Ok(s) => s,
356            Err(e) => {
357                error(
358                    &mut out,
359                    PARSE_FAILED,
360                    q.path.clone(),
361                    None,
362                    format!("can't read it: {e}"),
363                );
364                out.queries.push(pq);
365                continue;
366            }
367        };
368        let _ = r.take_sources();
369        if let Err(e) = r.render(&q.path, &src) {
370            error(
371                &mut out,
372                PARSE_FAILED,
373                e.file.clone(),
374                e.line,
375                format!("{ctx_name}: query `{}` doesn't parse: {}", q.query, e.message),
376            );
377            out.queries.push(pq);
378            continue;
379        }
380        let mut explicit: Vec<(String, String)> = Vec::new();
381        if let Some(p) = &pq.profile {
382            explicit.push((format!("query `{}` sets `profile: {p}`", q.query), p.clone()));
383        }
384        for (s, t) in r.take_sources() {
385            let resolve = sources
386                .as_ref()
387                .expect("source() succeeded, so sources are declared");
388            let rs = resolve(&s, &t).expect("resolved during the render");
389            let key = rs.key();
390            if let Some(p) = &rs.profile {
391                explicit.push((format!("source `{key}` is on connection `{p}`"), p.clone()));
392            }
393            out.sources.entry(key.clone()).or_insert(rs);
394            pq.sources.push(key);
395        }
396        pq.connection = match explicit.split_first() {
397            Some(((first_why, first), rest)) => {
398                if let Some((why, _)) = rest.iter().find(|(_, p)| p != first) {
399                    error(
400                        &mut out,
401                        "connection-conflict",
402                        q.path.clone(),
403                        None,
404                        format!(
405                            "{ctx_name}: query `{}` can't run on one connection: {first_why}, but {why}; one query runs on one connection",
406                            q.query
407                        ),
408                    );
409                    None
410                } else {
411                    Some(first.clone())
412                }
413            }
414            None => out.inherited.clone(),
415        };
416        // A profile that didn't render, or a conflict, is already the reason.
417        let conflicted = out.errors.iter().any(|e| e.file == q.path);
418        if pq.connection.is_none() && !failed && !conflicted {
419            error(
420                &mut out,
421                "no-connection",
422                report.file.clone(),
423                None,
424                format!(
425                    "{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)",
426                    q.query
427                ),
428            );
429        }
430        out.queries.push(pq);
431    }
432    for d in b.destinations() {
433        let p = render_profile(
434            &mut out,
435            "destination `profile`",
436            &d.profile,
437            (&report.file, None),
438            Some(&ctx_name),
439        );
440        out.destinations.push(p);
441    }
442    setup_warnings(report, b, &mut out);
443    out
444}
445
446/// A `tab: false` query prepares what later queries on its connection read (temp tables,
447/// `SET`s). One whose connection no later tab uses, while later tabs run elsewhere, can't reach
448/// them.
449fn setup_warnings(report: &Report, b: &Binding, out: &mut ParsedBinding) {
450    let conn = |name: &str| out.query(name).and_then(|q| q.connection.clone());
451    let mut warnings = Vec::new();
452    for (i, q) in b.queries.iter().enumerate() {
453        if q.tab {
454            continue;
455        }
456        let Some(c) = conn(&q.query) else { continue };
457        let later: Vec<(String, String)> = b.queries[i + 1..]
458            .iter()
459            .filter(|l| l.tab)
460            .filter_map(|l| Some((l.query.clone(), conn(&l.query)?)))
461            .collect();
462        if later.is_empty() || later.iter().any(|(_, lc)| *lc == c) {
463            continue;
464        }
465        let (other, oc) = &later[0];
466        warnings.push(Problem {
467            code: "setup-on-other-connection",
468            file: report.file.clone(),
469            line: None,
470            message: format!(
471                "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}`",
472                report.name, q.query
473            ),
474        });
475    }
476    out.warnings.extend(warnings);
477}
dre_core/parse.rs

dre_core/
parse.rs
