Skip to main content

dre_core/
run.rs

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