Skip to main content

dre_core/
run.rs

1//! The run path: turn a resolved project into delivered files.
2//!
3//! Per Binding: parse pass → render → split → (unmanaged check) → execute, each query on its
4//! connection's session (one per connection, opened on first use, in strict YAML order) →
5//! format into `target/run/` → schema-drift check → deliver → snapshot → `run_results.json`.
6
7use std::collections::BTreeMap;
8use std::fs::File;
9use std::path::{Path, PathBuf};
10use std::sync::atomic::{AtomicBool, Ordering};
11use std::sync::{Arc, Mutex};
12use std::time::{Duration, Instant};
13
14use arrow::array::RecordBatch;
15use arrow::datatypes::SchemaRef;
16use arrow::ipc::reader::FileReader;
17use arrow::ipc::writer::FileWriter;
18use chrono::NaiveDate;
19use dre_protocol::host::{Execution, HostError, LogSink, PluginProcess};
20use dre_protocol::msg::{ColumnOptions, DeliveryFile, ResultSetMeta};
21use dre_protocol::{CAP_LOAD, CAP_MULTI_FILE, CAP_READ_ONLY, CAP_SESSIONS};
22use serde::Serialize;
23use serde_json::{Map as JsonMap, Value as Json, json};
24
25use crate::dates::Calendar;
26use crate::lookups::Table;
27use crate::parse::ParsedBinding;
28use crate::profiles::{BUILTIN_LOCAL, LOCAL_TYPE, ProfileTarget, Profiles, Role};
29use crate::project::{Binding, PluginKind, Project, QueryEntry, Report};
30use crate::render::{
31    Column, Connection, Connections, Mode, QueryRows, QueryRunner, RenderError, Renderer, RendererConfig,
32};
33use crate::selector;
34use crate::sqlsplit::{self, StatementKind};
35
36/// What the caller asked for.
37#[derive(Debug, Clone, Default)]
38pub struct RunOptions {
39    pub selector: Option<String>,
40    /// `--set <name>` or `--set all`.
41    pub set: Option<String>,
42    /// `--target` as given; the run's target is [`Project::target_name`].
43    pub target: Option<String>,
44    /// `--profile`: the inherited connection, for every Binding.
45    pub profile: Option<String>,
46    pub vars: BTreeMap<String, String>,
47    pub output_name: Option<String>,
48    pub output_path: Option<String>,
49    pub dry_run: bool,
50    /// `--preview [N]`: row limit per query.
51    pub preview: Option<u64>,
52    pub accept_schema_change: bool,
53    /// Whether a person is at a terminal to answer prompts.
54    pub interactive: bool,
55    /// `run.date` (`DRE_RUN_DATE`).
56    pub date: Option<NaiveDate>,
57    /// The instant this run was scheduled for (`DRE_RUN_AT`): `run.now` and `run.scheduled_at`,
58    /// and `run.date` in the run's timezone unless `date` is set.
59    pub scheduled_at: Option<chrono::DateTime<chrono::Utc>>,
60    /// `validate --live`: check statements instead of executing them.
61    pub live_check: bool,
62    /// `--schedule <name>`: run exactly the Bindings that schedule targets, with its vars.
63    pub schedule: Option<String>,
64    /// `--timezone` or `DRE_TIMEZONE`: above every configured `timezone:`.
65    pub timezone: Option<String>,
66    /// SHA-256 of the manifest this command wrote, for `run_results.json`.
67    pub manifest_checksum: Option<String>,
68}
69
70impl RunOptions {
71    /// What the run was asked to do, for `run_results.json` and the log.
72    pub fn params(&self, date: NaiveDate) -> Json {
73        json!({
74            "selector": self.selector,
75            "set": self.set,
76            "schedule": self.schedule,
77            "target": self.target,
78            "profile": self.profile,
79            "vars": self.vars,
80            "run_date": date.to_string(),
81            "scheduled_at": self.scheduled_at.map(rfc3339),
82            "timezone": self.timezone,
83            "output_name": self.output_name,
84            "output_path": self.output_path,
85            "dry_run": self.dry_run,
86            "preview": self.preview,
87            "accept_schema_change": self.accept_schema_change,
88        })
89    }
90}
91
92/// How much a message matters: `Info` is shown by default, `Debug` with `-v`.
93#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
94pub enum Level {
95    Debug,
96    Info,
97}
98
99/// How the run reports progress. The engine emits structured events; the CLI decides how
100/// they look (colour, progress bar, JSON, log file).
101pub trait Ui {
102    /// The number of Bindings about to run, once Sets are resolved.
103    fn plan(&mut self, _bindings: usize) {}
104    fn binding_start(&mut self, _report: &str, _set: Option<&str>) {}
105    /// Right after `binding_start`: the schedule it runs under (if any) and every var it uses.
106    fn binding_vars(
107        &mut self,
108        _schedule: Option<&str>,
109        _schedule_vars: Option<&JsonMap<String, Json>>,
110        _vars: &JsonMap<String, Json>,
111    ) {
112    }
113    /// One step inside the current Binding: a short verb, a detail, and how long it took.
114    fn step(&mut self, level: Level, verb: &str, detail: &str, elapsed: Option<Duration>);
115    fn warn(&mut self, msg: &str);
116    fn binding_end(&mut self, _outcome: &BindingOutcome) {}
117    /// A Binding was compiled (`--dry-run`, `dre compile`, `dre validate`): what it would do.
118    fn compiled(&mut self, _plan: &BindingPlan) {}
119    /// Ask which Set to run; `None` means "all".
120    fn choose_set(&mut self, report: &str, sets: &[String]) -> Result<Option<String>, String>;
121    /// Where plugin stderr goes.
122    fn plugin_log(&self) -> LogSink;
123    /// Where the full text of every statement sent to a source goes: `(label, sql)`.
124    fn sql_log(&self) -> LogSink {
125        Arc::new(|_, _| {})
126    }
127}
128
129#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
130#[serde(rename_all = "snake_case")]
131pub enum Status {
132    Success,
133    Error,
134    DryRun,
135    Checked,
136}
137
138#[derive(Debug, Clone, Serialize)]
139pub struct BindingOutcome {
140    pub report: String,
141    #[serde(skip_serializing_if = "Option::is_none")]
142    pub set: Option<String>,
143    pub binding: String,
144    pub status: Status,
145    #[serde(skip_serializing_if = "Option::is_none")]
146    pub error: Option<String>,
147    pub files: Vec<PathBuf>,
148    /// One line describing what the Binding produced (result sets, rows, outputs, delivery).
149    pub summary: String,
150    /// The schedule this ran under (`--schedule`), and the vars it layered in.
151    pub schedule: Option<String>,
152    pub schedule_vars: Option<JsonMap<String, Json>>,
153    /// Every var the Binding rendered with: its own, the schedule's, then `--var`.
154    pub vars: JsonMap<String, Json>,
155    /// The run's timezone (IANA name); empty when the Binding never started.
156    #[serde(skip_serializing_if = "String::is_empty")]
157    pub timezone: String,
158    #[serde(skip)]
159    pub elapsed: Duration,
160}
161
162/// What a compiled Binding would use and produce if it ran.
163#[derive(Debug, Clone, Serialize)]
164pub struct BindingPlan {
165    pub report: String,
166    pub set: Option<String>,
167    /// Rendered SQL files, relative to the project.
168    pub compiled: Vec<PathBuf>,
169    /// The inherited connection, if any.
170    pub profile: Option<String>,
171    /// Each query's connection, its type and the sources it reads.
172    pub queries: Vec<PlannedQuery>,
173    pub target: String,
174    pub format: String,
175    /// The file written under `target/run/` (a single-table format writes one per result set
176    /// when there are several).
177    pub output: PathBuf,
178    pub destinations: Vec<PlannedDestination>,
179    pub schedules: Vec<String>,
180}
181
182#[derive(Debug, Clone, Serialize)]
183pub struct PlannedQuery {
184    pub query: String,
185    pub connection: String,
186    #[serde(rename = "type")]
187    pub kind: String,
188    #[serde(skip_serializing_if = "Vec::is_empty")]
189    pub sources: Vec<String>,
190}
191
192#[derive(Debug, Clone, Serialize)]
193pub struct PlannedDestination {
194    pub profile: String,
195    /// The destination type for the active target, when it has one.
196    pub kind: Option<String>,
197    pub target: Option<String>,
198    pub path: Option<String>,
199    /// False when the profile has no target for this run: the output stays in `target/`.
200    pub delivers: bool,
201}
202
203#[derive(Debug, Default)]
204pub struct RunSummary {
205    pub outcomes: Vec<BindingOutcome>,
206    /// Selection failed before anything ran.
207    pub error: Option<String>,
208}
209
210impl RunSummary {
211    pub fn failed(&self) -> bool {
212        self.error.is_some() || self.outcomes.iter().any(|o| o.status == Status::Error)
213    }
214}
215
216pub fn run(project: &Project, opts: &RunOptions, ui: &mut dyn Ui) -> RunSummary {
217    let mut summary = RunSummary::default();
218    if let Err(e) = crate::target::ensure(&project.target_dir) {
219        summary.error = Some(format!(
220            "can't write to the target path {} (from {}): {e}",
221            project.target_dir.display(),
222            project.target_source
223        ));
224        return summary;
225    }
226    if let Some(name) = &opts.schedule {
227        if !project.schedules.iter().any(|e| &e.name == name) {
228            summary.error = Some(unknown_schedule(project, name));
229            return summary;
230        }
231        let planned = match schedule_bindings(project, name, opts) {
232            Ok(p) => p,
233            Err(e) => {
234                summary.error = Some(e);
235                return summary;
236            }
237        };
238        ui.plan(planned.len());
239        for (report, b) in &planned {
240            ui.binding_start(&report.name, b.set.as_deref());
241            let mut r = BindingRun::new(project, report, b, opts, ui);
242            let outcome = r.run();
243            ui.binding_end(&outcome);
244            summary.outcomes.push(outcome);
245        }
246        return summary;
247    }
248    let reports: Vec<&Report> = match &opts.selector {
249        None => project.reports.iter().collect(),
250        Some(s) => match selector::resolve(project, s) {
251            Ok(r) if r.is_empty() => {
252                summary.error = Some(format!("selector `{s}` matches no report"));
253                return summary;
254            }
255            Ok(r) => r,
256            Err(e) => {
257                summary.error = Some(e.to_string());
258                return summary;
259            }
260        },
261    };
262    // Resolve every report's Bindings first (prompts happen here), so progress has a total.
263    let mut planned: Vec<(&Report, Binding)> = Vec::new();
264    for report in reports {
265        match choose_bindings(project, report, opts, ui) {
266            Ok(bs) => planned.extend(bs.into_iter().map(|b| (report, b))),
267            Err(e) => {
268                let outcome = BindingOutcome {
269                    report: report.name.clone(),
270                    set: None,
271                    binding: "-".into(),
272                    status: Status::Error,
273                    error: Some(e),
274                    files: Vec::new(),
275                    summary: String::new(),
276                    schedule: opts.schedule.clone(),
277                    schedule_vars: None,
278                    vars: JsonMap::new(),
279                    timezone: String::new(),
280                    elapsed: Duration::ZERO,
281                };
282                ui.binding_end(&outcome);
283                summary.outcomes.push(outcome);
284            }
285        }
286    }
287    ui.plan(planned.len());
288    for (report, b) in &planned {
289        ui.binding_start(&report.name, b.set.as_deref());
290        let mut r = BindingRun::new(project, report, b, opts, ui);
291        let outcome = r.run();
292        ui.binding_end(&outcome);
293        summary.outcomes.push(outcome);
294    }
295    summary
296}
297
298/// The Bindings `--schedule <name>` runs: all of them, or those a selector and/or `--set` pick.
299/// Picking something the schedule doesn't run is an error naming what it does run.
300fn schedule_bindings<'a>(
301    project: &'a Project,
302    name: &str,
303    opts: &RunOptions,
304) -> Result<Vec<(&'a Report, Binding)>, String> {
305    let reports: Option<Vec<&str>> = match &opts.selector {
306        None => None,
307        Some(s) => Some(
308            selector::resolve(project, s)
309                .map_err(|e| e.to_string())?
310                .into_iter()
311                .map(|r| r.name.as_str())
312                .collect(),
313        ),
314    };
315    let set = opts.set.as_deref().filter(|s| *s != "all");
316    let all: Vec<(&Report, &Binding)> = project
317        .reports
318        .iter()
319        .flat_map(|r| {
320            r.bindings
321                .iter()
322                .filter(|b| b.schedules.iter().any(|s| s == name))
323                .map(move |b| (r, b))
324        })
325        .collect();
326    let picked: Vec<(&Report, Binding)> = all
327        .iter()
328        .filter(|(r, _)| reports.as_ref().is_none_or(|rs| rs.contains(&r.name.as_str())))
329        .filter(|(_, b)| set.is_none() || b.set.as_deref() == set)
330        .map(|(r, b)| {
331            let mut b = (*b).clone();
332            if let Some(p) = &opts.profile {
333                b.profile = Some(p.clone());
334                b.profile_at = Some(profile_flag_at());
335            }
336            (*r, b)
337        })
338        .collect();
339    if picked.is_empty() && (reports.is_some() || set.is_some()) {
340        let mut asked = Vec::new();
341        if let Some(s) = &opts.selector {
342            asked.push(format!("`{s}`"));
343        }
344        if let Some(s) = set {
345            asked.push(format!("Set `{s}`"));
346        }
347        let runs: Vec<String> = all
348            .iter()
349            .map(|(r, b)| match &b.set {
350                Some(s) => format!("{}/{s}", r.name),
351                None => r.name.clone(),
352            })
353            .collect();
354        return Err(format!(
355            "schedule `{name}` doesn't run {}; it runs: {}",
356            asked.join(" with "),
357            if runs.is_empty() {
358                "nothing".to_string()
359            } else {
360                runs.join(", ")
361            }
362        ));
363    }
364    Ok(picked)
365}
366
367/// The usage error for `--schedule` with a name that isn't declared.
368pub fn unknown_schedule(project: &Project, name: &str) -> String {
369    let names: Vec<&str> = project.schedules.iter().map(|e| e.name.as_str()).collect();
370    if names.is_empty() {
371        format!("no schedule `{name}`: this project's schedules.yml declares none")
372    } else {
373        format!("no schedule `{name}`; valid names: {}", names.join(", "))
374    }
375}
376
377/// Which Bindings of a report run, following ADR 0009's Set rules.
378pub fn choose_bindings(
379    project: &Project,
380    report: &Report,
381    opts: &RunOptions,
382    ui: &mut dyn Ui,
383) -> Result<Vec<Binding>, String> {
384    let mut out: Vec<Binding> = match opts.set.as_deref() {
385        Some("all") => report.bindings.clone(),
386        Some(name) => match report.binding(name) {
387            Some(b) => vec![b.clone()],
388            None => vec![ad_hoc(project, report, name)],
389        },
390        None if !report.has_sets => report.bindings.clone(),
391        None => {
392            let names: Vec<String> = report.bindings.iter().filter_map(|b| b.set.clone()).collect();
393            let pick = report
394                .default_set
395                .clone()
396                .or_else(|| project.default_set.clone().filter(|d| names.contains(d)))
397                .or_else(|| (names.len() == 1).then(|| names[0].clone()));
398            match pick {
399                Some(p) => report.binding(&p).cloned().into_iter().collect(),
400                None if opts.interactive => match ui.choose_set(&report.name, &names)? {
401                    Some(p) => report.binding(&p).cloned().into_iter().collect(),
402                    None => report.bindings.clone(),
403                },
404                None => {
405                    return Err(format!(
406                        "report `{}` has several Sets ({}) and no `default_set`; declare `default_set:` or pass `--set <name>` or `--set all`",
407                        report.name,
408                        names.join(", ")
409                    ));
410                }
411            }
412        }
413    };
414    if let Some(p) = &opts.profile {
415        for b in &mut out {
416            b.profile = Some(p.clone());
417            b.profile_at = Some(profile_flag_at());
418        }
419    }
420    Ok(out)
421}
422
423/// A Set that isn't declared on the report: start from the report itself, then apply the
424/// `sets.yml` entry of that name if there is one. `--profile`/`--var` complete it.
425fn ad_hoc(project: &Project, report: &Report, name: &str) -> Binding {
426    let mut b = report.base.clone();
427    b.set = Some(name.to_string());
428    if let Some(reg) = project.sets.get(name) {
429        if let Some(p) = &reg.profile {
430            b.profile = Some(p.clone());
431            b.profile_at = Some(crate::project::ProfileAt {
432                file: reg.file.clone(),
433                line: reg.line,
434                key: format!("`profile` of Set `{name}`"),
435            });
436        }
437        b.vars.extend(reg.vars.clone());
438    }
439    b
440}
441
442/// `--profile` as the source of a Binding's inherited connection, for messages.
443fn profile_flag_at() -> crate::project::ProfileAt {
444    crate::project::ProfileAt {
445        file: PathBuf::from("--profile"),
446        line: None,
447        key: "`--profile`".into(),
448    }
449}
450
451/// The tab one query produced: its last statement's result set, spooled to disk.
452struct Produced {
453    query: String,
454    connection: String,
455    schema: SchemaRef,
456    rows: u64,
457    spool: PathBuf,
458    name: String,
459    anchor: Option<String>,
460    header: Option<bool>,
461    columns: BTreeMap<String, ColumnOptions>,
462}
463
464struct Statement {
465    query: String,
466    /// The connection it runs on.
467    connection: String,
468    file: PathBuf,
469    line: usize,
470    text: String,
471    kind: StatementKind,
472    /// The complete rendered query contained a registered secret before it was split. Every
473    /// statement inherits this because splitting can itself separate a secret into fragments.
474    sensitive: bool,
475    /// Whether this statement's result is its query's tab: the file's last statement, when the
476    /// query has `tab: true`.
477    tab: bool,
478}
479
480struct BindingRun<'a> {
481    project: &'a Project,
482    report: &'a Report,
483    b: &'a Binding,
484    /// The Binding's vars with the schedule's layered on (`--var` stays separate, on top).
485    vars: JsonMap<String, Json>,
486    /// Everything `var()` sees, `--var` included: what's recorded.
487    rendered_vars: JsonMap<String, Json>,
488    schedule_vars: Option<JsonMap<String, Json>>,
489    opts: &'a RunOptions,
490    /// `run.date`: `DRE_RUN_DATE`, else today in the run's timezone.
491    date: NaiveDate,
492    calendar: Calendar,
493    ui: &'a mut dyn Ui,
494    compiled_dir: PathBuf,
495    run_dir: PathBuf,
496    schema_dir: PathBuf,
497    /// The run's target (environment).
498    target: String,
499    /// The parse pass for this run's inputs.
500    parsed: ParsedBinding,
501    /// The Binding's connections and their sessions.
502    pool: Option<Arc<Pool>>,
503    /// One renderer per connection (`None`: the inherited one, for paths and template values).
504    renderers: BTreeMap<Option<String>, Arc<Renderer>>,
505    started: Instant,
506    started_at: chrono::DateTime<chrono::Utc>,
507    produced: Vec<Produced>,
508    files: Vec<(PathBuf, Option<String>)>,
509    /// The Binding's destinations with paths and options rendered.
510    dests: Vec<RenderedDest>,
511    /// One record per destination: `run_results.json`'s `deliveries`.
512    deliveries: Vec<Json>,
513    delivery_note: Option<String>,
514    drift: Vec<String>,
515    /// The current schema with stable identities and any protection carried from the baseline.
516    protected_snapshot: Option<Json>,
517}
518
519/// A destination entry after rendering.
520struct RenderedDest {
521    profile: String,
522    path: Option<String>,
523    options: JsonMap<String, Json>,
524}
525
526type Fail = String;
527
528fn masked_source_error(error: &HostError, sensitive: bool) -> String {
529    let error = error.to_string();
530    crate::secrets::mask_source_error(&error, sensitive).into_owned()
531}
532
533fn masked_sql_summary(sql: &str, max: usize, sensitive: bool) -> String {
534    if sensitive {
535        crate::secrets::MASK.to_string()
536    } else {
537        dre_protocol::util::summarize(&crate::secrets::mask(sql), max)
538    }
539}
540
541fn protected_sql(sql: &str, sensitive: bool) -> std::borrow::Cow<'_, str> {
542    if sensitive {
543        std::borrow::Cow::Borrowed(crate::secrets::MASK)
544    } else {
545        crate::secrets::mask(sql)
546    }
547}
548
549impl<'a> BindingRun<'a> {
550    fn new(
551        project: &'a Project,
552        report: &'a Report,
553        b: &'a Binding,
554        opts: &'a RunOptions,
555        ui: &'a mut dyn Ui,
556    ) -> Self {
557        let t = &project.target_dir;
558        let rel = Path::new(&report.name).join(b.dir_name());
559        let schedule = opts
560            .schedule
561            .as_ref()
562            .and_then(|n| project.schedules.iter().find(|e| &e.name == n));
563        let schedule_vars = schedule.map(|e| e.vars.clone());
564        // Names were checked when the project and the command line were read.
565        let calendar: Calendar = crate::parse::calendar(
566            project,
567            report,
568            opts.timezone.as_deref(),
569            opts.schedule.as_deref(),
570        );
571        let date = crate::parse::run_date(&calendar, opts.date, opts.scheduled_at);
572        // Binding vars, then the schedule's, then `--var` on top.
573        let mut vars = b.vars.clone();
574        vars.extend(schedule_vars.clone().unwrap_or_default());
575        let rendered_vars = {
576            let mut v = vars.clone();
577            v.extend(
578                opts.vars
579                    .iter()
580                    .map(|(k, x)| (k.clone(), Json::String(x.clone()))),
581            );
582            v
583        };
584        ui.binding_vars(opts.schedule.as_deref(), schedule_vars.as_ref(), &rendered_vars);
585        BindingRun {
586            project,
587            report,
588            b,
589            vars,
590            rendered_vars,
591            schedule_vars,
592            opts,
593            date,
594            calendar,
595            ui,
596            compiled_dir: t.join("compiled").join(&rel),
597            run_dir: t.join("run").join(&rel),
598            schema_dir: t.join("schema").join(&rel),
599            target: project.target_name.clone(),
600            parsed: ParsedBinding::default(),
601            pool: None,
602            renderers: BTreeMap::new(),
603            started: Instant::now(),
604            started_at: chrono::Utc::now(),
605            produced: Vec::new(),
606            files: Vec::new(),
607            dests: Vec::new(),
608            deliveries: Vec::new(),
609            delivery_note: None,
610            drift: Vec::new(),
611            protected_snapshot: None,
612        }
613    }
614
615    /// What `connection`, `destination` and `profile()` read for this Binding.
616    fn connections(&self) -> Arc<ProfileConnections> {
617        Arc::new(ProfileConnections {
618            profiles: self.project.profiles.clone(),
619            target: self.target.clone(),
620            root: self.project.root.clone(),
621            plugins: self.project.plugins.clone(),
622            log: self.ui.plugin_log(),
623        })
624    }
625
626    /// The renderer for SQL on `connection` (`None`: the inherited connection, for output paths
627    /// and template values), made on first use.
628    fn renderer(&mut self, connection: Option<&str>) -> Result<Arc<Renderer>, Fail> {
629        let key = connection.map(str::to_string);
630        if let Some(r) = self.renderers.get(&key) {
631            return Ok(r.clone());
632        }
633        let pool = self.pool.clone().expect("the pool is made before rendering");
634        // Outside query SQL, `connection.*` and `run_query()` mean the inherited connection.
635        let name = key.clone().or_else(|| self.parsed.inherited.clone());
636        let source_type = match (&key, &name) {
637            (Some(n), _) => pool.kind(n)?,
638            // The inherited connection's type matters only if something uses it.
639            (None, Some(n)) => pool.kind(n).unwrap_or_default(),
640            (None, None) => String::new(),
641        };
642        let context =
643            crate::parse::run_context(self.project, self.report, self.b, &self.inputs(), self.started_at);
644        let limited = Arc::new(crate::render::Limited::new(
645            context.clone(),
646            self.vars.clone(),
647            self.opts.vars.clone(),
648        ));
649        let r = Renderer::new(RendererConfig {
650            root: &self.project.root,
651            macros: &self.project.macros,
652            context,
653            vars: self.vars.clone(),
654            cli_vars: self.opts.vars.clone(),
655            runner: Some(Arc::new(PoolRunner {
656                pool: pool.clone(),
657                default: name.clone(),
658                log: self.ui.sql_log(),
659            })),
660            connections: Some(pool.connections.clone()),
661            mode: Mode::Run,
662            connection: name,
663            source_type,
664            sources: crate::parse::resolver(self.project, limited),
665            run_query_max_rows: self.project.run_query_max_rows,
666            sql: self.project.sql.clone(),
667            lookups: self.project.lookups.clone(),
668            lookup_inline_max_rows: self.project.lookup_inline_max_rows,
669            packages: self.project.packages.clone(),
670            project_name: self.project.name.clone(),
671            dispatch: self.project.dispatch.clone(),
672        })
673        .map_err(|e| e.to_string())?;
674        let r = Arc::new(r);
675        self.renderers.insert(key, r.clone());
676        Ok(r)
677    }
678
679    /// The parse pass's inputs for this run.
680    fn inputs(&self) -> crate::parse::Inputs {
681        crate::parse::Inputs {
682            target: self.target.clone(),
683            cli_vars: self.opts.vars.clone(),
684            date: self.opts.date,
685            scheduled_at: self.opts.scheduled_at,
686            timezone: self.opts.timezone.clone(),
687            schedule: self.opts.schedule.clone(),
688            started_at: Some(self.started_at),
689        }
690    }
691
692    /// Each query's connection, from the parse pass.
693    fn connection_of(&self, query: &str) -> Result<String, Fail> {
694        self.parsed
695            .query(query)
696            .and_then(|q| q.connection.clone())
697            .ok_or_else(|| format!("query `{query}` has no connection"))
698    }
699
700    fn outcome(&self, status: Status, error: Option<String>) -> BindingOutcome {
701        BindingOutcome {
702            report: self.report.name.clone(),
703            set: self.b.set.clone(),
704            binding: self.b.dir_name().to_string(),
705            status,
706            error,
707            files: self.files.iter().map(|(p, _)| p.clone()).collect(),
708            summary: self.summary_line(),
709            schedule: self.opts.schedule.clone(),
710            schedule_vars: self.schedule_vars.clone(),
711            vars: self.rendered_vars.clone(),
712            timezone: self.calendar.tz.name().to_string(),
713            elapsed: self.started.elapsed(),
714        }
715    }
716
717    fn run(&mut self) -> BindingOutcome {
718        let dry = self.opts.dry_run || self.opts.live_check;
719        if !dry {
720            let _ = std::fs::remove_dir_all(&self.run_dir);
721        }
722        let result = self.run_inner();
723        let status = match (&result, dry, self.opts.live_check) {
724            (Err(_), _, _) => Status::Error,
725            (Ok(()), _, true) => Status::Checked,
726            (Ok(()), true, _) => Status::DryRun,
727            (Ok(()), false, _) => Status::Success,
728        };
729        let err = result.err();
730        if !dry && let Err(e) = self.write_results(&status, err.as_deref()) {
731            self.ui.warn(&format!("can't write run_results.json: {e}"));
732        }
733        // Spools are scratch space.
734        let _ = std::fs::remove_dir_all(self.run_dir.join(".spool"));
735        self.outcome(status, err)
736    }
737
738    fn run_inner(&mut self) -> Result<(), Fail> {
739        // The parse pass with this run's inputs: each query's connection and sources.
740        self.parsed = crate::parse::binding(self.project, self.report, self.b, &self.vars, &self.inputs());
741        if !self.parsed.errors.is_empty() {
742            let msgs: Vec<String> = self
743                .parsed
744                .errors
745                .iter()
746                .map(|p| match p.line {
747                    Some(l) => format!("{}:{l}: {}", p.file.display(), p.message),
748                    None => format!("{}: {}", p.file.display(), p.message),
749                })
750                .collect();
751            return Err(msgs.join("\n    "));
752        }
753        // Sessions are opened only when something needs the database, so compiling a report
754        // whose templates don't query it works without the plugin or credentials.
755        let pool = Arc::new(Pool::new(
756            self.connections(),
757            self.project.root.clone(),
758            self.ui.plugin_log(),
759            !self.report.managed,
760        ));
761        self.pool = Some(pool.clone());
762
763        if !self.report.managed {
764            self.ui.warn(&format!(
765                "`{}` is an unmanaged report ({}): for quick tests only — add a YAML to make it a managed report",
766                self.report.name,
767                self.report.file.display()
768            ));
769        }
770
771        // 1. Render.
772        std::fs::create_dir_all(&self.compiled_dir).map_err(|e| e.to_string())?;
773        // How many queries run on each connection: one session holds them all.
774        let mut per_connection: BTreeMap<String, usize> = BTreeMap::new();
775        for q in &self.b.queries {
776            *per_connection.entry(self.connection_of(&q.query)?).or_default() += 1;
777        }
778        // A real run of a managed report renders each query just before running it, so a
779        // template can look at what earlier queries made (`columns()` of a temp table).
780        // Compiling, checking and unmanaged reports render everything first.
781        let interleave = !self.opts.dry_run && !self.opts.live_check && self.report.managed;
782        let mut statements = Vec::new();
783        // Reported after the unmanaged-report check, which matters more.
784        let mut two_tabs: Option<String> = None;
785        if interleave {
786            self.dests = self.render_destinations()?;
787            let mut checked: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
788            for (c, n) in &per_connection {
789                if *n > 1 {
790                    self.check_sessions(&pool, c, *n)?;
791                    checked.insert(c.clone());
792                }
793            }
794            std::fs::create_dir_all(self.run_dir.join(".spool")).map_err(|e| e.to_string())?;
795            let mut i = 0;
796            let queries = self.b.queries.clone();
797            for q in &queries {
798                let conn = self.connection_of(&q.query)?;
799                let renderer = self.renderer(Some(&conn))?;
800                let sts = self.render_query(&renderer, q, &conn, &mut two_tabs)?;
801                for w in renderer.take_warnings() {
802                    self.ui.warn(&w);
803                }
804                if let Some(e) = two_tabs.take() {
805                    return Err(e);
806                }
807                if sts.len() > 1 && checked.insert(conn.clone()) {
808                    self.check_sessions(&pool, &conn, sts.len())?;
809                }
810                let session = pool.session(&conn)?;
811                for st in &sts {
812                    self.execute_statement(&session, i, st)?;
813                    i += 1;
814                }
815            }
816        } else {
817            let queries = self.b.queries.clone();
818            for q in &queries {
819                let conn = self.connection_of(&q.query)?;
820                let renderer = self.renderer(Some(&conn))?;
821                statements.extend(self.render_query(&renderer, q, &conn, &mut two_tabs)?);
822            }
823            self.dests = self.render_destinations()?;
824            for r in self.renderers.values() {
825                for w in r.take_warnings() {
826                    self.ui.warn(&w);
827                }
828            }
829
830            // 3. Unmanaged: every rendered statement must only read (or create temp objects).
831            if !self.report.managed {
832                if let Some(bad) = statements.iter().find(|s| !s.kind.is_read_only_safe()) {
833                    return Err(format!(
834                        "{}:{}: unmanaged report `{}` may only run SELECT/WITH or CREATE [OR REPLACE] TEMP|TEMPORARY TABLE|VIEW, but found `{}`; nothing was run — rewrite the statement, or give the report a YAML to declare it",
835                        bad.file.display(),
836                        bad.line,
837                        self.report.name,
838                        masked_sql_summary(&bad.text, 60, bad.sensitive)
839                    ));
840                }
841                for conn in per_connection.keys() {
842                    let session = pool.session(conn)?;
843                    let creates_temp = statements
844                        .iter()
845                        .any(|s| &s.connection == conn && s.kind == StatementKind::TempCreate)
846                        || session.lock().unwrap().loaded;
847                    session.lock().unwrap().want_read_only(!creates_temp);
848                }
849            }
850            if let Some(e) = two_tabs {
851                return Err(e);
852            }
853
854            if self.opts.dry_run {
855                let plan = self.plan();
856                self.ui.compiled(&plan);
857                return Ok(());
858            }
859
860            // The one-session-per-connection guarantee.
861            for conn in per_connection.keys() {
862                let n = statements.iter().filter(|s| &s.connection == conn).count();
863                if n > 1 {
864                    self.check_sessions(&pool, conn, n)?;
865                }
866            }
867
868            if self.opts.live_check {
869                return self.live_check(&statements);
870            }
871
872            // 4. Execute in YAML order, each on its connection's session.
873            std::fs::create_dir_all(self.run_dir.join(".spool")).map_err(|e| e.to_string())?;
874            for (i, st) in statements.iter().enumerate() {
875                let session = pool.session(&st.connection)?;
876                self.execute_statement(&session, i, st)?;
877            }
878        }
879        pool.close_all();
880
881        self.name_result_sets()?;
882
883        // 5. Format into target/run/.
884        let filename = self.file_names();
885        self.format(&filename)?;
886
887        // Schema drift, before delivery.
888        if self.opts.preview.is_none() {
889            self.drift = self.schema_drift();
890            if !self.drift.is_empty() {
891                if self.opts.accept_schema_change {
892                    self.ui.warn(&format!(
893                        "  schema changed since the last successful run (accepted): {}",
894                        self.drift.join("; ")
895                    ));
896                } else {
897                    return Err(format!(
898                        "schema drift since the last successful run: {}; the output is in {} but was not delivered — pass --accept-schema-change to deliver it and accept the new schema",
899                        self.drift.join("; "),
900                        rel(&self.project.root, &self.run_dir).display()
901                    ));
902                }
903            }
904        }
905
906        // 6. Deliver.
907        if self.opts.preview.is_some() {
908            self.delivery_note = Some("preview: not delivered".into());
909            let dir = rel(&self.project.root, &self.run_dir);
910            self.ui.step(
911                Level::Info,
912                "Preview",
913                &format!("not delivered; output stays in {}", dir.display()),
914                None,
915            );
916        } else {
917            self.deliver()?;
918        }
919
920        // 7. Snapshot the schema for the next drift check.
921        if self.opts.preview.is_none()
922            && let Err(error) = self.write_snapshot()
923        {
924            self.ui.warn(&format!(
925                "delivery completed, but the schema snapshot could not be written: {error}; the next run may compare against an older schema"
926            ));
927        }
928        Ok(())
929    }
930
931    /// Render one query file into its statements, writing it to `target/compiled/`.
932    fn render_query(
933        &mut self,
934        renderer: &Renderer,
935        q: &QueryEntry,
936        connection: &str,
937        two_tabs: &mut Option<String>,
938    ) -> Result<Vec<Statement>, Fail> {
939        let mut out = Vec::new();
940        let src = std::fs::read_to_string(self.project.root.join(&q.path))
941            .map_err(|e| format!("{}: {e}", q.path.display()))?;
942        let _ = renderer.take_sources();
943        let sql = renderer
944            .render(&q.path, &src)
945            .map_err(|e: RenderError| e.to_string())?;
946        // The manifest's `depends_on` and the query's connection come from the parse pass, so
947        // a `source()` it couldn't see would make both wrong.
948        let known = self
949            .parsed
950            .query(&q.query)
951            .map(|p| p.sources.clone())
952            .unwrap_or_default();
953        if let Some((s, t)) = renderer
954            .take_sources()
955            .into_iter()
956            .find(|(s, t)| !known.contains(&format!("{s}.{t}")))
957        {
958            return Err(format!(
959                "{}: `source('{s}', '{t}')` was reached while rendering `{}`, but the parse pass didn't find it. The parse pass renders without data (`run_query()` returns no rows, `connection.*` nothing), so it can't see a `source()` that depends on them; call it where it's reached either way",
960                q.path.display(),
961                q.query
962            ));
963        }
964        let sensitive = crate::secrets::contains_secret(&sql);
965        std::fs::write(
966            self.compiled_dir.join(format!("{}.sql", q.query)),
967            crate::secrets::mask(&sql).as_bytes(),
968        )
969        .map_err(|e| e.to_string())?;
970        self.ui
971            .step(Level::Debug, "Rendered", &q.path.display().to_string(), None);
972        // 2. Split. One file makes at most one tab, from its last statement, so any earlier
973        // SELECT would be lost: that's an error, not a guess.
974        let parts = sqlsplit::split(&sql);
975        if q.tab && parts.is_empty() {
976            return Err(format!(
977                "{}: `{}` makes a tab but has no statements; add a query, or `tab: false` if it's meant to be empty",
978                q.path.display(),
979                q.query
980            ));
981        }
982        let n = parts.len();
983        for (i, st) in parts.into_iter().enumerate() {
984            let body = sqlsplit::strip_leading_comments(&st.text);
985            let line = st.line + st.text[..st.text.len() - body.len()].matches('\n').count();
986            let kind = sqlsplit::classify(&st.text);
987            let last = i + 1 == n;
988            if q.tab && !last && kind == StatementKind::Read && two_tabs.is_none() {
989                *two_tabs = Some(format!(
990                    "{}:{line}: `{}` has more than one SELECT, but one .sql file makes one tab (from its last statement); put each tab's query in its own .sql file and list each in `queries:`",
991                    q.path.display(),
992                    q.query
993                ));
994            }
995            out.push(Statement {
996                query: q.query.clone(),
997                connection: connection.to_string(),
998                file: q.path.clone(),
999                line,
1000                kind,
1001                text: st.text,
1002                sensitive,
1003                tab: q.tab && last,
1004            });
1005        }
1006        Ok(out)
1007    }
1008
1009    /// Fail unless `connection`'s plugin can hold one session across `n` statements.
1010    fn check_sessions(&self, pool: &Pool, connection: &str, n: usize) -> Result<(), Fail> {
1011        let session = pool.session(connection)?;
1012        let mut s = session.lock().unwrap();
1013        if !s.get()?.has(CAP_SESSIONS) {
1014            let kind = pool.kind(connection).unwrap_or_default();
1015            return Err(format!(
1016                "this Binding runs {n} statements on connection `{connection}`, but the `{kind}` source plugin can't hold one session across them; nothing was run"
1017            ));
1018        }
1019        Ok(())
1020    }
1021
1022    /// Run statement `i`, spooling its result if it makes a tab.
1023    fn execute_statement(
1024        &mut self,
1025        session: &Arc<Mutex<Session>>,
1026        i: usize,
1027        st: &Statement,
1028    ) -> Result<(), Fail> {
1029        let sql_log = self.ui.sql_log();
1030        sql_log(
1031            &format!("{}:{}", st.file.display(), st.line),
1032            &protected_sql(&st.text, st.sensitive),
1033        );
1034        let spool_path = self.run_dir.join(".spool").join(format!("{i}.arrow"));
1035        let mut writer: Option<FileWriter<File>> = None;
1036        let t = Instant::now();
1037        let exec = {
1038            let mut s = session.lock().unwrap();
1039            let (p, _log_scope) = s.statement(st.sensitive)?;
1040            p.execute(&st.text, self.opts.preview, |schema, batch| {
1041                if writer.is_none() {
1042                    let f = File::create(&spool_path).map_err(|e| e.to_string())?;
1043                    writer = Some(FileWriter::try_new(f, schema).map_err(|e| e.to_string())?);
1044                }
1045                writer.as_mut().unwrap().write(&batch).map_err(|e| e.to_string())
1046            })
1047            .map_err(|e| {
1048                format!(
1049                    "{}:{}: {}",
1050                    st.file.display(),
1051                    st.line,
1052                    masked_source_error(&e, st.sensitive)
1053                )
1054            })?
1055        };
1056        let what = match &exec {
1057            Execution::Result { rows, .. } => {
1058                format!("{} row{}", thousands(*rows), if *rows == 1 { "" } else { "s" })
1059            }
1060            Execution::NoResult {
1061                rows_affected: Some(n),
1062            } => format!("no result set ({n} affected)"),
1063            Execution::NoResult { .. } => "no result set".to_string(),
1064        };
1065        let at = format!("{}:{}", st.file.display(), st.line);
1066        self.ui.step(
1067            Level::Debug,
1068            "Executed",
1069            &format!("{at}  {what}"),
1070            Some(t.elapsed()),
1071        );
1072        // The YAML decides the tabs; the result only has to match it. A tab with no rows
1073        // still gets its column names.
1074        let (schema, rows) = match (st.tab, exec) {
1075            (true, Execution::Result { schema, rows }) => (schema, rows),
1076            (true, Execution::NoResult { .. }) => {
1077                return Err(format!(
1078                    "{at}: `{}` makes a tab, but its last statement returned no result set; if the query only prepares data (a temp view, a SET), add `tab: false` to it in the YAML",
1079                    st.query
1080                ));
1081            }
1082            (false, _) => {
1083                drop(writer);
1084                let _ = std::fs::remove_file(&spool_path);
1085                return Ok(());
1086            }
1087        };
1088        {
1089            let mut w = match writer {
1090                Some(w) => w,
1091                None => FileWriter::try_new(File::create(&spool_path).map_err(|e| e.to_string())?, &schema)
1092                    .map_err(|e| e.to_string())?,
1093            };
1094            w.finish().map_err(|e| e.to_string())?;
1095            self.produced.push(Produced {
1096                query: st.query.clone(),
1097                connection: st.connection.clone(),
1098                schema,
1099                rows,
1100                spool: spool_path,
1101                name: String::new(),
1102                anchor: None,
1103                header: None,
1104                columns: Default::default(),
1105            });
1106        }
1107        Ok(())
1108    }
1109
1110    /// Each destination's rendered `profile`, `path` and options; `destination.*` is that
1111    /// destination while its values render.
1112    fn render_destinations(&mut self) -> Result<Vec<RenderedDest>, Fail> {
1113        if self.b.output.destinations.is_empty() {
1114            return Ok(Vec::new());
1115        }
1116        let renderer = self.renderer(None)?;
1117        let connections = self.pool.as_ref().expect("pool").connections.clone();
1118        let mut out = Vec::new();
1119        for (i, d) in self.b.output.destinations.iter().enumerate() {
1120            let profile = self
1121                .parsed
1122                .destinations
1123                .get(i)
1124                .cloned()
1125                .flatten()
1126                .ok_or_else(|| format!("destination `{}` has no profile", d.profile))?;
1127            renderer.set_destination(connections.profile(&profile, Some("destination")).ok());
1128            let rendered = (|| {
1129                let path = match &d.path {
1130                    Some(p) => Some(
1131                        renderer
1132                            .render(&self.report.file, p)
1133                            .map_err(|e| format!("output path: {e}"))?,
1134                    ),
1135                    None => None,
1136                };
1137                let mut options = JsonMap::new();
1138                for (k, v) in &d.options {
1139                    let v = render_json(&renderer, &self.report.file, v)
1140                        .map_err(|e| format!("destination `{profile}` option `{k}`: {e}"))?;
1141                    options.insert(k.clone(), v);
1142                }
1143                Ok::<_, Fail>(RenderedDest {
1144                    profile: profile.clone(),
1145                    path,
1146                    options,
1147                })
1148            })();
1149            renderer.set_destination(None);
1150            out.push(rendered?);
1151        }
1152        Ok(out)
1153    }
1154
1155    /// Tab names: `tab_name`, else the query's basename.
1156    fn name_result_sets(&mut self) -> Result<(), Fail> {
1157        let entries: BTreeMap<&str, &QueryEntry> =
1158            self.b.queries.iter().map(|q| (q.query.as_str(), q)).collect();
1159        for p in &mut self.produced {
1160            let q = entries[p.query.as_str()];
1161            p.name = q.tab_name.clone().unwrap_or_else(|| q.query.clone());
1162            p.anchor = q.anchor.clone();
1163            p.header = q.header;
1164            p.columns = q.columns.clone();
1165        }
1166        if self.b.output.format == "xlsx" {
1167            let mut seen: BTreeMap<String, String> = BTreeMap::new();
1168            for p in &self.produced {
1169                check_sheet_name(&p.name)?;
1170                if let Some(prev) = seen.insert(p.name.to_lowercase(), p.query.clone()) {
1171                    return Err(format!(
1172                        "sheet name `{}` is used twice (by `{prev}` and `{}`); Excel sheet names must be unique",
1173                        p.name, p.query
1174                    ));
1175                }
1176            }
1177        }
1178        Ok(())
1179    }
1180
1181    /// What a real run of this Binding would use and produce, for `dre compile` and `validate`.
1182    fn plan(&mut self) -> BindingPlan {
1183        let file = self.file_names();
1184        let compiled = self
1185            .b
1186            .queries
1187            .iter()
1188            .map(|q| {
1189                rel(
1190                    &self.project.root,
1191                    &self.compiled_dir.join(format!("{}.sql", q.query)),
1192                )
1193            })
1194            .collect();
1195        let destinations = self
1196            .dests
1197            .iter()
1198            .map(|d| {
1199                let resolved = self.dest_output(&d.profile);
1200                PlannedDestination {
1201                    profile: d.profile.clone(),
1202                    kind: resolved.as_ref().map(|(_, o)| o.kind.clone()),
1203                    target: resolved.map(|(t, _)| t).or_else(|| self.dest_target(&d.profile)),
1204                    path: d.path.clone(),
1205                    delivers: self.dest_output(&d.profile).is_some(),
1206                }
1207            })
1208            .collect();
1209        let pool = self.pool.clone().expect("pool");
1210        let queries = self
1211            .parsed
1212            .queries
1213            .iter()
1214            .map(|q| {
1215                let connection = q.connection.clone().unwrap_or_default();
1216                PlannedQuery {
1217                    query: q.query.clone(),
1218                    kind: pool.kind(&connection).unwrap_or_default(),
1219                    connection,
1220                    sources: q.sources.clone(),
1221                }
1222            })
1223            .collect();
1224        BindingPlan {
1225            report: self.report.name.clone(),
1226            set: self.b.set.clone(),
1227            compiled,
1228            profile: self.parsed.inherited.clone(),
1229            queries,
1230            target: self.target.clone(),
1231            format: self.b.output.format.clone(),
1232            output: rel(&self.project.root, &self.run_dir.join(file)),
1233            destinations,
1234            schedules: self.b.schedules.clone(),
1235        }
1236    }
1237
1238    /// Apply `--output-path`/`--output-name` to every destination that has a path, and return
1239    /// the local file name: `--output-name`, else the first destination path's file name, else
1240    /// `<report>.<ext>`.
1241    fn file_names(&mut self) -> String {
1242        for d in self.dests.iter_mut().filter(|d| d.path.is_some()) {
1243            if let Some(p) = &self.opts.output_path {
1244                d.path = Some(p.clone());
1245            }
1246            if let (Some(n), Some(r)) = (&self.opts.output_name, d.path.as_mut()) {
1247                *r = match r.rfind(['/', '\\']) {
1248                    Some(i) => format!("{}{n}", &r[..=i]),
1249                    None => n.clone(),
1250                };
1251            }
1252        }
1253        let ext = match &self.b.output.extension {
1254            Some(e) => e.as_str(),
1255            None => extension(&self.b.output.format),
1256        };
1257        let from_remote = self
1258            .dests
1259            .iter()
1260            .find_map(|d| d.path.as_deref())
1261            .and_then(|r| r.rsplit(['/', '\\']).next())
1262            .filter(|s| !s.is_empty())
1263            .map(str::to_string);
1264        self.opts
1265            .output_name
1266            .clone()
1267            .or(from_remote)
1268            .unwrap_or_else(|| match ext {
1269                "" => self.report.name.clone(),
1270                ext => format!("{}.{ext}", self.report.name),
1271            })
1272    }
1273
1274    fn format(&mut self, filename: &str) -> Result<(), Fail> {
1275        std::fs::create_dir_all(&self.run_dir).map_err(|e| e.to_string())?;
1276        if self.produced.is_empty() {
1277            self.ui
1278                .warn("  no query makes a tab (every query has `tab: false`); no output file was written");
1279            return Ok(());
1280        }
1281        let out = &self.b.output;
1282        let plugin = find_plugin(self.project, PluginKind::Format, &out.format)?;
1283        let mut p = plugin
1284            .start(self.ui.plugin_log(), Some(&self.project.root))
1285            .map_err(|e| e.to_string())?;
1286        let multi = out.format == "xlsx" || out.template.is_some() || !is_single_table(&out.format);
1287        let groups: Vec<(String, Vec<usize>)> = if multi || self.produced.len() == 1 {
1288            vec![(filename.to_string(), (0..self.produced.len()).collect())]
1289        } else {
1290            let (stem, ext) = split_ext(filename);
1291            let mut g = Vec::new();
1292            for (i, r) in self.produced.iter().enumerate() {
1293                if r.name.contains(['/', '\\']) {
1294                    return Err(format!("`{}` can't be used in a file name", r.name));
1295                }
1296                g.push((format!("{stem}_{}{ext}", r.name), vec![i]));
1297            }
1298            g
1299        };
1300        let template = match &out.template {
1301            Some(t) => Some(self.template_payload(t)?),
1302            None => None,
1303        };
1304        for (name, idxs) in groups {
1305            let t = Instant::now();
1306            let path = self.run_dir.join(&name);
1307            let metas: Vec<ResultSetMeta> = idxs
1308                .iter()
1309                .map(|&i| {
1310                    let r = &self.produced[i];
1311                    ResultSetMeta {
1312                        name: r.name.clone(),
1313                        query: r.query.clone(),
1314                        result_index: 1,
1315                        anchor: r.anchor.clone(),
1316                        header: r.header,
1317                        columns: r.columns.clone(),
1318                    }
1319                })
1320                .collect();
1321            p.write_begin(
1322                &path.to_string_lossy(),
1323                &out.format,
1324                out.options.clone(),
1325                metas,
1326                template.clone(),
1327            )
1328            .map_err(|e| e.to_string())?;
1329            for &i in &idxs {
1330                let r = &self.produced[i];
1331                let reader = FileReader::try_new(File::open(&r.spool).map_err(|e| e.to_string())?, None)
1332                    .map_err(|e| e.to_string())?;
1333                let schema = r.schema.clone();
1334                // Stream batch by batch so memory stays flat.
1335                let mut any = false;
1336                for b in reader {
1337                    let b = b.map_err(|e| e.to_string())?;
1338                    any = true;
1339                    p.send_batch(&b)
1340                        .map_err(|e| format!("{} format: {e}", out.format))?;
1341                }
1342                if !any {
1343                    p.send_batch(&RecordBatch::new_empty(schema))
1344                        .map_err(|e| format!("{} format: {e}", out.format))?;
1345                }
1346                p.send(&dre_protocol::msg::Request::ResultSetEnd {})
1347                    .map_err(|e| format!("{} format: {e}", out.format))?;
1348            }
1349            let (files, warnings) = p
1350                .write_finish_with_warnings()
1351                .map_err(|e| format!("{} format: {e}", out.format))?;
1352            for w in warnings {
1353                self.ui.warn(&format!("  {} format: {w}", out.format));
1354            }
1355            for f in files {
1356                let f = PathBuf::from(f);
1357                let size = std::fs::metadata(&f).map(|m| m.len()).unwrap_or(0);
1358                let shown = rel(&self.project.root, &f);
1359                self.ui.step(
1360                    Level::Debug,
1361                    "Wrote",
1362                    &format!("{} ({})", shown.display(), human_bytes(size)),
1363                    Some(t.elapsed()),
1364                );
1365                self.files.push((f, None));
1366            }
1367        }
1368        let _ = p.close();
1369        Ok(())
1370    }
1371
1372    fn template_payload(&mut self, t: &crate::project::Template) -> Result<Json, Fail> {
1373        let root = &self.project.root;
1374        let file = [root.join(&t.file), root.join("templates").join(&t.file)]
1375            .into_iter()
1376            .find(|p| p.is_file())
1377            .ok_or_else(|| format!("template file `{}` doesn't exist", t.file))?;
1378        let renderer = self.renderer(None)?;
1379        let mut values = JsonMap::new();
1380        for b in &t.bindings {
1381            if let (Some(cell), Some(v)) = (&b.cell, &b.value) {
1382                let rendered = renderer
1383                    .render(&self.report.file, v)
1384                    .map_err(|e| format!("template value: {e}"))?;
1385                values.insert(format!("{}!{}", b.sheet, cell), Json::String(rendered));
1386            }
1387        }
1388        Ok(json!({"file": file.to_string_lossy(), "bindings": t.bindings, "values": values}))
1389    }
1390
1391    /// Deliver to every destination in order. A failure is recorded and the rest are still
1392    /// attempted; the Binding fails if any did.
1393    fn deliver(&mut self) -> Result<(), Fail> {
1394        if self.dests.is_empty() {
1395            self.delivery_note = Some("no destination declared; output stays in target/".into());
1396            let dir = rel(&self.project.root, &self.run_dir);
1397            self.ui.step(
1398                Level::Debug,
1399                "Kept",
1400                &format!("no destination declared: output stays in {}", dir.display()),
1401                None,
1402            );
1403            return Ok(());
1404        }
1405        if self.files.is_empty() {
1406            return Ok(());
1407        }
1408        let dests = std::mem::take(&mut self.dests);
1409        let mut failures = Vec::new();
1410        let mut skipped = Vec::new();
1411        for d in &dests {
1412            // The type of the output used, or for a skipped entry the profile's default one.
1413            let kind = self
1414                .dest_output(&d.profile)
1415                .map(|(_, o)| o.kind.clone())
1416                .or_else(|| {
1417                    let p = self.project.profiles.get(Role::Destination, &d.profile)?;
1418                    p.targets.values().next().map(|o| o.kind.clone())
1419                });
1420            let (status, location, error) = match self.deliver_one(d) {
1421                Ok(Some(loc)) => ("delivered", Some(loc), None),
1422                Ok(None) => {
1423                    skipped.push(self.skip_note(&d.profile));
1424                    ("skipped", None, None)
1425                }
1426                Err(e) => {
1427                    failures.push(e.clone());
1428                    ("failed", None, Some(e))
1429                }
1430            };
1431            let mut record = json!({"profile": d.profile, "type": kind, "status": status});
1432            if let Some(l) = location {
1433                record["location"] = json!(l);
1434            }
1435            if let Some(e) = error {
1436                record["error"] = json!(e);
1437            }
1438            self.deliveries.push(record);
1439        }
1440        self.dests = dests;
1441        if self.files.iter().all(|(_, d)| d.is_none()) && !skipped.is_empty() {
1442            self.delivery_note = Some(skipped.join("; "));
1443        }
1444        match failures.len() {
1445            0 => Ok(()),
1446            1 if self.dests.len() == 1 => Err(failures.remove(0)),
1447            n => Err(format!(
1448                "{n} of {} destinations failed: {}",
1449                self.dests.len(),
1450                failures.join("; ")
1451            )),
1452        }
1453    }
1454
1455    /// The destination profile's settings for the run's target.
1456    fn dest_output(&self, profile: &str) -> Option<(String, &ProfileTarget)> {
1457        if self.project.profiles.is_builtin_local(profile) {
1458            return Some((self.target.clone(), &BUILTIN_LOCAL));
1459        }
1460        self.project
1461            .profiles
1462            .target(Role::Destination, profile, &self.target)
1463            .map(|o| (self.target.clone(), o))
1464    }
1465
1466    /// The target a destination profile delivers for: the run's, when the profile exists.
1467    fn dest_target(&self, profile: &str) -> Option<String> {
1468        self.project
1469            .profiles
1470            .get(Role::Destination, profile)
1471            .map(|_| self.target.clone())
1472    }
1473
1474    fn skip_note(&self, profile: &str) -> String {
1475        let dtarget = self.dest_target(profile).unwrap_or_default();
1476        format!(
1477            "destination profile `{profile}` has no `{dtarget}` target: not delivered, output stays in target/"
1478        )
1479    }
1480
1481    /// Deliver every file to one destination. `Ok(None)`: its profile has no output for the
1482    /// active target, so nothing was sent.
1483    fn deliver_one(&mut self, d: &RenderedDest) -> Result<Option<String>, Fail> {
1484        if self.project.profiles.get(Role::Destination, &d.profile).is_none()
1485            && !self.project.profiles.is_builtin_local(&d.profile)
1486        {
1487            return Err(format!(
1488                "destination profile `{}` isn't in profiles.yml",
1489                d.profile
1490            ));
1491        }
1492        let Some((_, out)) = self.dest_output(&d.profile) else {
1493            let note = self.skip_note(&d.profile);
1494            self.ui.step(Level::Info, "Kept", &note, None);
1495            return Ok(None);
1496        };
1497        let kind = out.kind.clone();
1498        let connection = render_connection(out)?;
1499        let targets: Vec<DeliveryFile> = self
1500            .files
1501            .iter()
1502            .map(|(f, _)| {
1503                let name = f.file_name().unwrap().to_string_lossy().to_string();
1504                let r = d.path.as_deref().map(|r| {
1505                    if self.files.len() == 1 {
1506                        r.to_string()
1507                    } else {
1508                        match r.rfind(['/', '\\']) {
1509                            Some(i) => format!("{}{name}", &r[..=i]),
1510                            None => name.clone(),
1511                        }
1512                    }
1513                });
1514                DeliveryFile {
1515                    local_path: f.to_string_lossy().to_string(),
1516                    remote_path: r,
1517                }
1518            })
1519            .collect();
1520        let mut locations = Vec::new();
1521        if kind == LOCAL_TYPE {
1522            if let Some(k) = d.options.keys().next() {
1523                return Err(format!(
1524                    "the local destination takes no options, but `{}` has `{k}`; check the key's spelling",
1525                    d.profile
1526                ));
1527            }
1528            for (i, f) in targets.iter().enumerate() {
1529                let t = Instant::now();
1530                let r = f
1531                    .remote_path
1532                    .as_ref()
1533                    .ok_or("the local destination needs `output.destination.path`")?;
1534                let dst = self.project.root.join(r);
1535                if let Some(parent) = dst.parent() {
1536                    std::fs::create_dir_all(parent)
1537                        .map_err(|e| format!("can't create {}: {e}", parent.display()))?;
1538                }
1539                std::fs::copy(&f.local_path, &dst).map_err(|e| {
1540                    format!(
1541                        "delivery to {} failed: {e}; the output is still in target/",
1542                        dst.display()
1543                    )
1544                })?;
1545                let loc = dst.to_string_lossy().to_string();
1546                self.ui.step(Level::Debug, "Delivered", &loc, Some(t.elapsed()));
1547                self.files[i].1.get_or_insert_with(|| loc.clone());
1548                locations.push(loc);
1549            }
1550            return Ok(Some(locations.join(", ")));
1551        }
1552        let failed = |e: dre_protocol::host::HostError| {
1553            format!("delivery through `{kind}` failed: {e}; the output is still in target/")
1554        };
1555        let plugin = find_plugin(self.project, PluginKind::Destination, &kind)?;
1556        let mut p = plugin
1557            .start(self.ui.plugin_log(), Some(&self.project.root))
1558            .map_err(|e| e.to_string())?;
1559        let batches: Vec<Vec<usize>> = if targets.len() > 1 && p.has(CAP_MULTI_FILE) {
1560            vec![(0..targets.len()).collect()]
1561        } else {
1562            (0..targets.len()).map(|i| vec![i]).collect()
1563        };
1564        for batch in batches {
1565            let t = Instant::now();
1566            let files: Vec<DeliveryFile> = batch.iter().map(|&i| targets[i].clone()).collect();
1567            let loc = p
1568                .deliver_files(&files, connection.clone(), d.options.clone())
1569                .map_err(failed)?;
1570            self.ui.step(Level::Debug, "Delivered", &loc, Some(t.elapsed()));
1571            for i in batch {
1572                self.files[i].1.get_or_insert_with(|| loc.clone());
1573            }
1574            locations.push(loc);
1575        }
1576        let _ = p.close();
1577        Ok(Some(locations.join(", ")))
1578    }
1579
1580    fn snapshot(&self) -> Json {
1581        let sets: Vec<Json> = self
1582            .produced
1583            .iter()
1584            .map(|p| {
1585                json!({
1586                    "name": p.name,
1587                    "columns": p.schema.fields().iter().map(|f| json!({
1588                        "name": f.name(),
1589                        "type": f.data_type().to_string(),
1590                        "type_shape": crate::schema::type_shape(f.data_type()),
1591                    })).collect::<Vec<_>>(),
1592                })
1593            })
1594            .collect();
1595        json!({ "result_sets": sets })
1596    }
1597
1598    fn schema_drift(&mut self) -> Vec<String> {
1599        let path = self.schema_dir.join("last_success.json");
1600        let mut current = self.snapshot();
1601        let Ok(prev) = std::fs::read_to_string(&path) else {
1602            crate::schema::redact(&mut current);
1603            self.protected_snapshot = Some(current);
1604            return Vec::new();
1605        };
1606        let Ok(mut prev) = serde_json::from_str::<Json>(&prev) else {
1607            crate::schema::redact(&mut current);
1608            self.protected_snapshot = Some(current);
1609            return Vec::new();
1610        };
1611        let original = prev.clone();
1612        crate::schema::redact(&mut prev);
1613        crate::schema::redact(&mut current);
1614        if crate::secrets::enabled() {
1615            crate::schema::align_protection(&mut prev, &mut current);
1616        }
1617        if prev != original {
1618            let protected =
1619                serde_json::to_string_pretty(&prev).expect("serializing a serde_json::Value cannot fail");
1620            if let Err(error) = std::fs::write(&path, protected) {
1621                self.ui.warn(&format!(
1622                    "can't protect the existing schema snapshot at {}: {error}; continuing without rewriting it",
1623                    path.display()
1624                ));
1625            }
1626        }
1627        let drift = crate::schema::drift(&prev, &current);
1628        self.protected_snapshot = Some(current);
1629        drift
1630    }
1631
1632    fn write_snapshot(&self) -> std::io::Result<()> {
1633        std::fs::create_dir_all(&self.schema_dir)?;
1634        let snapshot = self.protected_snapshot.clone().unwrap_or_else(|| {
1635            let mut snapshot = self.snapshot();
1636            crate::schema::redact(&mut snapshot);
1637            snapshot
1638        });
1639        std::fs::write(
1640            self.schema_dir.join("last_success.json"),
1641            serde_json::to_string_pretty(&snapshot)?,
1642        )
1643    }
1644
1645    fn write_results(&self, status: &Status, error: Option<&str>) -> std::io::Result<()> {
1646        std::fs::create_dir_all(&self.run_dir)?;
1647        let outputs: Vec<Json> = self
1648            .files
1649            .iter()
1650            .map(|(f, delivered)| {
1651                json!({
1652                    "path": record_path(self.project, f),
1653                    "size": std::fs::metadata(f).map(|m| m.len()).unwrap_or(0),
1654                    "delivered_to": delivered,
1655                })
1656            })
1657            .collect();
1658        let results = json!({
1659            "report": self.report.name,
1660            "set": self.b.set,
1661            "binding": self.b.dir_name(),
1662            "managed": self.report.managed,
1663            "profile": self.parsed.inherited,
1664            "connections": self.parsed.connections(),
1665            "target": self.target,
1666            "schedule": self.opts.schedule,
1667            "schedule_vars": self.schedule_vars,
1668            "vars": self.rendered_vars,
1669            "run_date": self.date.to_string(),
1670            "scheduled_at": self.opts.scheduled_at.map(rfc3339),
1671            "timezone": self.calendar.tz.name(),
1672            "params": self.opts.params(self.date),
1673            "status": status,
1674            "error": error,
1675            "preview": self.opts.preview.is_some(),
1676            "row_limit": self.opts.preview,
1677            "started_at": self.started_at.to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
1678            "duration_ms": self.started.elapsed().as_millis() as u64,
1679            "result_sets": self.produced.iter().map(|p| json!({
1680                "name": p.name,
1681                "query": p.query,
1682                "connection": p.connection,
1683                "rows": p.rows,
1684                "columns": p.schema.fields().iter().map(|f| f.name().clone()).collect::<Vec<_>>(),
1685            })).collect::<Vec<_>>(),
1686            "outputs": outputs,
1687            "delivery": self.delivery_note,
1688            "deliveries": self.deliveries,
1689            "schema_drift": self.drift,
1690            "target_path": self.project.target_dir,
1691            "manifest_checksum": self.opts.manifest_checksum,
1692        });
1693        std::fs::write(
1694            self.run_dir.join("run_results.json"),
1695            (crate::secrets::to_json_pretty(&results)? + "\n").as_bytes(),
1696        )
1697    }
1698
1699    /// `dre validate --live`: execute temp creates, `check` everything else, each on its
1700    /// connection; then check declared source columns against the database.
1701    fn live_check(&mut self, statements: &[Statement]) -> Result<(), Fail> {
1702        let pool = self.pool.clone().expect("pool");
1703        let mut failures = Vec::new();
1704        let mut unexecuted_setup: BTreeMap<String, (PathBuf, usize)> = BTreeMap::new();
1705        let mut uncheckable: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
1706        let sql_log = self.ui.sql_log();
1707        for st in statements {
1708            if uncheckable.contains(&st.connection) {
1709                continue;
1710            }
1711            let session = pool.session(&st.connection)?;
1712            let mut s = session.lock().unwrap();
1713            if !s.get()?.has(dre_protocol::CAP_CHECK) {
1714                let kind = pool.kind(&st.connection).unwrap_or_default();
1715                self.ui.warn(&format!(
1716                    "  not checkable: the `{kind}` source plugin (connection `{}`) can't check statements without running them",
1717                    st.connection
1718                ));
1719                uncheckable.insert(st.connection.clone());
1720                continue;
1721            }
1722            let verb = if st.kind == StatementKind::TempCreate {
1723                ""
1724            } else {
1725                " (check)"
1726            };
1727            sql_log(
1728                &format!("{}:{}{verb}", st.file.display(), st.line),
1729                &protected_sql(&st.text, st.sensitive),
1730            );
1731            if st.kind == StatementKind::TempCreate {
1732                let result = {
1733                    let (process, _log_scope) = s.statement(st.sensitive)?;
1734                    process.execute(&st.text, None, |_, _| Ok(()))
1735                };
1736                if let Err(e) = result {
1737                    failures.push(format!(
1738                        "{}:{}: {}",
1739                        st.file.display(),
1740                        st.line,
1741                        masked_source_error(&e, st.sensitive)
1742                    ));
1743                }
1744                continue;
1745            }
1746            let result = {
1747                let (process, _log_scope) = s.statement(st.sensitive)?;
1748                process.check(&st.text)
1749            };
1750            if let Err(e) = result {
1751                let mut msg = format!(
1752                    "{}:{}: {}",
1753                    st.file.display(),
1754                    st.line,
1755                    masked_source_error(&e, st.sensitive)
1756                );
1757                if let Some((f, l)) = unexecuted_setup.get(&st.connection) {
1758                    msg.push_str(&format!(
1759                        " (may be a false positive: the setup statement at {}:{l} wasn't executed during the check)",
1760                        f.display()
1761                    ));
1762                }
1763                failures.push(msg);
1764            }
1765            if st.kind == StatementKind::Other {
1766                unexecuted_setup.insert(st.connection.clone(), (st.file.clone(), st.line));
1767            }
1768        }
1769        self.check_source_columns(&pool, &mut failures);
1770        if failures.is_empty() {
1771            Ok(())
1772        } else {
1773            Err(failures.join("\n    "))
1774        }
1775    }
1776
1777    /// Declared source `columns` against the real table, once per source table, connection and
1778    /// target in this process: each must exist, and match its `data_type` (loosely) where given.
1779    fn check_source_columns(&mut self, pool: &Arc<Pool>, failures: &mut Vec<String>) {
1780        /// `(project, source.table, connection, target)`.
1781        type Checked = std::collections::BTreeSet<(PathBuf, String, String, String)>;
1782        static CHECKED: std::sync::LazyLock<Mutex<Checked>> = std::sync::LazyLock::new(Mutex::default);
1783        for q in self.parsed.queries.clone() {
1784            let Some(conn) = q.connection.clone() else {
1785                continue;
1786            };
1787            for key in &q.sources {
1788                let Some(r) = self.parsed.sources.get(key).cloned() else {
1789                    continue;
1790                };
1791                let Some(declared) = self
1792                    .project
1793                    .sources
1794                    .get(&r.source)
1795                    .and_then(|s| s.table(&r.table))
1796                    .map(|t| t.columns.clone())
1797                    .filter(|c| !c.is_empty())
1798                else {
1799                    continue;
1800                };
1801                let id = (
1802                    self.project.root.clone(),
1803                    key.clone(),
1804                    conn.clone(),
1805                    self.target.clone(),
1806                );
1807                if !CHECKED.lock().unwrap().insert(id) {
1808                    continue;
1809                }
1810                let quote = if r.quoting.any() {
1811                    let kind = pool.kind(&conn).unwrap_or_default();
1812                    match pool.connections.identifier_quote(&kind) {
1813                        Ok(q) => q,
1814                        Err(e) => {
1815                            failures.push(format!("source `{key}`: can't quote its name: {e}"));
1816                            continue;
1817                        }
1818                    }
1819                } else {
1820                    None
1821                };
1822                let rel = r.relation(quote.as_deref());
1823                let runner = PoolRunner {
1824                    pool: pool.clone(),
1825                    default: Some(conn.clone()),
1826                    log: self.ui.sql_log(),
1827                };
1828                let actual = match runner.columns(&format!("select * from {rel} where 1=0"), None) {
1829                    Ok(c) => c,
1830                    Err(e) => {
1831                        failures.push(format!("source `{key}` ({rel} on connection `{conn}`): {e}"));
1832                        continue;
1833                    }
1834                };
1835                for c in &declared {
1836                    let Some(a) = actual.iter().find(|a| a.name.eq_ignore_ascii_case(&c.name)) else {
1837                        failures.push(format!(
1838                            "source `{key}`: declared column `{}` isn't in {rel} on connection `{conn}` (it has: {})",
1839                            c.name,
1840                            actual.iter().map(|a| a.name.as_str()).collect::<Vec<_>>().join(", ")
1841                        ));
1842                        continue;
1843                    };
1844                    let Some(dt) = &c.data_type else { continue };
1845                    match crate::coltypes::matches(dt, &a.data_type) {
1846                        Some(true) => {}
1847                        Some(false) => failures.push(format!(
1848                            "source `{key}`: column `{}` is declared `{dt}`, but {rel} on connection `{conn}` returns {}",
1849                            c.name, a.data_type
1850                        )),
1851                        None => self.ui.warn(&format!(
1852                            "  source `{key}`: column `{}`'s `data_type: {dt}` isn't a type DRE can compare, so only its presence was checked",
1853                            c.name
1854                        )),
1855                    }
1856                }
1857            }
1858        }
1859    }
1860}
1861
1862/// The source session, started on first use (rendering may need it for `run_query()`).
1863struct Session {
1864    path: Result<crate::plugins::Located, String>,
1865    connection: Result<JsonMap<String, Json>, String>,
1866    cwd: PathBuf,
1867    log: LogSink,
1868    /// Whether plugin diagnostics for the active statement must be hidden rather than exactly
1869    /// masked. Connection diagnostics always use exact-value masking.
1870    opaque_log: Arc<AtomicBool>,
1871    /// Unmanaged reports run read-only unless they create temp objects.
1872    unmanaged: bool,
1873    read_only: bool,
1874    /// A lookup was loaded into a temp table, so the session must stay open as it is.
1875    loaded: bool,
1876    proc_: Option<PluginProcess>,
1877}
1878
1879impl Session {
1880    fn new(
1881        path: Result<crate::plugins::Located, String>,
1882        connection: Result<JsonMap<String, Json>, String>,
1883        cwd: PathBuf,
1884        log: LogSink,
1885        unmanaged: bool,
1886    ) -> Session {
1887        let opaque_log = Arc::new(AtomicBool::new(false));
1888        let protect = Arc::clone(&opaque_log);
1889        let downstream = log;
1890        let log: LogSink = Arc::new(move |plugin, line| {
1891            let line = crate::secrets::mask_source_error(line, protect.load(Ordering::SeqCst));
1892            downstream(plugin, &line);
1893        });
1894        Session {
1895            path,
1896            connection,
1897            cwd,
1898            log,
1899            opaque_log,
1900            unmanaged,
1901            read_only: unmanaged,
1902            loaded: false,
1903            proc_: None,
1904        }
1905    }
1906
1907    fn want_read_only(&mut self, ro: bool) {
1908        if self.read_only != ro {
1909            // Reopen if a session was already started in the other mode.
1910            self.close();
1911            self.read_only = ro;
1912        }
1913    }
1914
1915    fn get(&mut self) -> Result<&mut PluginProcess, String> {
1916        if self.proc_.is_none() {
1917            let plugin = self.path.clone()?;
1918            let connection = self.connection.clone()?;
1919            let mut p = plugin
1920                .start(self.log.clone(), Some(&self.cwd))
1921                .map_err(|e| masked_source_error(&e, false))?;
1922            let ro = self.unmanaged && self.read_only && p.has(CAP_READ_ONLY);
1923            p.open(connection, ro).map_err(|e| {
1924                let protected = masked_source_error(&e, false);
1925                format!("can't open the source connection: {protected}")
1926            })?;
1927            self.proc_ = Some(p);
1928        }
1929        Ok(self.proc_.as_mut().unwrap())
1930    }
1931
1932    fn statement(&mut self, sensitive: bool) -> Result<(&mut PluginProcess, OpaqueLogScope), String> {
1933        let opaque_log = Arc::clone(&self.opaque_log);
1934        let process = self.get()?;
1935        let scope = OpaqueLogScope::new(opaque_log, sensitive);
1936        Ok((process, scope))
1937    }
1938
1939    fn close(&mut self) {
1940        if let Some(p) = self.proc_.take() {
1941            let _ = p.close();
1942        }
1943    }
1944}
1945
1946struct OpaqueLogScope(Arc<AtomicBool>);
1947
1948impl OpaqueLogScope {
1949    fn new(active: Arc<AtomicBool>, sensitive: bool) -> Self {
1950        active.store(sensitive, Ordering::SeqCst);
1951        Self(active)
1952    }
1953}
1954
1955impl Drop for OpaqueLogScope {
1956    fn drop(&mut self) {
1957        self.0.store(false, Ordering::SeqCst);
1958    }
1959}
1960
1961/// The Binding's connections: each one's session, opened when a query first needs it and held
1962/// until the report ends.
1963struct Pool {
1964    connections: Arc<ProfileConnections>,
1965    cwd: PathBuf,
1966    log: LogSink,
1967    unmanaged: bool,
1968    sessions: Mutex<BTreeMap<String, Arc<Mutex<Session>>>>,
1969}
1970
1971impl Pool {
1972    fn new(connections: Arc<ProfileConnections>, cwd: PathBuf, log: LogSink, unmanaged: bool) -> Pool {
1973        Pool {
1974            connections,
1975            cwd,
1976            log,
1977            unmanaged,
1978            sessions: Mutex::default(),
1979        }
1980    }
1981
1982    /// A connection's settings for the run's target.
1983    fn target(&self, name: &str) -> Result<ProfileTarget, String> {
1984        let profiles = &self.connections.profiles;
1985        let target = &self.connections.target;
1986        let p = profiles.get(Role::Connection, name).ok_or_else(|| {
1987            format!(
1988                "connection `{name}` isn't under `{}:` in {}",
1989                profiles.section_key(Role::Connection),
1990                profiles.path.display()
1991            )
1992        })?;
1993        p.targets.get(target).cloned().ok_or_else(|| {
1994            format!(
1995                "connection `{name}` has no `{target}` target (it has: {})",
1996                p.targets.keys().cloned().collect::<Vec<_>>().join(", ")
1997            )
1998        })
1999    }
2000
2001    /// A connection's plugin type.
2002    fn kind(&self, name: &str) -> Result<String, String> {
2003        Ok(self.target(name)?.kind)
2004    }
2005
2006    fn session(&self, name: &str) -> Result<Arc<Mutex<Session>>, String> {
2007        if let Some(s) = self.sessions.lock().unwrap().get(name) {
2008            return Ok(s.clone());
2009        }
2010        let output = self.target(name)?;
2011        let id = crate::project::PluginId::new(PluginKind::Source, output.kind.clone());
2012        let path = crate::plugins::locate_in(&self.connections.root, &self.connections.plugins, &id)
2013            .map_err(String::from);
2014        let session = Arc::new(Mutex::new(Session::new(
2015            path,
2016            render_connection(&output),
2017            self.cwd.clone(),
2018            self.log.clone(),
2019            self.unmanaged,
2020        )));
2021        self.sessions
2022            .lock()
2023            .unwrap()
2024            .insert(name.to_string(), session.clone());
2025        Ok(session)
2026    }
2027
2028    fn close_all(&self) {
2029        for s in self.sessions.lock().unwrap().values() {
2030            if let Ok(mut s) = s.lock() {
2031                s.close();
2032            }
2033        }
2034    }
2035}
2036
2037/// `run_query()`, `columns()` and lookups for one renderer: on the named connection's session,
2038/// else on the renderer's own (`default`).
2039struct PoolRunner {
2040    pool: Arc<Pool>,
2041    default: Option<String>,
2042    log: LogSink,
2043}
2044
2045impl PoolRunner {
2046    fn session(&self, profile: Option<&str>) -> Result<Arc<Mutex<Session>>, String> {
2047        let name = profile.or(self.default.as_deref()).ok_or(
2048            "no connection here: pass `profile=`, or give the report a `profile:` (this isn't a query's SQL)",
2049        )?;
2050        self.pool.session(name)
2051    }
2052}
2053
2054impl QueryRunner for PoolRunner {
2055    fn run_query(&self, sql: &str, max_rows: u64, profile: Option<&str>) -> Result<QueryRows, String> {
2056        let session = self.session(profile)?;
2057        let mut s = session.lock().map_err(|_| "session lock poisoned".to_string())?;
2058        let sensitive = crate::secrets::contains_secret(sql);
2059        // An unmanaged report may only read, and run_query() runs before the file's own
2060        // statements are checked, so it gets the same rule up front.
2061        if s.unmanaged
2062            && let Some(bad) = sqlsplit::split(sql)
2063                .into_iter()
2064                .find(|st| sqlsplit::classify(&st.text) != StatementKind::Read)
2065        {
2066            return Err(format!(
2067                "run_query() in an unmanaged report may only read, but got `{}`; give the report a YAML to declare it",
2068                masked_sql_summary(&bad.text, 60, sensitive)
2069            ));
2070        }
2071        (self.log)("run_query()", &protected_sql(sql, sensitive));
2072        let (p, _log_scope) = s.statement(sensitive)?;
2073        let mut out = QueryRows::default();
2074        let mut too_many = false;
2075        let exec = p
2076            .execute(sql, Some(max_rows + 1), |schema, batch| {
2077                if out.columns.is_empty() {
2078                    out.columns = schema.fields().iter().map(|f| f.name().clone()).collect();
2079                }
2080                if out.rows.len() as u64 + batch.num_rows() as u64 > max_rows {
2081                    too_many = true;
2082                    return Ok(());
2083                }
2084                out.rows.extend(crate::values::batch_rows(&batch));
2085                Ok(())
2086            })
2087            .map_err(|e| format!("run_query() failed: {}", masked_source_error(&e, sensitive)))?;
2088        if too_many {
2089            return Err(format!(
2090                "run_query() returned more than {max_rows} rows; it's meant for small lookups — raise the cap with `run_query(sql, max_rows=N)` or `run_query_max_rows` in dre_project.yml"
2091            ));
2092        }
2093        if let Execution::Result { schema, .. } = exec
2094            && out.columns.is_empty()
2095        {
2096            out.columns = schema.fields().iter().map(|f| f.name().clone()).collect();
2097        }
2098        Ok(out)
2099    }
2100
2101    fn columns(&self, sql: &str, profile: Option<&str>) -> Result<Vec<Column>, String> {
2102        let session = self.session(profile)?;
2103        let mut s = session.lock().map_err(|_| "session lock poisoned".to_string())?;
2104        let sensitive = crate::secrets::contains_secret(sql);
2105        (self.log)("columns()", &protected_sql(sql, sensitive));
2106        let (p, _log_scope) = s.statement(sensitive)?;
2107        let mut schema: Option<SchemaRef> = None;
2108        let exec = p
2109            .execute(sql, Some(1), |sch, _| {
2110                schema.get_or_insert_with(|| sch.clone());
2111                Ok(())
2112            })
2113            .map_err(|e| masked_source_error(&e, sensitive))?;
2114        if let Execution::Result { schema: sch, .. } = exec {
2115            schema.get_or_insert(sch);
2116        }
2117        let schema = schema.ok_or("the query returned no result set")?;
2118        Ok(schema
2119            .fields()
2120            .iter()
2121            .map(|f| Column {
2122                name: f.name().clone(),
2123                data_type: f.data_type().to_string(),
2124            })
2125            .collect())
2126    }
2127
2128    fn load(&self, name: &str, table: &Table) -> Result<Option<(String, Option<String>)>, String> {
2129        let session = self.session(None)?;
2130        let mut s = session.lock().map_err(|_| "session lock poisoned".to_string())?;
2131        // A temp table is allowed even in an unmanaged report; it needs a writable session.
2132        s.want_read_only(false);
2133        if !s.get()?.has(CAP_LOAD) {
2134            return Ok(None);
2135        }
2136        let batch = table.to_batch()?;
2137        let schema = batch.schema();
2138        (self.log)(
2139            &format!("lookup `{name}`"),
2140            &format!(
2141                "-- {} rows loaded through the plugin's `load` request",
2142                batch.num_rows()
2143            ),
2144        );
2145        let r = s.get()?.load(name, &schema, [batch]).map_err(|e| e.to_string())?;
2146        s.loaded = true;
2147        Ok(Some(r))
2148    }
2149}
2150
2151/// Render every string inside a destination option value.
2152fn render_json(renderer: &Renderer, file: &Path, v: &Json) -> Result<Json, RenderError> {
2153    Ok(match v {
2154        Json::String(s) => Json::String(renderer.render(file, s)?),
2155        Json::Array(a) => Json::Array(
2156            a.iter()
2157                .map(|v| render_json(renderer, file, v))
2158                .collect::<Result<_, _>>()?,
2159        ),
2160        Json::Object(o) => Json::Object(
2161            o.iter()
2162                .map(|(k, v)| Ok((k.clone(), render_json(renderer, file, v)?)))
2163                .collect::<Result<_, RenderError>>()?,
2164        ),
2165        other => other.clone(),
2166    })
2167}
2168
2169/// Render `env_var()` (and only that) inside a profile output's string fields.
2170fn render_connection(output: &ProfileTarget) -> Result<JsonMap<String, Json>, String> {
2171    let mut env = minijinja::Environment::new();
2172    env.set_undefined_behavior(minijinja::UndefinedBehavior::Strict);
2173    env.add_function("env_var", |name: String, default: Option<String>| -> Result<String, minijinja::Error> {
2174        std::env::var(&name).ok().or(default).ok_or_else(|| {
2175            minijinja::Error::new(
2176                minijinja::ErrorKind::UndefinedError,
2177                format!("`env_var('{name}')`: environment variable `{name}` is not set and no default is given"),
2178            )
2179        })
2180    });
2181    fn walk(env: &minijinja::Environment<'_>, v: &Json) -> Result<Json, String> {
2182        Ok(match v {
2183            Json::String(s) if crate::preflight::is_templated(s) => Json::String(
2184                env.render_str(s, ())
2185                    .map_err(|e| format!("profiles.yml: {}", e.detail().unwrap_or("render error")))?,
2186            ),
2187            Json::Array(a) => Json::Array(a.iter().map(|x| walk(env, x)).collect::<Result<_, _>>()?),
2188            Json::Object(o) => Json::Object(
2189                o.iter()
2190                    .map(|(k, x)| Ok((k.clone(), walk(env, x)?)))
2191                    .collect::<Result<_, String>>()?,
2192            ),
2193            other => other.clone(),
2194        })
2195    }
2196    match walk(&env, &Json::Object(output.fields.clone()))? {
2197        Json::Object(o) => Ok(o),
2198        _ => unreachable!(),
2199    }
2200}
2201
2202/// Locate the plugin for a declared type among the project's packages, honouring `dre.lock`
2203/// pins and declared constraints.
2204pub fn find_plugin(
2205    project: &Project,
2206    kind: PluginKind,
2207    name: &str,
2208) -> Result<crate::plugins::Located, crate::plugins::LocateError> {
2209    crate::plugins::locate(project, &crate::project::PluginId::new(kind, name))
2210}
2211
2212/// `connection`, `destination` and `profile()` from `profiles.yml`, for the run's target. A
2213/// field is secret when the plugin's `describe` says so; when the plugin can't be asked, when its
2214/// name looks like one; and always when its value comes from a `DRE_SECRET_*` variable.
2215struct ProfileConnections {
2216    profiles: Profiles,
2217    /// The run's target.
2218    target: String,
2219    root: PathBuf,
2220    plugins: Vec<crate::project::PluginRequirement>,
2221    log: LogSink,
2222}
2223
2224/// What `describe` said, by `(project, role, type)`, `None` when it couldn't be asked: each
2225/// plugin is asked once per process, not once per Binding.
2226static DESCRIBED: std::sync::LazyLock<Mutex<BTreeMap<DescribeKey, Option<Described>>>> =
2227    std::sync::LazyLock::new(Mutex::default);
2228
2229/// `(project root, role, plugin type)`.
2230type DescribeKey = (PathBuf, Role, String);
2231
2232#[derive(Clone)]
2233struct Described {
2234    secrets: Vec<String>,
2235    identifier_quote: Option<String>,
2236}
2237
2238const SECRET_NAME_WORDS: &[&str] = &["password", "secret", "token", "key", "credential"];
2239
2240impl ProfileConnections {
2241    fn view(&self, role: Role, name: &str) -> Result<Connection, String> {
2242        let (target, out): (String, ProfileTarget) = match self.profiles.get(role, name) {
2243            Some(p) => {
2244                let t = &self.target;
2245                let o = p.targets.get(t).ok_or_else(|| {
2246                    format!(
2247                        "{} `{name}` has no `{t}` target (it has: {})",
2248                        role.as_str(),
2249                        p.targets.keys().cloned().collect::<Vec<_>>().join(", ")
2250                    )
2251                })?;
2252                (t.clone(), o.clone())
2253            }
2254            None if role == Role::Destination && self.profiles.is_builtin_local(name) => {
2255                (self.target.clone(), BUILTIN_LOCAL.clone())
2256            }
2257            None => {
2258                return Err(format!(
2259                    "no {} profile `{name}` in {}",
2260                    role.as_str(),
2261                    self.profiles.path.display()
2262                ));
2263            }
2264        };
2265        let mut secrets: Vec<String> = out
2266            .fields
2267            .iter()
2268            .filter(|(_, v)| v.as_str().is_some_and(|s| s.contains(crate::secrets::PREFIX)))
2269            .map(|(k, _)| k.clone())
2270            .collect();
2271        match self.described(role, &out.kind) {
2272            Some(d) => secrets.extend(d.secrets),
2273            None => secrets.extend(
2274                out.fields
2275                    .keys()
2276                    .filter(|k| {
2277                        let k = k.to_lowercase();
2278                        SECRET_NAME_WORDS.iter().any(|w| k.contains(w))
2279                    })
2280                    .cloned(),
2281            ),
2282        }
2283        let fields = render_connection(&out)?;
2284        Ok(Connection {
2285            profile: name.to_string(),
2286            target,
2287            kind: out.kind.clone(),
2288            fields,
2289            secrets,
2290        })
2291    }
2292
2293    /// The plugin's `describe` reply, asked once per plugin type.
2294    fn described(&self, role: Role, kind: &str) -> Option<Described> {
2295        let key = (self.root.clone(), role, kind.to_string());
2296        if let Some(v) = DESCRIBED.lock().unwrap().get(&key) {
2297            return v.clone();
2298        }
2299        let plugin_kind = match role {
2300            Role::Connection => PluginKind::Source,
2301            Role::Destination => PluginKind::Destination,
2302        };
2303        let asked = (|| {
2304            if kind == LOCAL_TYPE {
2305                return Some(Described {
2306                    secrets: Vec::new(),
2307                    identifier_quote: None,
2308                });
2309            }
2310            let id = crate::project::PluginId::new(plugin_kind, kind);
2311            let plugin = crate::plugins::locate_in(&self.root, &self.plugins, &id).ok()?;
2312            let mut p = plugin.start(self.log.clone(), Some(&self.root)).ok()?;
2313            let d = p.description().ok();
2314            let _ = p.close();
2315            let d = d?;
2316            Some(Described {
2317                secrets: d
2318                    .connection_fields
2319                    .into_iter()
2320                    .filter(|f| f.secret)
2321                    .map(|f| f.name)
2322                    .collect(),
2323                identifier_quote: d.identifier_quote,
2324            })
2325        })();
2326        DESCRIBED.lock().unwrap().insert(key, asked.clone());
2327        asked
2328    }
2329}
2330
2331impl Connections for ProfileConnections {
2332    fn profile(&self, name: &str, role: Option<&str>) -> Result<Connection, String> {
2333        let role = match role {
2334            Some("connection") => Role::Connection,
2335            Some(_) => Role::Destination,
2336            None => {
2337                let c = self.profiles.get(Role::Connection, name).is_some();
2338                let d = self.profiles.get(Role::Destination, name).is_some();
2339                match (c, d) {
2340                    (true, true) => {
2341                        return Err(format!(
2342                            "`{name}` is both a connection and a destination profile; say which with `profile('{name}', role='connection')` or `role='destination'`"
2343                        ));
2344                    }
2345                    (true, false) => Role::Connection,
2346                    _ => Role::Destination,
2347                }
2348            }
2349        };
2350        self.view(role, name)
2351    }
2352
2353    fn identifier_quote(&self, kind: &str) -> Result<Option<String>, String> {
2354        match self.described(Role::Connection, kind) {
2355            Some(d) => Ok(d.identifier_quote),
2356            None => Err(format!(
2357                "the `{kind}` source plugin couldn't be asked for its identifier quote character"
2358            )),
2359        }
2360    }
2361}
2362
2363fn check_sheet_name(n: &str) -> Result<(), String> {
2364    if n.is_empty() || n.chars().count() > 31 {
2365        return Err(format!("sheet name `{n}` must be 1 to 31 characters long"));
2366    }
2367    if let Some(c) = n.chars().find(|c| "[]:*?/\\".contains(*c)) {
2368        return Err(format!(
2369            "sheet name `{n}` contains `{c}`, which Excel doesn't allow in sheet names"
2370        ));
2371    }
2372    if n.starts_with('\'') || n.ends_with('\'') {
2373        return Err(format!("sheet name `{n}` can't start or end with an apostrophe"));
2374    }
2375    if n.eq_ignore_ascii_case("history") {
2376        return Err("`History` is reserved by Excel and can't be a sheet name".into());
2377    }
2378    Ok(())
2379}
2380
2381fn is_single_table(format: &str) -> bool {
2382    matches!(format, "csv" | "delimited" | "fixed_width" | "parquet")
2383}
2384
2385fn extension(format: &str) -> &str {
2386    match format {
2387        "delimited" | "fixed_width" => "txt",
2388        f => f,
2389    }
2390}
2391
2392fn split_ext(name: &str) -> (&str, &str) {
2393    match name.rfind('.') {
2394        Some(i) if i > 0 => (&name[..i], &name[i..]),
2395        _ => (name, ""),
2396    }
2397}
2398
2399fn rel(root: &Path, p: &Path) -> PathBuf {
2400    crate::slash(p.strip_prefix(root).unwrap_or(p))
2401}
2402
2403/// An instant as DRE records it: RFC 3339 in UTC, to the second.
2404pub fn rfc3339(t: chrono::DateTime<chrono::Utc>) -> String {
2405    t.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
2406}
2407
2408/// How a generated file is recorded in `run_results.json`: relative to the project root when
2409/// it's inside it, else relative to the target path, so moving the target folder doesn't
2410/// change what's recorded.
2411fn record_path(project: &Project, p: &Path) -> PathBuf {
2412    match p.strip_prefix(&project.root) {
2413        Ok(r) => crate::slash(r),
2414        Err(_) => rel(&project.target_dir, p),
2415    }
2416}
2417
2418impl BindingRun<'_> {
2419    /// What this Binding produced, for its end-of-run line.
2420    fn summary_line(&self) -> String {
2421        let rows: u64 = self.produced.iter().map(|p| p.rows).sum();
2422        let mut parts = Vec::new();
2423        if !self.produced.is_empty() {
2424            let n = self.produced.len();
2425            parts.push(format!(
2426                "{n} result set{}, {} row{}",
2427                if n == 1 { "" } else { "s" },
2428                thousands(rows),
2429                if rows == 1 { "" } else { "s" }
2430            ));
2431        }
2432        let names: Vec<String> = self
2433            .files
2434            .iter()
2435            .map(|(f, _)| f.file_name().unwrap_or_default().to_string_lossy().to_string())
2436            .collect();
2437        if !names.is_empty() {
2438            parts.push(format!("→ {}", names.join(", ")));
2439        }
2440        // Every destination's location, not just the first one each file reached.
2441        let delivered: Vec<&str> = self
2442            .deliveries
2443            .iter()
2444            .filter_map(|d| d.get("location").and_then(Json::as_str))
2445            .collect();
2446        if !delivered.is_empty() {
2447            parts.push(format!("→ {}", delivered.join(", ")));
2448        } else if let Some(n) = &self.delivery_note
2449            && self.opts.preview.is_none()
2450            && !self.b.output.destinations.is_empty()
2451        {
2452            parts.push(format!("({n})"));
2453        }
2454        parts.join(" ")
2455    }
2456}
2457
2458/// `1234567` → `1,234,567`.
2459pub fn thousands(n: u64) -> String {
2460    let s = n.to_string();
2461    let mut out = String::new();
2462    for (i, c) in s.chars().enumerate() {
2463        if i > 0 && (s.len() - i).is_multiple_of(3) {
2464            out.push(',');
2465        }
2466        out.push(c);
2467    }
2468    out
2469}
2470
2471fn human_bytes(n: u64) -> String {
2472    match n {
2473        n if n < 1024 => format!("{n} B"),
2474        n if n < 1024 * 1024 => format!("{:.1} KB", n as f64 / 1024.0),
2475        n => format!("{:.1} MB", n as f64 / (1024.0 * 1024.0)),
2476    }
2477}
2478
2479/// `dre clean`: remove the target folder `t`.
2480pub fn clean(t: &Path) -> std::io::Result<bool> {
2481    if t.exists() {
2482        std::fs::remove_dir_all(t)?;
2483        Ok(true)
2484    } else {
2485        Ok(false)
2486    }
2487}