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