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