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