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::codes::Code;
22use crate::dates::Calendar;
23use crate::project::{Binding, Project, Report};
24use crate::render::{Limited, Mode, Renderer, RendererConfig, ResolvedSource, RunContext, SourceResolver};
25
26/// The code of a template the parse pass couldn't render.
27pub const PARSE_FAILED: Code = Code::ParseFailed;
28
29/// What the parse pass renders `run.*`, `var()` and `target.name` with: the run's inputs.
30#[derive(Debug, Clone, Default)]
31pub struct Inputs {
32    pub target: String,
33    /// `--var`.
34    pub cli_vars: BTreeMap<String, String>,
35    /// `DRE_RUN_DATE`.
36    pub date: Option<NaiveDate>,
37    /// `DRE_RUN_AT`.
38    pub scheduled_at: Option<DateTime<Utc>>,
39    /// `--timezone`/`DRE_TIMEZONE`.
40    pub timezone: Option<String>,
41    /// `--schedule`.
42    pub schedule: Option<String>,
43    /// When the run started (`run.now` without `DRE_RUN_AT`); `None`: now.
44    pub started_at: Option<DateTime<Utc>>,
45}
46
47/// A problem found by the parse pass, at a file and line.
48#[derive(Debug, Clone, PartialEq, Eq)]
49pub struct Problem {
50    pub code: Code,
51    pub file: PathBuf,
52    pub line: Option<usize>,
53    pub message: String,
54}
55
56/// One query of a Binding after the parse pass.
57#[derive(Debug, Clone, Default, PartialEq, Eq)]
58pub struct ParsedQuery {
59    pub query: String,
60    /// Its own `profile:`, rendered.
61    pub profile: Option<String>,
62    /// The connection it runs on; `None` when none resolves (an error says why).
63    pub connection: Option<String>,
64    /// `source.table` of every `source()` it calls (directly, through `ref()` or a macro), in
65    /// order of first use.
66    pub sources: Vec<String>,
67}
68
69/// A Binding after the parse pass.
70#[derive(Debug, Clone, Default)]
71pub struct ParsedBinding {
72    /// The inherited connection, rendered.
73    pub inherited: Option<String>,
74    pub queries: Vec<ParsedQuery>,
75    /// Each `output.destination` entry's `profile`, rendered (`None` where that failed).
76    pub destinations: Vec<Option<String>>,
77    /// Every source the queries use, resolved, by `source.table`.
78    pub sources: BTreeMap<String, ResolvedSource>,
79    pub errors: Vec<Problem>,
80    pub warnings: Vec<Problem>,
81}
82
83impl ParsedBinding {
84    pub fn query(&self, name: &str) -> Option<&ParsedQuery> {
85        self.queries.iter().find(|q| q.query == name)
86    }
87
88    /// Every connection the queries use, in order of first use.
89    pub fn connections(&self) -> Vec<&str> {
90        let mut out: Vec<&str> = Vec::new();
91        for c in self.queries.iter().filter_map(|q| q.connection.as_deref()) {
92            if !out.contains(&c) {
93                out.push(c);
94            }
95        }
96        out
97    }
98
99    /// `source.table` of every source used, in order of first use.
100    pub fn source_keys(&self) -> Vec<&str> {
101        let mut out: Vec<&str> = Vec::new();
102        for s in self.queries.iter().flat_map(|q| q.sources.iter()) {
103            if !out.contains(&s.as_str()) {
104                out.push(s);
105            }
106        }
107        out
108    }
109}
110
111/// The run's calendar for a report: `--timezone`, else the schedule's, else the report's, else
112/// UTC.
113pub fn calendar(
114    project: &Project,
115    report: &Report,
116    timezone: Option<&str>,
117    schedule: Option<&str>,
118) -> Calendar {
119    let schedule_tz = schedule
120        .and_then(|n| project.schedules.iter().find(|e| e.name == n))
121        .and_then(|e| e.timezone.as_deref());
122    let tz = timezone
123        .or(schedule_tz)
124        .or(report.timezone.as_deref())
125        .and_then(|t| crate::dates::parse_tz(t).ok())
126        .unwrap_or(chrono_tz::Tz::UTC);
127    Calendar {
128        tz,
129        week_start: project.week_start,
130        numbering: project.week_numbering,
131    }
132}
133
134/// `run.date`: `DRE_RUN_DATE`, else `DRE_RUN_AT`'s date in the run's timezone, else today there.
135pub fn run_date(cal: &Calendar, date: Option<NaiveDate>, scheduled_at: Option<DateTime<Utc>>) -> NaiveDate {
136    date.or(scheduled_at.map(|t| t.with_timezone(&cal.tz).date_naive()))
137        .unwrap_or_else(|| cal.today())
138}
139
140/// The `run.*` context a Binding renders with.
141pub fn run_context(
142    project: &Project,
143    report: &Report,
144    b: &Binding,
145    inputs: &Inputs,
146    now: DateTime<Utc>,
147) -> RunContext {
148    let calendar = calendar(
149        project,
150        report,
151        inputs.timezone.as_deref(),
152        inputs.schedule.as_deref(),
153    );
154    RunContext {
155        report: report.name.clone(),
156        set: b.set.clone(),
157        target: inputs.target.clone(),
158        schedule: inputs.schedule.clone(),
159        date: run_date(&calendar, inputs.date, inputs.scheduled_at),
160        now: inputs.scheduled_at.unwrap_or(now),
161        scheduled_at: inputs.scheduled_at,
162        calendar,
163        // Checked when the project was loaded.
164        locale: b
165            .locale
166            .as_deref()
167            .and_then(|l| crate::numbers::Locale::parse(l).ok())
168            .unwrap_or_default(),
169    }
170}
171
172/// Resolved sources by `(source, table)`, failures included.
173type Resolved = BTreeMap<(String, String), Result<ResolvedSource, String>>;
174
175/// Resolves `source()` for one Binding: fields rendered with `limited`, once each.
176pub fn resolver(project: &Project, limited: Arc<Limited>) -> Option<SourceResolver> {
177    if project.sources.is_empty() {
178        return None;
179    }
180    let sources = project.sources.clone();
181    let cache: Arc<Mutex<Resolved>> = Arc::default();
182    Some(Arc::new(move |source: &str, table: &str| {
183        let key = (source.to_string(), table.to_string());
184        if let Some(r) = cache.lock().unwrap().get(&key) {
185            return r.clone();
186        }
187        let r = (|| {
188            let Some(s) = sources.get(source) else {
189                return Err(format!(
190                    "no source `{source}` is declared (declared: {})",
191                    sources.keys().cloned().collect::<Vec<_>>().join(", ")
192                ));
193            };
194            let Some(t) = s.table(table) else {
195                return Err(format!(
196                    "source `{source}` has no table `{table}` (it has: {})",
197                    s.tables
198                        .iter()
199                        .map(|t| t.name.as_str())
200                        .collect::<Vec<_>>()
201                        .join(", ")
202                ));
203            };
204            let at = |key: &str| format!("source `{source}`: `{key}`");
205            let render = |key: &str, v: &str| -> Result<String, String> {
206                let out = limited.render(&at(key), v)?;
207                if out.is_empty() {
208                    return Err(format!("{} `{v}` renders empty", at(key)));
209                }
210                Ok(out)
211            };
212            Ok(ResolvedSource {
213                source: source.to_string(),
214                table: table.to_string(),
215                profile: s.profile.as_deref().map(|p| render("profile", p)).transpose()?,
216                database: s.database.as_deref().map(|d| render("database", d)).transpose()?,
217                schema: render("schema", s.schema.as_deref().unwrap_or(&s.name))?,
218                identifier: render("identifier", t.identifier.as_deref().unwrap_or(&t.name))?,
219                quoting: t.quoting.over(&s.quoting),
220            })
221        })();
222        cache.lock().unwrap().insert(key, r.clone());
223        r
224    }))
225}
226
227/// Parse one Binding: render each query without a database, then resolve its connection.
228/// `vars` are the Binding's vars as the run sees them (with a schedule's layered in).
229pub fn binding(
230    project: &Project,
231    report: &Report,
232    b: &Binding,
233    vars: &JsonMap<String, Json>,
234    inputs: &Inputs,
235) -> ParsedBinding {
236    let ctx_name = match &b.set {
237        Some(s) => format!("report `{}`, Set `{s}`", report.name),
238        None => format!("report `{}`", report.name),
239    };
240    let context = run_context(
241        project,
242        report,
243        b,
244        inputs,
245        inputs.started_at.unwrap_or_else(Utc::now),
246    );
247    let limited = Arc::new(Limited::new(
248        context.clone(),
249        vars.clone(),
250        inputs.cli_vars.clone(),
251    ));
252    let mut out = ParsedBinding::default();
253    let error = |out: &mut ParsedBinding, code: Code, 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: Code::InvalidProfileValue,
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                        Code::ConnectionConflict,
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                Code::NoConnection,
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: Code::SetupOnOtherConnection,
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
