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