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