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