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