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::dates::Calendar;
26use crate::lookups::Table;
27use crate::message::QueryResult;
28use crate::parse::ParsedBinding;
29use crate::profiles::{Entry, LOCAL_TYPE, ProfileTarget, Profiles, Role};
30use crate::project::{Binding, PluginKind, Project, QueryEntry, Report};
31use crate::render::{
32 Column, Connection, Connections, Mode, QueryRows, QueryRunner, RenderError, Renderer, RendererConfig,
33};
34use crate::selector;
35use crate::sqlsplit::{self, StatementKind};
36
37#[derive(Debug, Clone, Default)]
39pub struct RunOptions {
40 pub selector: Option<String>,
41 pub set: Option<String>,
43 pub target: Option<String>,
45 pub profile: Option<String>,
47 pub vars: BTreeMap<String, String>,
48 pub output_name: Option<String>,
49 pub output_path: Option<String>,
50 pub dry_run: bool,
51 pub preview: Option<u64>,
53 pub accept_schema_change: bool,
54 pub interactive: bool,
56 pub date: Option<NaiveDate>,
58 pub scheduled_at: Option<chrono::DateTime<chrono::Utc>>,
61 pub live_check: bool,
63 pub schedule: Option<String>,
65 pub timezone: Option<String>,
67 pub manifest_checksum: Option<String>,
69}
70
71impl RunOptions {
72 pub fn params(&self, date: NaiveDate) -> Json {
74 json!({
75 "selector": self.selector,
76 "set": self.set,
77 "schedule": self.schedule,
78 "target": self.target,
79 "profile": self.profile,
80 "vars": self.vars,
81 "run_date": date.to_string(),
82 "scheduled_at": self.scheduled_at.map(rfc3339),
83 "timezone": self.timezone,
84 "output_name": self.output_name,
85 "output_path": self.output_path,
86 "dry_run": self.dry_run,
87 "preview": self.preview,
88 "accept_schema_change": self.accept_schema_change,
89 })
90 }
91}
92
93#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
95pub enum Level {
96 Debug,
97 Info,
98}
99
100pub trait Ui {
103 fn targets(&mut self, _targets: &RunTargets) {}
105 fn plan(&mut self, _bindings: usize) {}
107 fn binding_start(&mut self, _report: &str, _set: Option<&str>) {}
108 fn binding_vars(
110 &mut self,
111 _schedule: Option<&str>,
112 _schedule_vars: Option<&JsonMap<String, Json>>,
113 _vars: &JsonMap<String, Json>,
114 ) {
115 }
116 fn step(&mut self, level: Level, verb: &str, detail: &str, elapsed: Option<Duration>);
118 fn warn(&mut self, msg: &str);
119 fn binding_end(&mut self, _outcome: &BindingOutcome) {}
120 fn compiled(&mut self, _plan: &BindingPlan) {}
122 fn message(&mut self, _message: &ShownMessage) {}
124 fn choose_set(&mut self, report: &str, sets: &[String]) -> Result<Option<String>, String>;
126 fn plugin_log(&self) -> LogSink;
128 fn sql_log(&self) -> LogSink {
130 Arc::new(|_, _| {})
131 }
132}
133
134#[derive(Debug, Clone, Serialize)]
136pub struct ShownMessage {
137 pub output: Option<String>,
139 pub title: String,
140 pub text: String,
142 pub notes: Vec<String>,
144}
145
146#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
147#[serde(rename_all = "snake_case")]
148pub enum Status {
149 Success,
150 Error,
151 DryRun,
152 Checked,
153}
154
155#[derive(Debug, Clone, Serialize)]
156pub struct BindingOutcome {
157 pub report: String,
158 #[serde(skip_serializing_if = "Option::is_none")]
159 pub set: Option<String>,
160 pub binding: String,
161 pub status: Status,
162 #[serde(skip_serializing_if = "Option::is_none")]
163 pub error: Option<String>,
164 pub files: Vec<PathBuf>,
165 pub summary: String,
167 pub schedule: Option<String>,
169 pub schedule_vars: Option<JsonMap<String, Json>>,
170 pub vars: JsonMap<String, Json>,
172 #[serde(skip_serializing_if = "String::is_empty")]
174 pub timezone: String,
175 #[serde(skip)]
176 pub elapsed: Duration,
177}
178
179#[derive(Debug, Clone, Serialize)]
181pub struct BindingPlan {
182 pub report: String,
183 pub set: Option<String>,
184 pub compiled: Vec<PathBuf>,
186 pub profile: Option<String>,
188 pub queries: Vec<PlannedQuery>,
190 pub target: String,
191 pub format: String,
192 pub output: PathBuf,
195 pub destinations: Vec<PlannedDestination>,
196 pub outputs: Vec<PlannedOutput>,
198 pub schedules: Vec<String>,
199}
200
201#[derive(Debug, Clone, Serialize)]
202pub struct PlannedOutput {
203 pub name: Option<String>,
204 pub format: String,
205 pub output: PathBuf,
207 pub destinations: Vec<PlannedDestination>,
208}
209
210#[derive(Debug, Clone, Serialize)]
211pub struct PlannedQuery {
212 pub query: String,
213 pub connection: String,
214 #[serde(rename = "type")]
215 pub kind: String,
216 pub target: String,
218 #[serde(skip_serializing_if = "Vec::is_empty")]
219 pub sources: Vec<String>,
220}
221
222#[derive(Debug, Clone, Serialize)]
223pub struct PlannedDestination {
224 pub profile: String,
225 pub kind: Option<String>,
227 pub target: Option<String>,
229 pub path: Option<String>,
230 pub delivers: bool,
232}
233
234#[derive(Debug, Default)]
235pub struct RunSummary {
236 pub outcomes: Vec<BindingOutcome>,
237 pub error: Option<String>,
239 pub missing_entry: bool,
241}
242
243impl RunSummary {
244 pub fn failed(&self) -> bool {
245 self.error.is_some() || self.outcomes.iter().any(|o| o.status == Status::Error)
246 }
247}
248
249pub fn run(project: &Project, opts: &RunOptions, ui: &mut dyn Ui) -> RunSummary {
250 let mut summary = RunSummary::default();
251 if let Err(e) = crate::target::ensure(&project.target_dir) {
252 summary.error = Some(format!(
253 "can't write to the target path {} (from {}): {e}",
254 project.target_dir.display(),
255 project.target_source
256 ));
257 return summary;
258 }
259 let planned: Vec<(&Report, Binding)> = match &opts.schedule {
260 Some(name) => {
261 if !project.schedules.iter().any(|e| &e.name == name) {
262 summary.error = Some(unknown_schedule(project, name));
263 return summary;
264 }
265 match schedule_bindings(project, name, opts) {
266 Ok(p) => p,
267 Err(e) => {
268 summary.error = Some(e);
269 return summary;
270 }
271 }
272 }
273 None => {
274 let reports: Vec<&Report> = match &opts.selector {
275 None => project.reports.iter().collect(),
276 Some(s) => match selector::resolve(project, s) {
277 Ok(r) if r.is_empty() => {
278 summary.error = Some(format!("selector `{s}` matches no report"));
279 return summary;
280 }
281 Ok(r) => r,
282 Err(e) => {
283 summary.error = Some(e.to_string());
284 return summary;
285 }
286 },
287 };
288 let mut planned = Vec::new();
291 for report in reports {
292 match choose_bindings(project, report, opts, ui) {
293 Ok(bs) => planned.extend(bs.into_iter().map(|b| (report, b))),
294 Err(e) => {
295 let outcome = BindingOutcome {
296 report: report.name.clone(),
297 set: None,
298 binding: "-".into(),
299 status: Status::Error,
300 error: Some(e),
301 files: Vec::new(),
302 summary: String::new(),
303 schedule: opts.schedule.clone(),
304 schedule_vars: None,
305 vars: JsonMap::new(),
306 timezone: String::new(),
307 elapsed: Duration::ZERO,
308 };
309 ui.binding_end(&outcome);
310 summary.outcomes.push(outcome);
311 }
312 }
313 }
314 planned
315 }
316 };
317 match run_targets(project, &planned, opts) {
319 Ok(t) => ui.targets(&t),
320 Err(e) => {
321 summary.error = Some(e);
322 summary.missing_entry = true;
323 return summary;
324 }
325 }
326 ui.plan(planned.len());
327 for (report, b) in &planned {
328 ui.binding_start(&report.name, b.set.as_deref());
329 let mut r = BindingRun::new(project, report, b, opts, ui);
330 let outcome = r.run();
331 ui.binding_end(&outcome);
332 summary.outcomes.push(outcome);
333 }
334 summary
335}
336
337#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
339pub struct UsedProfile {
340 pub role: &'static str,
342 pub profile: String,
343 pub target: String,
345 pub deliver: bool,
347}
348
349#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
351pub struct RunTargets {
352 pub name: String,
354 pub from: String,
356 pub profiles: Vec<UsedProfile>,
357}
358
359impl RunTargets {
360 pub fn line(&self) -> String {
362 let mut out = format!("{} ({})", self.name, self.from);
363 for p in self.profiles.iter().filter(|p| p.target != self.name) {
364 out.push_str(&format!("; {} `{}`: {}", p.role, p.profile, p.target));
365 }
366 out
367 }
368
369 pub fn mismatch(&self) -> Option<String> {
372 let first = &self.profiles.first()?.target;
373 if first == &self.name || self.profiles.iter().any(|p| &p.target != first) {
374 return None;
375 }
376 Some(format!(
377 "every profile is on `{first}` but the run's target is `{}`: pass `--target {first}` or set `{}`",
378 self.name,
379 crate::profiles::TARGET_ENV
380 ))
381 }
382}
383
384pub fn run_targets(
388 project: &Project,
389 planned: &[(&Report, Binding)],
390 opts: &RunOptions,
391) -> Result<RunTargets, String> {
392 let profiles = &project.profiles;
393 let schedule_vars = opts
394 .schedule
395 .as_ref()
396 .and_then(|n| project.schedules.iter().find(|e| &e.name == n))
397 .map(|e| e.vars.clone());
398 let inputs = crate::parse::Inputs {
399 target: project.target_name.clone(),
400 cli_vars: opts.vars.clone(),
401 date: opts.date,
402 scheduled_at: opts.scheduled_at,
403 timezone: opts.timezone.clone(),
404 schedule: opts.schedule.clone(),
405 started_at: None,
406 };
407 let mut used: Vec<(Role, String)> = Vec::new();
408 for (report, b) in planned {
409 let mut vars = b.vars.clone();
410 vars.extend(schedule_vars.clone().unwrap_or_default());
411 let parsed = crate::parse::binding(project, report, b, &vars, &inputs);
413 let names = parsed
414 .connections()
415 .into_iter()
416 .map(|c| (Role::Connection, c.to_string()))
417 .chain(
418 parsed
419 .destinations
420 .iter()
421 .flatten()
422 .filter(|d| !profiles.is_builtin_local(d))
423 .map(|d| (Role::Destination, d.clone())),
424 );
425 for u in names {
426 if !used.contains(&u) {
427 used.push(u);
428 }
429 }
430 }
431 let mut missing = Vec::new();
432 let mut out = Vec::new();
433 for (role, name) in &used {
434 let deliver = match profiles.entry(*role, name) {
435 Entry::Use(_) => true,
436 Entry::Nowhere => false,
437 Entry::Missing => {
438 missing.push(profiles.missing_entry(*role, name));
439 continue;
440 }
441 Entry::Unknown => continue,
443 };
444 out.push(UsedProfile {
445 role: role.as_str(),
446 profile: name.clone(),
447 target: profiles.target_of(*role, name),
448 deliver,
449 });
450 }
451 match missing.len() {
452 0 => Ok(RunTargets {
453 name: project.target_name.clone(),
454 from: project.target_from.to_string(),
455 profiles: out,
456 }),
457 1 => Err(format!("{}; nothing was run", missing[0])),
458 n => Err(format!(
459 "{n} profiles this run uses have no entry for it; nothing was run:\n {}",
460 missing.join("\n ")
461 )),
462 }
463}
464
465fn schedule_bindings<'a>(
468 project: &'a Project,
469 name: &str,
470 opts: &RunOptions,
471) -> Result<Vec<(&'a Report, Binding)>, String> {
472 let reports: Option<Vec<&str>> = match &opts.selector {
473 None => None,
474 Some(s) => Some(
475 selector::resolve(project, s)
476 .map_err(|e| e.to_string())?
477 .into_iter()
478 .map(|r| r.name.as_str())
479 .collect(),
480 ),
481 };
482 let set = opts.set.as_deref().filter(|s| *s != "all");
483 let all: Vec<(&Report, &Binding)> = project
484 .reports
485 .iter()
486 .flat_map(|r| {
487 r.bindings
488 .iter()
489 .filter(|b| b.schedules.iter().any(|s| s == name))
490 .map(move |b| (r, b))
491 })
492 .collect();
493 let picked: Vec<(&Report, Binding)> = all
494 .iter()
495 .filter(|(r, _)| reports.as_ref().is_none_or(|rs| rs.contains(&r.name.as_str())))
496 .filter(|(_, b)| set.is_none() || b.set.as_deref() == set)
497 .map(|(r, b)| {
498 let mut b = (*b).clone();
499 if let Some(p) = &opts.profile {
500 b.profile = Some(p.clone());
501 b.profile_at = Some(profile_flag_at());
502 }
503 (*r, b)
504 })
505 .collect();
506 if picked.is_empty() && (reports.is_some() || set.is_some()) {
507 let mut asked = Vec::new();
508 if let Some(s) = &opts.selector {
509 asked.push(format!("`{s}`"));
510 }
511 if let Some(s) = set {
512 asked.push(format!("Set `{s}`"));
513 }
514 let runs: Vec<String> = all
515 .iter()
516 .map(|(r, b)| match &b.set {
517 Some(s) => format!("{}/{s}", r.name),
518 None => r.name.clone(),
519 })
520 .collect();
521 return Err(format!(
522 "schedule `{name}` doesn't run {}; it runs: {}",
523 asked.join(" with "),
524 if runs.is_empty() {
525 "nothing".to_string()
526 } else {
527 runs.join(", ")
528 }
529 ));
530 }
531 Ok(picked)
532}
533
534pub fn unknown_schedule(project: &Project, name: &str) -> String {
536 let names: Vec<&str> = project.schedules.iter().map(|e| e.name.as_str()).collect();
537 if names.is_empty() {
538 format!("no schedule `{name}`: this project's schedules.yml declares none")
539 } else {
540 format!("no schedule `{name}`; valid names: {}", names.join(", "))
541 }
542}
543
544pub fn choose_bindings(
546 project: &Project,
547 report: &Report,
548 opts: &RunOptions,
549 ui: &mut dyn Ui,
550) -> Result<Vec<Binding>, String> {
551 let mut out: Vec<Binding> = match opts.set.as_deref() {
552 Some("all") => report.bindings.clone(),
553 Some(name) => match report.binding(name) {
554 Some(b) => vec![b.clone()],
555 None => vec![ad_hoc(project, report, name)],
556 },
557 None if !report.has_sets => report.bindings.clone(),
558 None => {
559 let names: Vec<String> = report.bindings.iter().filter_map(|b| b.set.clone()).collect();
560 let pick = report
561 .default_set
562 .clone()
563 .or_else(|| project.default_set.clone().filter(|d| names.contains(d)))
564 .or_else(|| (names.len() == 1).then(|| names[0].clone()));
565 match pick {
566 Some(p) => report.binding(&p).cloned().into_iter().collect(),
567 None if opts.interactive => match ui.choose_set(&report.name, &names)? {
568 Some(p) => report.binding(&p).cloned().into_iter().collect(),
569 None => report.bindings.clone(),
570 },
571 None => {
572 return Err(format!(
573 "report `{}` has several Sets ({}) and no `default_set`; declare `default_set:` or pass `--set <name>` or `--set all`",
574 report.name,
575 names.join(", ")
576 ));
577 }
578 }
579 }
580 };
581 if let Some(p) = &opts.profile {
582 for b in &mut out {
583 b.profile = Some(p.clone());
584 b.profile_at = Some(profile_flag_at());
585 }
586 }
587 Ok(out)
588}
589
590fn ad_hoc(project: &Project, report: &Report, name: &str) -> Binding {
593 let mut b = report.base.clone();
594 b.set = Some(name.to_string());
595 if let Some(reg) = project.sets.get(name) {
596 if let Some(p) = ®.profile {
597 b.profile = Some(p.clone());
598 b.profile_at = Some(crate::project::ProfileAt {
599 file: reg.file.clone(),
600 line: reg.line,
601 key: format!("`profile` of Set `{name}`"),
602 });
603 }
604 b.vars.extend(reg.vars.clone());
605 }
606 b
607}
608
609fn profile_flag_at() -> crate::project::ProfileAt {
611 crate::project::ProfileAt {
612 file: PathBuf::from("--profile"),
613 line: None,
614 key: "`--profile`".into(),
615 }
616}
617
618struct Produced {
620 query: String,
621 connection: String,
622 schema: SchemaRef,
623 rows: u64,
624 spool: PathBuf,
625 name: String,
626 anchor: Option<String>,
627 header: Option<bool>,
628 columns: BTreeMap<String, ColumnOptions>,
629}
630
631struct Statement {
632 query: String,
633 connection: String,
635 file: PathBuf,
636 line: usize,
637 text: String,
638 kind: StatementKind,
639 sensitive: bool,
642 tab: bool,
645}
646
647struct BindingRun<'a> {
648 project: &'a Project,
649 report: &'a Report,
650 b: &'a Binding,
651 vars: JsonMap<String, Json>,
653 rendered_vars: JsonMap<String, Json>,
655 schedule_vars: Option<JsonMap<String, Json>>,
656 opts: &'a RunOptions,
657 date: NaiveDate,
659 calendar: Calendar,
660 ui: &'a mut dyn Ui,
661 compiled_dir: PathBuf,
662 run_dir: PathBuf,
663 schema_dir: PathBuf,
664 target: String,
666 parsed: ParsedBinding,
668 pool: Option<Arc<Pool>>,
670 renderers: BTreeMap<Option<String>, Arc<Renderer>>,
672 started: Instant,
673 started_at: chrono::DateTime<chrono::Utc>,
674 produced: Vec<Produced>,
675 outs: Vec<OutputRun>,
677 drift: Vec<String>,
678 protected_snapshot: Option<Json>,
680}
681
682#[derive(Default)]
684struct OutputRun {
685 dests: Vec<RenderedDest>,
687 files: Vec<(PathBuf, Option<String>)>,
689 deliveries: Vec<Json>,
691 delivery_note: Option<String>,
692 status: Option<OutputStatus>,
694 error: Option<String>,
695 when: Option<bool>,
697 message: Option<(String, String)>,
699}
700
701#[derive(Debug, Clone, Copy, PartialEq, Eq)]
703enum OutputStatus {
704 Delivered,
705 Kept,
707 Skipped,
709 Failed,
710}
711
712impl OutputStatus {
713 fn as_str(self) -> &'static str {
714 match self {
715 OutputStatus::Delivered => "delivered",
716 OutputStatus::Kept => "kept",
717 OutputStatus::Skipped => "skipped",
718 OutputStatus::Failed => "failed",
719 }
720 }
721}
722
723struct RenderedDest {
725 profile: String,
726 path: Option<String>,
727 options: JsonMap<String, Json>,
728 attach: Vec<String>,
730}
731
732type Fail = String;
733
734fn masked_source_error(error: &HostError, sensitive: bool) -> String {
735 let error = error.to_string();
736 crate::secrets::mask_source_error(&error, sensitive).into_owned()
737}
738
739fn masked_sql_summary(sql: &str, max: usize, sensitive: bool) -> String {
740 if sensitive {
741 crate::secrets::MASK.to_string()
742 } else {
743 dre_protocol::util::summarize(&crate::secrets::mask(sql), max)
744 }
745}
746
747fn protected_sql(sql: &str, sensitive: bool) -> std::borrow::Cow<'_, str> {
748 if sensitive {
749 std::borrow::Cow::Borrowed(crate::secrets::MASK)
750 } else {
751 crate::secrets::mask(sql)
752 }
753}
754
755impl<'a> BindingRun<'a> {
756 fn new(
757 project: &'a Project,
758 report: &'a Report,
759 b: &'a Binding,
760 opts: &'a RunOptions,
761 ui: &'a mut dyn Ui,
762 ) -> Self {
763 let t = &project.target_dir;
764 let rel = Path::new(&report.name).join(b.dir_name());
765 let schedule = opts
766 .schedule
767 .as_ref()
768 .and_then(|n| project.schedules.iter().find(|e| &e.name == n));
769 let schedule_vars = schedule.map(|e| e.vars.clone());
770 let calendar: Calendar = crate::parse::calendar(
772 project,
773 report,
774 opts.timezone.as_deref(),
775 opts.schedule.as_deref(),
776 );
777 let date = crate::parse::run_date(&calendar, opts.date, opts.scheduled_at);
778 let mut vars = b.vars.clone();
780 vars.extend(schedule_vars.clone().unwrap_or_default());
781 let rendered_vars = {
782 let mut v = vars.clone();
783 v.extend(
784 opts.vars
785 .iter()
786 .map(|(k, x)| (k.clone(), Json::String(x.clone()))),
787 );
788 v
789 };
790 ui.binding_vars(opts.schedule.as_deref(), schedule_vars.as_ref(), &rendered_vars);
791 BindingRun {
792 project,
793 report,
794 b,
795 vars,
796 rendered_vars,
797 schedule_vars,
798 opts,
799 date,
800 calendar,
801 ui,
802 compiled_dir: t.join("compiled").join(&rel),
803 run_dir: t.join("run").join(&rel),
804 schema_dir: t.join("schema").join(&rel),
805 target: project.target_name.clone(),
806 parsed: ParsedBinding::default(),
807 pool: None,
808 renderers: BTreeMap::new(),
809 started: Instant::now(),
810 started_at: chrono::Utc::now(),
811 produced: Vec::new(),
812 outs: b.outputs.iter().map(|_| OutputRun::default()).collect(),
813 drift: Vec::new(),
814 protected_snapshot: None,
815 }
816 }
817
818 fn connections(&self) -> Arc<ProfileConnections> {
820 Arc::new(ProfileConnections {
821 profiles: self.project.profiles.clone(),
822 root: self.project.root.clone(),
823 plugins: self.project.plugins.clone(),
824 log: self.ui.plugin_log(),
825 })
826 }
827
828 fn renderer(&mut self, connection: Option<&str>) -> Result<Arc<Renderer>, Fail> {
831 let key = connection.map(str::to_string);
832 if let Some(r) = self.renderers.get(&key) {
833 return Ok(r.clone());
834 }
835 let pool = self.pool.clone().expect("the pool is made before rendering");
836 let name = key.clone().or_else(|| self.parsed.inherited.clone());
838 let source_type = match (&key, &name) {
839 (Some(n), _) => pool.kind(n)?,
840 (None, Some(n)) => pool.kind(n).unwrap_or_default(),
842 (None, None) => String::new(),
843 };
844 let context =
845 crate::parse::run_context(self.project, self.report, self.b, &self.inputs(), self.started_at);
846 let limited = Arc::new(crate::render::Limited::new(
847 context.clone(),
848 self.vars.clone(),
849 self.opts.vars.clone(),
850 ));
851 let r = Renderer::new(RendererConfig {
852 root: &self.project.root,
853 macros: &self.project.macros,
854 context,
855 vars: self.vars.clone(),
856 cli_vars: self.opts.vars.clone(),
857 runner: Some(Arc::new(PoolRunner {
858 pool: pool.clone(),
859 default: name.clone(),
860 log: self.ui.sql_log(),
861 })),
862 connections: Some(pool.connections.clone()),
863 mode: Mode::Run,
864 connection: name,
865 source_type,
866 sources: crate::parse::resolver(self.project, limited),
867 run_query_max_rows: self.project.run_query_max_rows,
868 sql: self.project.sql.clone(),
869 lookups: self.project.lookups.clone(),
870 lookup_inline_max_rows: self.project.lookup_inline_max_rows,
871 packages: self.project.packages.clone(),
872 project_name: self.project.name.clone(),
873 dispatch: self.project.dispatch.clone(),
874 })
875 .map_err(|e| e.to_string())?;
876 let r = Arc::new(r);
877 self.renderers.insert(key, r.clone());
878 Ok(r)
879 }
880
881 fn inputs(&self) -> crate::parse::Inputs {
883 crate::parse::Inputs {
884 target: self.target.clone(),
885 cli_vars: self.opts.vars.clone(),
886 date: self.opts.date,
887 scheduled_at: self.opts.scheduled_at,
888 timezone: self.opts.timezone.clone(),
889 schedule: self.opts.schedule.clone(),
890 started_at: Some(self.started_at),
891 }
892 }
893
894 fn connection_of(&self, query: &str) -> Result<String, Fail> {
896 self.parsed
897 .query(query)
898 .and_then(|q| q.connection.clone())
899 .ok_or_else(|| format!("query `{query}` has no connection"))
900 }
901
902 fn outcome(&self, status: Status, error: Option<String>) -> BindingOutcome {
903 BindingOutcome {
904 report: self.report.name.clone(),
905 set: self.b.set.clone(),
906 binding: self.b.dir_name().to_string(),
907 status,
908 error,
909 files: self.files().map(|(p, _)| p.clone()).collect(),
910 summary: self.summary_line(),
911 schedule: self.opts.schedule.clone(),
912 schedule_vars: self.schedule_vars.clone(),
913 vars: self.rendered_vars.clone(),
914 timezone: self.calendar.tz.name().to_string(),
915 elapsed: self.started.elapsed(),
916 }
917 }
918
919 fn run(&mut self) -> BindingOutcome {
920 let dry = self.opts.dry_run || self.opts.live_check;
921 if !dry {
922 let _ = std::fs::remove_dir_all(&self.run_dir);
923 }
924 let result = self.run_inner();
925 let status = match (&result, dry, self.opts.live_check) {
926 (Err(_), _, _) => Status::Error,
927 (Ok(()), _, true) => Status::Checked,
928 (Ok(()), true, _) => Status::DryRun,
929 (Ok(()), false, _) => Status::Success,
930 };
931 let err = result.err();
932 if !dry && let Err(e) = self.write_results(&status, err.as_deref()) {
933 self.ui.warn(&format!("can't write run_results.json: {e}"));
934 }
935 let _ = std::fs::remove_dir_all(self.run_dir.join(".spool"));
937 self.outcome(status, err)
938 }
939
940 fn run_inner(&mut self) -> Result<(), Fail> {
941 self.parsed = crate::parse::binding(self.project, self.report, self.b, &self.vars, &self.inputs());
943 if !self.parsed.errors.is_empty() {
944 let msgs: Vec<String> = self
945 .parsed
946 .errors
947 .iter()
948 .map(|p| match p.line {
949 Some(l) => format!("{}:{l}: {}", p.file.display(), p.message),
950 None => format!("{}: {}", p.file.display(), p.message),
951 })
952 .collect();
953 return Err(msgs.join("\n "));
954 }
955 let pool = Arc::new(Pool::new(
958 self.connections(),
959 self.project.root.clone(),
960 self.ui.plugin_log(),
961 !self.report.managed,
962 ));
963 self.pool = Some(pool.clone());
964
965 if !self.report.managed {
966 self.ui.warn(&format!(
967 "`{}` is an unmanaged report ({}): for quick tests only — add a YAML to make it a managed report",
968 self.report.name,
969 self.report.file.display()
970 ));
971 }
972
973 std::fs::create_dir_all(&self.compiled_dir).map_err(|e| e.to_string())?;
975 let mut per_connection: BTreeMap<String, usize> = BTreeMap::new();
977 for q in &self.b.queries {
978 *per_connection.entry(self.connection_of(&q.query)?).or_default() += 1;
979 }
980 let interleave = !self.opts.dry_run && !self.opts.live_check && self.report.managed;
984 let mut statements = Vec::new();
985 let mut two_tabs: Option<String> = None;
987 if interleave {
988 self.render_destinations()?;
989 let mut checked: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
990 for (c, n) in &per_connection {
991 if *n > 1 {
992 self.check_sessions(&pool, c, *n)?;
993 checked.insert(c.clone());
994 }
995 }
996 std::fs::create_dir_all(self.run_dir.join(".spool")).map_err(|e| e.to_string())?;
997 let mut i = 0;
998 let queries = self.b.queries.clone();
999 for q in &queries {
1000 let conn = self.connection_of(&q.query)?;
1001 let renderer = self.renderer(Some(&conn))?;
1002 let sts = self.render_query(&renderer, q, &conn, &mut two_tabs)?;
1003 for w in renderer.take_warnings() {
1004 self.ui.warn(&w);
1005 }
1006 if let Some(e) = two_tabs.take() {
1007 return Err(e);
1008 }
1009 if sts.len() > 1 && checked.insert(conn.clone()) {
1010 self.check_sessions(&pool, &conn, sts.len())?;
1011 }
1012 let session = pool.session(&conn)?;
1013 for st in &sts {
1014 self.execute_statement(&session, i, st)?;
1015 i += 1;
1016 }
1017 }
1018 } else {
1019 let queries = self.b.queries.clone();
1020 for q in &queries {
1021 let conn = self.connection_of(&q.query)?;
1022 let renderer = self.renderer(Some(&conn))?;
1023 statements.extend(self.render_query(&renderer, q, &conn, &mut two_tabs)?);
1024 }
1025 self.render_destinations()?;
1026 for r in self.renderers.values() {
1027 for w in r.take_warnings() {
1028 self.ui.warn(&w);
1029 }
1030 }
1031
1032 if !self.report.managed {
1034 if let Some(bad) = statements.iter().find(|s| !s.kind.is_read_only_safe()) {
1035 return Err(format!(
1036 "{}:{}: 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",
1037 bad.file.display(),
1038 bad.line,
1039 self.report.name,
1040 masked_sql_summary(&bad.text, 60, bad.sensitive)
1041 ));
1042 }
1043 for conn in per_connection.keys() {
1044 let session = pool.session(conn)?;
1045 let creates_temp = statements
1046 .iter()
1047 .any(|s| &s.connection == conn && s.kind == StatementKind::TempCreate)
1048 || session.lock().unwrap().loaded;
1049 session.lock().unwrap().want_read_only(!creates_temp);
1050 }
1051 }
1052 if let Some(e) = two_tabs {
1053 return Err(e);
1054 }
1055
1056 if self.opts.dry_run {
1057 let plan = self.plan();
1058 self.ui.compiled(&plan);
1059 return Ok(());
1060 }
1061
1062 for conn in per_connection.keys() {
1064 let n = statements.iter().filter(|s| &s.connection == conn).count();
1065 if n > 1 {
1066 self.check_sessions(&pool, conn, n)?;
1067 }
1068 }
1069
1070 if self.opts.live_check {
1071 return self.live_check(&statements);
1072 }
1073
1074 std::fs::create_dir_all(self.run_dir.join(".spool")).map_err(|e| e.to_string())?;
1076 for (i, st) in statements.iter().enumerate() {
1077 let session = pool.session(&st.connection)?;
1078 self.execute_statement(&session, i, st)?;
1079 }
1080 }
1081 pool.close_all();
1082
1083 self.name_result_sets()?;
1084
1085 let (file_outs, message_outs): (Vec<usize>, Vec<usize>) =
1087 (0..self.outs.len()).partition(|&i| !self.b.outputs[i].is_message());
1088 for &oi in &file_outs {
1089 if !self.passes_when(oi)? {
1090 continue;
1091 }
1092 let filename = self.file_names(oi);
1093 self.format(oi, &filename)?;
1094 }
1095
1096 if self.opts.preview.is_none() {
1098 self.drift = self.schema_drift();
1099 if !self.drift.is_empty() {
1100 if self.opts.accept_schema_change {
1101 self.ui.warn(&format!(
1102 " schema changed since the last successful run (accepted): {}",
1103 self.drift.join("; ")
1104 ));
1105 } else {
1106 return Err(format!(
1107 "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",
1108 self.drift.join("; "),
1109 rel(&self.project.root, &self.run_dir).display()
1110 ));
1111 }
1112 }
1113 }
1114
1115 let preview = self.opts.preview.is_some();
1118 let mut failures = Vec::new();
1119 for oi in file_outs.into_iter().chain(message_outs) {
1120 if self.outs[oi].status == Some(OutputStatus::Skipped) {
1121 continue;
1122 }
1123 if self.b.outputs[oi].is_message() {
1124 match self.render_message(oi) {
1125 Ok(true) => {}
1126 Ok(false) => continue,
1127 Err(e) => {
1128 let e = format!("{}: {e}", self.b.outputs[oi].label(oi));
1129 self.outs[oi].status = Some(OutputStatus::Failed);
1130 self.outs[oi].error = Some(e.clone());
1131 failures.push(e);
1132 continue;
1133 }
1134 }
1135 }
1136 if preview {
1137 self.outs[oi].delivery_note = Some("preview: not delivered".into());
1138 self.outs[oi].status = Some(OutputStatus::Kept);
1139 } else if let Err(e) = self.deliver(oi) {
1140 failures.push(e);
1141 }
1142 }
1143 if preview {
1144 let dir = rel(&self.project.root, &self.run_dir);
1145 self.ui.step(
1146 Level::Info,
1147 "Preview",
1148 &format!("not delivered; output stays in {}", dir.display()),
1149 None,
1150 );
1151 }
1152 if !failures.is_empty() {
1153 return Err(failures.join("; "));
1154 }
1155
1156 if self.opts.preview.is_none()
1158 && let Err(error) = self.write_snapshot()
1159 {
1160 self.ui.warn(&format!(
1161 "delivery completed, but the schema snapshot could not be written: {error}; the next run may compare against an older schema"
1162 ));
1163 }
1164 Ok(())
1165 }
1166
1167 fn results_for(&mut self, oi: usize, max_rows: u64) -> Result<Vec<(String, Arc<QueryResult>)>, Fail> {
1169 let o = &self.b.outputs[oi];
1170 let mut out = Vec::new();
1171 for p in self.produced.iter().filter(|p| o.feeds(&p.query)) {
1172 let r = QueryResult::read(&p.spool, p.rows, max_rows, self.calendar)
1173 .map_err(|e| format!("can't read the result of `{}`: {e}", p.query))?;
1174 out.push((p.query.clone(), Arc::new(r)));
1175 }
1176 Ok(out)
1177 }
1178
1179 fn template_context(&self, oi: usize, results: &[(String, Arc<QueryResult>)]) -> minijinja::Value {
1182 use minijinja::Value;
1183 let results: BTreeMap<String, Value> = results
1184 .iter()
1185 .map(|(q, r)| (q.clone(), r.clone().value()))
1186 .collect();
1187 let mut outputs: BTreeMap<String, Value> = BTreeMap::new();
1188 for (j, (o, run)) in self.b.outputs.iter().zip(&self.outs).enumerate() {
1189 let Some(name) = &o.name else { continue };
1190 if j == oi {
1191 continue;
1192 }
1193 let locations: Vec<&str> = run
1194 .deliveries
1195 .iter()
1196 .filter_map(|d| d.get("location").and_then(Json::as_str))
1197 .collect();
1198 let files: Vec<String> = run
1199 .files
1200 .iter()
1201 .map(|(f, _)| record_path(self.project, f).to_string_lossy().to_string())
1202 .collect();
1203 let status = run.status.map_or("pending", OutputStatus::as_str);
1204 outputs.insert(
1205 name.clone(),
1206 Value::from_serialize(json!({
1207 "location": locations.join(", "),
1208 "files": files,
1209 "status": status,
1210 })),
1211 );
1212 }
1213 Value::from(BTreeMap::from([
1214 ("results".to_string(), Value::from(results)),
1215 ("outputs".to_string(), Value::from(outputs)),
1216 ]))
1217 }
1218
1219 fn passes_when(&mut self, oi: usize) -> Result<bool, Fail> {
1221 if self.b.outputs[oi].when.is_none() {
1222 return Ok(true);
1223 }
1224 let results = self.results_for(oi, crate::message::DEFAULT_MAX_ROWS)?;
1225 self.when_holds(oi, &results)
1226 }
1227
1228 fn when_holds(&mut self, oi: usize, results: &[(String, Arc<QueryResult>)]) -> Result<bool, Fail> {
1230 let Some(when) = self.b.outputs[oi].when.clone() else {
1231 return Ok(true);
1232 };
1233 let ctx = self.template_context(oi, results);
1234 let renderer = self.renderer(None)?;
1235 let label = self.b.outputs[oi].label(oi);
1236 let src = format!("{{% if {when} %}}true{{% endif %}}");
1237 let passed = renderer
1238 .render_with(&self.report.file, &src, ctx, false)
1239 .map_err(|e| format!("{label}: `when` `{when}`: {}", e.message))?
1240 == "true";
1241 self.outs[oi].when = Some(passed);
1242 if !passed {
1243 self.skip(oi, &format!("`when` is false ({when})"));
1244 }
1245 Ok(passed)
1246 }
1247
1248 fn skip(&mut self, oi: usize, why: &str) {
1249 let label = self.b.outputs[oi].label(oi);
1250 self.outs[oi].status = Some(OutputStatus::Skipped);
1251 self.outs[oi].delivery_note = Some(format!("skipped: {why}"));
1252 self.ui
1253 .step(Level::Info, "Skipped", &format!("{label}: {why}"), None);
1254 }
1255
1256 fn render_message(&mut self, oi: usize) -> Result<bool, Fail> {
1259 let o = &self.b.outputs[oi];
1260 let opt = |k: &str| o.options.get(k).and_then(Json::as_str).map(str::to_string);
1261 let (text_src, file_src, title_src) = (opt("text"), opt("file"), opt("title"));
1262 let max_rows = o
1263 .options
1264 .get("max_rows")
1265 .and_then(Json::as_u64)
1266 .unwrap_or(crate::message::DEFAULT_MAX_ROWS);
1267 let results = self.results_for(oi, max_rows)?;
1268 if !self.when_holds(oi, &results)? {
1269 return Ok(false);
1270 }
1271 for (q, r) in results.iter().filter(|(_, r)| r.capped()) {
1272 self.ui.warn(&format!(
1273 " results.{q}.rows holds the first {} of {} rows (`max_rows`); aggregate in SQL, or raise `max_rows`",
1274 thousands(r.rows.len() as u64),
1275 thousands(r.row_count)
1276 ));
1277 }
1278 let ctx = self.template_context(oi, &results);
1279 let renderer = self.renderer(None)?;
1280 let file = self.report.file.clone();
1281 let text = match (text_src, file_src) {
1282 (Some(t), _) => renderer
1283 .render_with(&file, &t, ctx.clone(), true)
1284 .map_err(|e| e.message)?,
1285 (None, Some(f)) => {
1286 let root = &self.project.root;
1287 let path = crate::project::find_template(root, &f)
1288 .ok_or_else(|| format!("message file `{f}` doesn't exist"))?;
1289 let src = std::fs::read_to_string(&path).map_err(|e| format!("{}: {e}", path.display()))?;
1290 let shown = rel(root, &path);
1291 renderer
1292 .render_with(&shown, &src, ctx.clone(), true)
1293 .map_err(|e| e.to_string())?
1294 }
1295 (None, None) => crate::message::default_text(&results, self.locale()),
1296 };
1297 let text = text.trim().to_string();
1298 if text.is_empty() {
1299 self.skip(oi, "the message is empty");
1300 return Ok(false);
1301 }
1302 let title = match title_src {
1303 Some(t) => renderer
1304 .render_with(&file, &t, ctx, false)
1305 .map_err(|e| e.message)?,
1306 None => format!("{}: {}", self.report.name, self.date),
1307 };
1308 let title = title.split_whitespace().collect::<Vec<_>>().join(" ");
1309 let name = self.file_names(oi);
1310 let path = self.run_dir.join(&name);
1311 if let Some(prev) = (0..oi).find(|&j| self.outs[j].files.iter().any(|(f, _)| f == &path)) {
1312 return Err(format!(
1313 "it writes {name}, as {} does; give them different `name:`s",
1314 self.b.outputs[prev].label(prev)
1315 ));
1316 }
1317 std::fs::create_dir_all(&self.run_dir).map_err(|e| e.to_string())?;
1318 std::fs::write(&path, crate::message::file_body(&title, &text)).map_err(|e| e.to_string())?;
1319 self.outs[oi].files.push((path, None));
1320 if self.opts.preview.is_some() {
1321 let mut notes = Vec::new();
1322 if let Some(n) = self.opts.preview {
1323 notes.push(format!(
1324 "numbers come from a sample of at most {} rows per query (--preview)",
1325 thousands(n)
1326 ));
1327 }
1328 if let Some(w) = &self.b.outputs[oi].when {
1329 notes.push(format!("`when` passed ({w})"));
1330 }
1331 let length = text.chars().count();
1332 notes.push(format!("{} characters", thousands(length as u64)));
1333 notes.extend(self.destination_limits(oi, length));
1334 self.ui.message(&ShownMessage {
1335 output: self.b.outputs[oi].name.clone(),
1336 title: title.clone(),
1337 text: text.clone(),
1338 notes,
1339 });
1340 } else {
1341 self.ui.step(
1342 Level::Info,
1343 "Message",
1344 &crate::message::excerpt(&title, &text, 100),
1345 None,
1346 );
1347 }
1348 self.outs[oi].message = Some((title, text));
1349 Ok(true)
1350 }
1351
1352 fn destination_limits(&self, oi: usize, length: usize) -> Vec<String> {
1354 let mut out = Vec::new();
1355 let Some(pool) = &self.pool else { return out };
1356 for d in &self.outs[oi].dests {
1357 let Some(t) = self.project.profiles.target(Role::Destination, &d.profile) else {
1358 out.push(format!("{}: delivers nowhere on this target", d.profile));
1359 continue;
1360 };
1361 let described = pool.connections.described(Role::Destination, &t.kind);
1362 let takes = described
1363 .as_ref()
1364 .is_some_and(|x| x.capabilities.iter().any(|c| c == CAP_MESSAGE));
1365 let limit = described.and_then(|x| x.message_limit);
1366 out.push(match (takes, limit) {
1367 (false, _) => format!("{} ({}): delivers the .md file", d.profile, t.kind),
1368 (true, Some(l)) if length as u64 > l => format!(
1369 "{} ({}): {} of {} characters: over the limit, so it will be cut short",
1370 d.profile,
1371 t.kind,
1372 thousands(length as u64),
1373 thousands(l)
1374 ),
1375 (true, Some(l)) => format!(
1376 "{} ({}): {} of {} characters",
1377 d.profile,
1378 t.kind,
1379 thousands(length as u64),
1380 thousands(l)
1381 ),
1382 (true, None) => format!("{} ({}): posts the message", d.profile, t.kind),
1383 });
1384 }
1385 out
1386 }
1387
1388 fn locale(&self) -> crate::numbers::Locale {
1390 self.b
1391 .locale
1392 .as_deref()
1393 .and_then(|l| crate::numbers::Locale::parse(l).ok())
1394 .unwrap_or_default()
1395 }
1396
1397 fn render_query(
1399 &mut self,
1400 renderer: &Renderer,
1401 q: &QueryEntry,
1402 connection: &str,
1403 two_tabs: &mut Option<String>,
1404 ) -> Result<Vec<Statement>, Fail> {
1405 let mut out = Vec::new();
1406 let src = std::fs::read_to_string(self.project.root.join(&q.path))
1407 .map_err(|e| format!("{}: {e}", q.path.display()))?;
1408 let _ = renderer.take_sources();
1409 let sql = renderer
1410 .render(&q.path, &src)
1411 .map_err(|e: RenderError| e.to_string())?;
1412 let known = self
1415 .parsed
1416 .query(&q.query)
1417 .map(|p| p.sources.clone())
1418 .unwrap_or_default();
1419 if let Some((s, t)) = renderer
1420 .take_sources()
1421 .into_iter()
1422 .find(|(s, t)| !known.contains(&format!("{s}.{t}")))
1423 {
1424 return Err(format!(
1425 "{}: `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",
1426 q.path.display(),
1427 q.query
1428 ));
1429 }
1430 let sensitive = crate::secrets::contains_secret(&sql);
1431 std::fs::write(
1432 self.compiled_dir.join(format!("{}.sql", q.query)),
1433 crate::secrets::mask(&sql).as_bytes(),
1434 )
1435 .map_err(|e| e.to_string())?;
1436 self.ui
1437 .step(Level::Debug, "Rendered", &q.path.display().to_string(), None);
1438 let parts = sqlsplit::split(&sql);
1441 if q.tab && parts.is_empty() {
1442 return Err(format!(
1443 "{}: `{}` makes a tab but has no statements; add a query, or `tab: false` if it's meant to be empty",
1444 q.path.display(),
1445 q.query
1446 ));
1447 }
1448 let n = parts.len();
1449 for (i, st) in parts.into_iter().enumerate() {
1450 let body = sqlsplit::strip_leading_comments(&st.text);
1451 let line = st.line + st.text[..st.text.len() - body.len()].matches('\n').count();
1452 let kind = sqlsplit::classify(&st.text);
1453 let last = i + 1 == n;
1454 if q.tab && !last && kind == StatementKind::Read && two_tabs.is_none() {
1455 *two_tabs = Some(format!(
1456 "{}:{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:`",
1457 q.path.display(),
1458 q.query
1459 ));
1460 }
1461 out.push(Statement {
1462 query: q.query.clone(),
1463 connection: connection.to_string(),
1464 file: q.path.clone(),
1465 line,
1466 kind,
1467 text: st.text,
1468 sensitive,
1469 tab: q.tab && last,
1470 });
1471 }
1472 Ok(out)
1473 }
1474
1475 fn check_sessions(&self, pool: &Pool, connection: &str, n: usize) -> Result<(), Fail> {
1477 let session = pool.session(connection)?;
1478 let mut s = session.lock().unwrap();
1479 if !s.get()?.has(CAP_SESSIONS) {
1480 let kind = pool.kind(connection).unwrap_or_default();
1481 return Err(format!(
1482 "this Binding runs {n} statements on connection `{connection}`, but the `{kind}` source plugin can't hold one session across them; nothing was run"
1483 ));
1484 }
1485 Ok(())
1486 }
1487
1488 fn execute_statement(
1490 &mut self,
1491 session: &Arc<Mutex<Session>>,
1492 i: usize,
1493 st: &Statement,
1494 ) -> Result<(), Fail> {
1495 let sql_log = self.ui.sql_log();
1496 sql_log(
1497 &format!("{}:{}", st.file.display(), st.line),
1498 &protected_sql(&st.text, st.sensitive),
1499 );
1500 let spool_path = self.run_dir.join(".spool").join(format!("{i}.arrow"));
1501 let mut writer: Option<FileWriter<File>> = None;
1502 let t = Instant::now();
1503 let exec = {
1504 let mut s = session.lock().unwrap();
1505 let (p, _log_scope) = s.statement(st.sensitive)?;
1506 p.execute(&st.text, self.opts.preview, |schema, batch| {
1507 if writer.is_none() {
1508 let f = File::create(&spool_path).map_err(|e| e.to_string())?;
1509 writer = Some(FileWriter::try_new(f, schema).map_err(|e| e.to_string())?);
1510 }
1511 writer.as_mut().unwrap().write(&batch).map_err(|e| e.to_string())
1512 })
1513 .map_err(|e| {
1514 format!(
1515 "{}:{}: {}",
1516 st.file.display(),
1517 st.line,
1518 masked_source_error(&e, st.sensitive)
1519 )
1520 })?
1521 };
1522 let what = match &exec {
1523 Execution::Result { rows, .. } => {
1524 format!("{} row{}", thousands(*rows), if *rows == 1 { "" } else { "s" })
1525 }
1526 Execution::NoResult {
1527 rows_affected: Some(n),
1528 } => format!("no result set ({n} affected)"),
1529 Execution::NoResult { .. } => "no result set".to_string(),
1530 };
1531 let at = format!("{}:{}", st.file.display(), st.line);
1532 self.ui.step(
1533 Level::Debug,
1534 "Executed",
1535 &format!("{at} {what}"),
1536 Some(t.elapsed()),
1537 );
1538 let (schema, rows) = match (st.tab, exec) {
1541 (true, Execution::Result { schema, rows }) => (schema, rows),
1542 (true, Execution::NoResult { .. }) => {
1543 return Err(format!(
1544 "{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",
1545 st.query
1546 ));
1547 }
1548 (false, _) => {
1549 drop(writer);
1550 let _ = std::fs::remove_file(&spool_path);
1551 return Ok(());
1552 }
1553 };
1554 {
1555 let mut w = match writer {
1556 Some(w) => w,
1557 None => FileWriter::try_new(File::create(&spool_path).map_err(|e| e.to_string())?, &schema)
1558 .map_err(|e| e.to_string())?,
1559 };
1560 w.finish().map_err(|e| e.to_string())?;
1561 self.produced.push(Produced {
1562 query: st.query.clone(),
1563 connection: st.connection.clone(),
1564 schema,
1565 rows,
1566 spool: spool_path,
1567 name: String::new(),
1568 anchor: None,
1569 header: None,
1570 columns: Default::default(),
1571 });
1572 }
1573 Ok(())
1574 }
1575
1576 fn render_destinations(&mut self) -> Result<(), Fail> {
1579 if self.b.destinations().next().is_none() {
1580 return Ok(());
1581 }
1582 let renderer = self.renderer(None)?;
1583 let connections = self.pool.as_ref().expect("pool").connections.clone();
1584 let numbered: Vec<(usize, usize, &crate::project::Destination)> = self
1586 .b
1587 .outputs
1588 .iter()
1589 .enumerate()
1590 .flat_map(|(oi, o)| o.destinations.iter().map(move |d| (oi, d)))
1591 .enumerate()
1592 .map(|(i, (oi, d))| (i, oi, d))
1593 .collect();
1594 for (i, oi, d) in numbered {
1595 let profile = self
1596 .parsed
1597 .destinations
1598 .get(i)
1599 .cloned()
1600 .flatten()
1601 .ok_or_else(|| format!("destination `{}` has no profile", d.profile))?;
1602 if matches!(
1604 self.project.profiles.entry(Role::Destination, &profile),
1605 Entry::Nowhere
1606 ) {
1607 self.outs[oi].dests.push(RenderedDest {
1608 profile,
1609 path: None,
1610 options: JsonMap::new(),
1611 attach: Vec::new(),
1612 });
1613 continue;
1614 }
1615 renderer.set_destination(connections.profile(&profile, Some("destination")).ok());
1616 let rendered = (|| {
1617 let path = match &d.path {
1618 Some(p) => Some(
1619 renderer
1620 .render(&self.report.file, p)
1621 .map_err(|e| format!("output path: {e}"))?,
1622 ),
1623 None => None,
1624 };
1625 let mut options = JsonMap::new();
1626 for (k, v) in &d.options {
1627 let v = render_json(&renderer, &self.report.file, v)
1628 .map_err(|e| format!("destination `{profile}` option `{k}`: {e}"))?;
1629 options.insert(k.clone(), v);
1630 }
1631 Ok::<_, Fail>(RenderedDest {
1632 profile: profile.clone(),
1633 path,
1634 options,
1635 attach: d.attach.clone(),
1636 })
1637 })();
1638 renderer.set_destination(None);
1639 self.outs[oi].dests.push(rendered?);
1640 }
1641 Ok(())
1642 }
1643
1644 fn name_result_sets(&mut self) -> Result<(), Fail> {
1646 let entries: BTreeMap<&str, &QueryEntry> =
1647 self.b.queries.iter().map(|q| (q.query.as_str(), q)).collect();
1648 for p in &mut self.produced {
1649 let q = entries[p.query.as_str()];
1650 p.name = q.tab_name.clone().unwrap_or_else(|| q.query.clone());
1651 p.anchor = q.anchor.clone();
1652 p.header = q.header;
1653 p.columns = q.columns.clone();
1654 }
1655 for o in self.b.outputs.iter().filter(|o| o.format == "xlsx") {
1656 let mut seen: BTreeMap<String, String> = BTreeMap::new();
1657 for p in self.produced.iter().filter(|p| o.feeds(&p.query)) {
1658 check_sheet_name(&p.name)?;
1659 if let Some(prev) = seen.insert(p.name.to_lowercase(), p.query.clone()) {
1660 return Err(format!(
1661 "sheet name `{}` is used twice (by `{prev}` and `{}`); Excel sheet names must be unique",
1662 p.name, p.query
1663 ));
1664 }
1665 }
1666 }
1667 Ok(())
1668 }
1669
1670 fn plan(&mut self) -> BindingPlan {
1672 let compiled = self
1673 .b
1674 .queries
1675 .iter()
1676 .map(|q| {
1677 rel(
1678 &self.project.root,
1679 &self.compiled_dir.join(format!("{}.sql", q.query)),
1680 )
1681 })
1682 .collect();
1683 let mut outputs = Vec::new();
1684 for oi in 0..self.outs.len() {
1685 let file = self.file_names(oi);
1686 let o = &self.b.outputs[oi];
1687 let destinations = self.outs[oi]
1688 .dests
1689 .iter()
1690 .map(|d| {
1691 let out = self.project.profiles.target(Role::Destination, &d.profile);
1692 PlannedDestination {
1693 profile: d.profile.clone(),
1694 kind: out.map(|o| o.kind.clone()),
1695 target: Some(self.project.profiles.target_of(Role::Destination, &d.profile)),
1696 path: d.path.clone(),
1697 delivers: out.is_some(),
1698 }
1699 })
1700 .collect();
1701 outputs.push(PlannedOutput {
1702 name: o.name.clone(),
1703 format: o.format.clone(),
1704 output: rel(&self.project.root, &self.run_dir.join(file)),
1705 destinations,
1706 });
1707 }
1708 let first = outputs.first().cloned();
1709 let pool = self.pool.clone().expect("pool");
1710 let queries = self
1711 .parsed
1712 .queries
1713 .iter()
1714 .map(|q| {
1715 let connection = q.connection.clone().unwrap_or_default();
1716 PlannedQuery {
1717 query: q.query.clone(),
1718 kind: pool.kind(&connection).unwrap_or_default(),
1719 target: self.project.profiles.target_of(Role::Connection, &connection),
1720 connection,
1721 sources: q.sources.clone(),
1722 }
1723 })
1724 .collect();
1725 BindingPlan {
1726 report: self.report.name.clone(),
1727 set: self.b.set.clone(),
1728 compiled,
1729 profile: self.parsed.inherited.clone(),
1730 queries,
1731 target: self.target.clone(),
1732 format: first.as_ref().map(|o| o.format.clone()).unwrap_or_default(),
1733 output: first.as_ref().map(|o| o.output.clone()).unwrap_or_default(),
1734 destinations: first.map(|o| o.destinations).unwrap_or_default(),
1735 outputs,
1736 schedules: self.b.schedules.clone(),
1737 }
1738 }
1739
1740 fn file_names(&mut self, oi: usize) -> String {
1745 let first = oi == 0;
1746 let (output_path, output_name) = if first {
1747 (self.opts.output_path.clone(), self.opts.output_name.clone())
1748 } else {
1749 (None, None)
1750 };
1751 for d in self.outs[oi].dests.iter_mut().filter(|d| d.path.is_some()) {
1752 if let Some(p) = &output_path {
1753 d.path = Some(p.clone());
1754 }
1755 if let (Some(n), Some(r)) = (&output_name, d.path.as_mut()) {
1756 *r = match r.rfind(['/', '\\']) {
1757 Some(i) => format!("{}{n}", &r[..=i]),
1758 None => n.clone(),
1759 };
1760 }
1761 }
1762 let o = &self.b.outputs[oi];
1763 let ext = match &o.extension {
1764 Some(e) => e.as_str(),
1765 None => extension(&o.format),
1766 };
1767 let stem = o.name.clone().unwrap_or_else(|| self.report.name.clone());
1768 let from_remote = self.outs[oi]
1769 .dests
1770 .iter()
1771 .find_map(|d| d.path.as_deref())
1772 .and_then(|r| r.rsplit(['/', '\\']).next())
1773 .filter(|s| !s.is_empty())
1774 .map(str::to_string);
1775 output_name.or(from_remote).unwrap_or_else(|| match ext {
1776 "" => stem,
1777 ext => format!("{stem}.{ext}"),
1778 })
1779 }
1780
1781 fn format(&mut self, oi: usize, filename: &str) -> Result<(), Fail> {
1783 std::fs::create_dir_all(&self.run_dir).map_err(|e| e.to_string())?;
1784 let b = self.b;
1785 let out = &b.outputs[oi];
1786 let mine: Vec<usize> = (0..self.produced.len())
1787 .filter(|&i| out.feeds(&self.produced[i].query))
1788 .collect();
1789 if mine.is_empty() {
1790 let what = if self.outs.len() > 1 {
1791 format!("{} gets no tab", out.label(oi))
1792 } else {
1793 "no query makes a tab (every query has `tab: false`)".to_string()
1794 };
1795 self.ui.warn(&format!(" {what}; no output file was written"));
1796 return Ok(());
1797 }
1798 let multi = out.format == "xlsx" || out.template.is_some() || !is_single_table(&out.format);
1799 let groups: Vec<(String, Vec<usize>)> = if multi || mine.len() == 1 {
1800 vec![(filename.to_string(), mine)]
1801 } else {
1802 let (stem, ext) = split_ext(filename);
1803 let mut g = Vec::new();
1804 for i in mine {
1805 let r = &self.produced[i];
1806 if r.name.contains(['/', '\\']) {
1807 return Err(format!("`{}` can't be used in a file name", r.name));
1808 }
1809 g.push((format!("{stem}_{}{ext}", r.name), vec![i]));
1810 }
1811 g
1812 };
1813 for (name, _) in &groups {
1815 let path = self.run_dir.join(name);
1816 if let Some(prev) = (0..oi).find(|&j| self.outs[j].files.iter().any(|(f, _)| f == &path)) {
1817 return Err(format!(
1818 "{} and {} both write {name}; give them different `name:`s or destination paths",
1819 self.b.outputs[prev].label(prev),
1820 out.label(oi)
1821 ));
1822 }
1823 }
1824 let plugin = find_plugin(self.project, PluginKind::Format, &out.format)?;
1825 let mut p = plugin
1826 .start(self.ui.plugin_log(), Some(&self.project.root))
1827 .map_err(|e| e.to_string())?;
1828 let template = match &out.template {
1829 Some(t) => Some(self.template_payload(t)?),
1830 None => None,
1831 };
1832 for (name, idxs) in groups {
1833 let t = Instant::now();
1834 let path = self.run_dir.join(&name);
1835 let metas: Vec<ResultSetMeta> = idxs
1836 .iter()
1837 .map(|&i| {
1838 let r = &self.produced[i];
1839 ResultSetMeta {
1840 name: r.name.clone(),
1841 query: r.query.clone(),
1842 result_index: 1,
1843 anchor: r.anchor.clone(),
1844 header: r.header,
1845 columns: r.columns.clone(),
1846 }
1847 })
1848 .collect();
1849 p.write_begin(
1850 &path.to_string_lossy(),
1851 &out.format,
1852 out.options.clone(),
1853 metas,
1854 template.clone(),
1855 )
1856 .map_err(|e| e.to_string())?;
1857 for &i in &idxs {
1858 let r = &self.produced[i];
1859 let reader = FileReader::try_new(File::open(&r.spool).map_err(|e| e.to_string())?, None)
1860 .map_err(|e| e.to_string())?;
1861 let schema = r.schema.clone();
1862 let mut any = false;
1864 for b in reader {
1865 let b = b.map_err(|e| e.to_string())?;
1866 any = true;
1867 p.send_batch(&b)
1868 .map_err(|e| format!("{} format: {e}", out.format))?;
1869 }
1870 if !any {
1871 p.send_batch(&RecordBatch::new_empty(schema))
1872 .map_err(|e| format!("{} format: {e}", out.format))?;
1873 }
1874 p.send(&dre_protocol::msg::Request::ResultSetEnd {})
1875 .map_err(|e| format!("{} format: {e}", out.format))?;
1876 }
1877 let (files, warnings) = p
1878 .write_finish_with_warnings()
1879 .map_err(|e| format!("{} format: {e}", out.format))?;
1880 for w in warnings {
1881 self.ui.warn(&format!(" {} format: {w}", out.format));
1882 }
1883 for f in files {
1884 let f = PathBuf::from(f);
1885 let size = std::fs::metadata(&f).map(|m| m.len()).unwrap_or(0);
1886 let shown = rel(&self.project.root, &f);
1887 self.ui.step(
1888 Level::Debug,
1889 "Wrote",
1890 &format!("{} ({})", shown.display(), human_bytes(size)),
1891 Some(t.elapsed()),
1892 );
1893 self.outs[oi].files.push((f, None));
1894 }
1895 }
1896 let _ = p.close();
1897 Ok(())
1898 }
1899
1900 fn template_payload(&mut self, t: &crate::project::Template) -> Result<Json, Fail> {
1901 let file = crate::project::find_template(&self.project.root, &t.file)
1902 .ok_or_else(|| format!("template file `{}` doesn't exist", t.file))?;
1903 let renderer = self.renderer(None)?;
1904 let mut values = JsonMap::new();
1905 for b in &t.bindings {
1906 if let (Some(cell), Some(v)) = (&b.cell, &b.value) {
1907 let rendered = renderer
1908 .render(&self.report.file, v)
1909 .map_err(|e| format!("template value: {e}"))?;
1910 values.insert(format!("{}!{}", b.sheet, cell), Json::String(rendered));
1911 }
1912 }
1913 Ok(json!({"file": file.to_string_lossy(), "bindings": t.bindings, "values": values}))
1914 }
1915
1916 fn deliver(&mut self, oi: usize) -> Result<(), Fail> {
1919 let result = self.deliver_output(oi);
1920 let o = &mut self.outs[oi];
1921 o.status = Some(match &result {
1922 Err(_) => OutputStatus::Failed,
1923 Ok(()) if o.deliveries.iter().any(|d| d["status"] == "delivered") => OutputStatus::Delivered,
1924 Ok(()) => OutputStatus::Kept,
1925 });
1926 o.error = result.as_ref().err().cloned();
1927 result
1928 }
1929
1930 fn deliver_output(&mut self, oi: usize) -> Result<(), Fail> {
1931 if self.outs[oi].dests.is_empty() {
1932 self.outs[oi].delivery_note = Some("no destination declared; output stays in target/".into());
1933 let dir = rel(&self.project.root, &self.run_dir);
1934 self.ui.step(
1935 Level::Debug,
1936 "Kept",
1937 &format!("no destination declared: output stays in {}", dir.display()),
1938 None,
1939 );
1940 return Ok(());
1941 }
1942 if self.outs[oi].files.is_empty() {
1943 return Ok(());
1944 }
1945 let dests = std::mem::take(&mut self.outs[oi].dests);
1946 let mut failures = Vec::new();
1947 let mut nowhere = Vec::new();
1948 for d in &dests {
1949 let profiles = &self.project.profiles;
1950 let target = profiles.target_of(Role::Destination, &d.profile);
1951 let kind = profiles
1953 .target(Role::Destination, &d.profile)
1954 .map(|o| o.kind.clone());
1955 let (status, location, error) = match self.deliver_one(oi, d) {
1956 Ok(Some(loc)) => ("delivered", Some(loc), None),
1957 Ok(None) => {
1958 nowhere.push(nowhere_note(&d.profile, &target));
1959 ("not_delivered", None, None)
1960 }
1961 Err(e) => {
1962 failures.push(e.clone());
1963 ("failed", None, Some(e))
1964 }
1965 };
1966 let mut record = json!({"profile": d.profile, "type": kind, "target": target, "status": status});
1967 if let Some(l) = location {
1968 record["location"] = json!(l);
1969 }
1970 if let Some(e) = error {
1971 record["error"] = json!(e);
1972 }
1973 self.outs[oi].deliveries.push(record);
1974 }
1975 let o = &mut self.outs[oi];
1976 o.dests = dests;
1977 if o.files.iter().all(|(_, d)| d.is_none()) && !nowhere.is_empty() {
1978 o.delivery_note = Some(nowhere.join("; "));
1979 }
1980 match failures.len() {
1981 0 => Ok(()),
1982 1 if o.dests.len() == 1 => Err(failures.remove(0)),
1983 n => Err(format!(
1984 "{n} of {} destinations failed: {}",
1985 o.dests.len(),
1986 failures.join("; ")
1987 )),
1988 }
1989 }
1990
1991 fn deliver_one(&mut self, oi: usize, d: &RenderedDest) -> Result<Option<String>, Fail> {
1994 let profiles = &self.project.profiles;
1995 let out = match profiles.entry(Role::Destination, &d.profile) {
1996 Entry::Use(o) => o,
1997 Entry::Nowhere => {
1998 let note = nowhere_note(&d.profile, &profiles.target_of(Role::Destination, &d.profile));
1999 self.ui.step(Level::Info, "Kept", ¬e, None);
2000 return Ok(None);
2001 }
2002 Entry::Missing => return Err(profiles.missing_entry(Role::Destination, &d.profile)),
2004 Entry::Unknown => {
2005 return Err(format!(
2006 "destination profile `{}` isn't in profiles.yml",
2007 d.profile
2008 ));
2009 }
2010 };
2011 let kind = out.kind.clone();
2012 let connection = render_connection(out)?;
2013 let files = &self.outs[oi].files;
2014 let targets: Vec<DeliveryFile> = files
2015 .iter()
2016 .map(|(f, _)| {
2017 let name = f.file_name().unwrap().to_string_lossy().to_string();
2018 let r = d.path.as_deref().map(|r| {
2019 if files.len() == 1 {
2020 r.to_string()
2021 } else {
2022 match r.rfind(['/', '\\']) {
2023 Some(i) => format!("{}{name}", &r[..=i]),
2024 None => name.clone(),
2025 }
2026 }
2027 });
2028 DeliveryFile {
2029 local_path: f.to_string_lossy().to_string(),
2030 remote_path: r,
2031 }
2032 })
2033 .collect();
2034 let mut locations = Vec::new();
2035 if kind == LOCAL_TYPE {
2036 if let Some(k) = d.options.keys().next() {
2037 return Err(format!(
2038 "the local destination takes no options, but `{}` has `{k}`; check the key's spelling",
2039 d.profile
2040 ));
2041 }
2042 for (i, f) in targets.iter().enumerate() {
2043 let t = Instant::now();
2044 let r = f
2045 .remote_path
2046 .as_ref()
2047 .ok_or("the local destination needs `output.destination.path`")?;
2048 let dst = self.project.root.join(r);
2049 if let Some(parent) = dst.parent() {
2050 std::fs::create_dir_all(parent)
2051 .map_err(|e| format!("can't create {}: {e}", parent.display()))?;
2052 }
2053 std::fs::copy(&f.local_path, &dst).map_err(|e| {
2054 format!(
2055 "delivery to {} failed: {e}; the output is still in target/",
2056 dst.display()
2057 )
2058 })?;
2059 let loc = dst.to_string_lossy().to_string();
2060 self.ui.step(Level::Debug, "Delivered", &loc, Some(t.elapsed()));
2061 self.outs[oi].files[i].1.get_or_insert_with(|| loc.clone());
2062 locations.push(loc);
2063 }
2064 return Ok(Some(locations.join(", ")));
2065 }
2066 let failed = |e: dre_protocol::host::HostError| {
2067 format!("delivery through `{kind}` failed: {e}; the output is still in target/")
2068 };
2069 let plugin = find_plugin(self.project, PluginKind::Destination, &kind)?;
2070 let mut p = plugin
2071 .start(self.ui.plugin_log(), Some(&self.project.root))
2072 .map_err(|e| e.to_string())?;
2073 let is_message = self.b.outputs[oi].is_message();
2074 if is_message
2077 && p.has(CAP_MESSAGE)
2078 && let Some((title, text)) = self.outs[oi].message.clone()
2079 {
2080 let t = Instant::now();
2081 let message = dre_protocol::msg::Message {
2082 title,
2083 text,
2084 html: None,
2085 path: targets[0].local_path.clone(),
2086 };
2087 let attach: Vec<DeliveryFile> = d
2088 .attach
2089 .iter()
2090 .filter_map(|name| {
2091 self.b
2092 .outputs
2093 .iter()
2094 .position(|o| o.name.as_deref() == Some(name))
2095 })
2096 .flat_map(|j| self.outs[j].files.iter())
2097 .map(|(f, _)| DeliveryFile {
2098 local_path: f.to_string_lossy().to_string(),
2099 remote_path: None,
2100 })
2101 .collect();
2102 let loc = p
2103 .deliver_message(&message, &attach, connection, d.options.clone())
2104 .map_err(failed)?;
2105 self.ui.step(Level::Debug, "Delivered", &loc, Some(t.elapsed()));
2106 self.outs[oi].files[0].1.get_or_insert_with(|| loc.clone());
2107 let _ = p.close();
2108 return Ok(Some(loc));
2109 }
2110 if !is_message && p.has(CAP_MESSAGE_ONLY) {
2111 let _ = p.close();
2112 return Err(format!(
2113 "`{kind}` only takes messages, but this output is `{}`; deliver the file elsewhere and link it from a message",
2114 self.b.outputs[oi].format
2115 ));
2116 }
2117 let batches: Vec<Vec<usize>> = if targets.len() > 1 && p.has(CAP_MULTI_FILE) {
2118 vec![(0..targets.len()).collect()]
2119 } else {
2120 (0..targets.len()).map(|i| vec![i]).collect()
2121 };
2122 for batch in batches {
2123 let t = Instant::now();
2124 let files: Vec<DeliveryFile> = batch.iter().map(|&i| targets[i].clone()).collect();
2125 let loc = p
2126 .deliver_files(&files, connection.clone(), d.options.clone())
2127 .map_err(failed)?;
2128 self.ui.step(Level::Debug, "Delivered", &loc, Some(t.elapsed()));
2129 for i in batch {
2130 self.outs[oi].files[i].1.get_or_insert_with(|| loc.clone());
2131 }
2132 locations.push(loc);
2133 }
2134 let _ = p.close();
2135 Ok(Some(locations.join(", ")))
2136 }
2137
2138 fn snapshot(&self) -> Json {
2139 let sets: Vec<Json> = self
2140 .produced
2141 .iter()
2142 .map(|p| {
2143 json!({
2144 "name": p.name,
2145 "columns": p.schema.fields().iter().map(|f| json!({
2146 "name": f.name(),
2147 "type": f.data_type().to_string(),
2148 "type_shape": crate::schema::type_shape(f.data_type()),
2149 })).collect::<Vec<_>>(),
2150 })
2151 })
2152 .collect();
2153 json!({ "result_sets": sets })
2154 }
2155
2156 fn schema_drift(&mut self) -> Vec<String> {
2157 let path = self.schema_dir.join("last_success.json");
2158 let mut current = self.snapshot();
2159 let Ok(prev) = std::fs::read_to_string(&path) else {
2160 crate::schema::redact(&mut current);
2161 self.protected_snapshot = Some(current);
2162 return Vec::new();
2163 };
2164 let Ok(mut prev) = serde_json::from_str::<Json>(&prev) else {
2165 crate::schema::redact(&mut current);
2166 self.protected_snapshot = Some(current);
2167 return Vec::new();
2168 };
2169 let original = prev.clone();
2170 crate::schema::redact(&mut prev);
2171 crate::schema::redact(&mut current);
2172 if crate::secrets::enabled() {
2173 crate::schema::align_protection(&mut prev, &mut current);
2174 }
2175 if prev != original {
2176 let protected =
2177 serde_json::to_string_pretty(&prev).expect("serializing a serde_json::Value cannot fail");
2178 if let Err(error) = std::fs::write(&path, protected) {
2179 self.ui.warn(&format!(
2180 "can't protect the existing schema snapshot at {}: {error}; continuing without rewriting it",
2181 path.display()
2182 ));
2183 }
2184 }
2185 let drift = crate::schema::drift(&prev, ¤t);
2186 self.protected_snapshot = Some(current);
2187 drift
2188 }
2189
2190 fn write_snapshot(&self) -> std::io::Result<()> {
2191 std::fs::create_dir_all(&self.schema_dir)?;
2192 let snapshot = self.protected_snapshot.clone().unwrap_or_else(|| {
2193 let mut snapshot = self.snapshot();
2194 crate::schema::redact(&mut snapshot);
2195 snapshot
2196 });
2197 std::fs::write(
2198 self.schema_dir.join("last_success.json"),
2199 serde_json::to_string_pretty(&snapshot)?,
2200 )
2201 }
2202
2203 fn write_results(&self, status: &Status, error: Option<&str>) -> std::io::Result<()> {
2204 std::fs::create_dir_all(&self.run_dir)?;
2205 let mut outputs: Vec<Json> = Vec::new();
2207 let mut output_results: Vec<Json> = Vec::new();
2208 for (o, run) in self.b.outputs.iter().zip(&self.outs) {
2209 let files: Vec<Json> = run
2210 .files
2211 .iter()
2212 .map(|(f, delivered)| {
2213 json!({
2214 "path": record_path(self.project, f),
2215 "size": std::fs::metadata(f).map(|m| m.len()).unwrap_or(0),
2216 "delivered_to": delivered,
2217 })
2218 })
2219 .collect();
2220 for f in &files {
2221 let mut f = f.clone();
2222 f["output"] = json!(o.name);
2223 outputs.push(f);
2224 }
2225 let queries: Vec<&str> = self
2226 .b
2227 .queries
2228 .iter()
2229 .filter(|q| q.tab && o.feeds(&q.query))
2230 .map(|q| q.query.as_str())
2231 .collect();
2232 let status = match (run.status, status) {
2233 (Some(s), _) => s,
2234 (None, Status::Error) => OutputStatus::Failed,
2235 (None, _) => OutputStatus::Kept,
2236 }
2237 .as_str();
2238 let mut record = json!({
2239 "name": o.name,
2240 "format": o.format,
2241 "queries": queries,
2242 "status": status,
2243 "error": run.error,
2244 "files": files,
2245 "delivery": run.delivery_note,
2246 "deliveries": run.deliveries,
2247 });
2248 if let Some(w) = run.when {
2249 record["when"] = json!(w);
2250 }
2251 if let Some((title, text)) = &run.message {
2252 record["message"] = json!({"title": title, "text": text});
2253 }
2254 output_results.push(record);
2255 }
2256 let order = (0..self.outs.len())
2258 .filter(|&i| !self.b.outputs[i].is_message())
2259 .chain((0..self.outs.len()).filter(|&i| self.b.outputs[i].is_message()));
2260 let deliveries: Vec<&Json> = order.flat_map(|i| self.outs[i].deliveries.iter()).collect();
2261 let notes: Vec<&str> = self
2262 .outs
2263 .iter()
2264 .filter_map(|o| o.delivery_note.as_deref())
2265 .collect();
2266 let delivery = (!notes.is_empty()).then(|| notes.join("; "));
2267 let results = json!({
2268 "report": self.report.name,
2269 "set": self.b.set,
2270 "binding": self.b.dir_name(),
2271 "managed": self.report.managed,
2272 "profile": self.parsed.inherited,
2273 "connections": self.parsed.connections(),
2274 "target": self.target,
2275 "schedule": self.opts.schedule,
2276 "schedule_vars": self.schedule_vars,
2277 "vars": self.rendered_vars,
2278 "run_date": self.date.to_string(),
2279 "scheduled_at": self.opts.scheduled_at.map(rfc3339),
2280 "timezone": self.calendar.tz.name(),
2281 "params": self.opts.params(self.date),
2282 "status": status,
2283 "error": error,
2284 "preview": self.opts.preview.is_some(),
2285 "row_limit": self.opts.preview,
2286 "started_at": self.started_at.to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
2287 "duration_ms": self.started.elapsed().as_millis() as u64,
2288 "result_sets": self.produced.iter().map(|p| json!({
2289 "name": p.name,
2290 "query": p.query,
2291 "connection": p.connection,
2292 "rows": p.rows,
2293 "columns": p.schema.fields().iter().map(|f| f.name().clone()).collect::<Vec<_>>(),
2294 })).collect::<Vec<_>>(),
2295 "outputs": outputs,
2296 "output_results": output_results,
2297 "delivery": delivery,
2298 "deliveries": deliveries,
2299 "schema_drift": self.drift,
2300 "target_path": self.project.target_dir,
2301 "manifest_checksum": self.opts.manifest_checksum,
2302 });
2303 std::fs::write(
2304 self.run_dir.join("run_results.json"),
2305 (crate::secrets::to_json_pretty(&results)? + "\n").as_bytes(),
2306 )
2307 }
2308
2309 fn live_check(&mut self, statements: &[Statement]) -> Result<(), Fail> {
2312 let pool = self.pool.clone().expect("pool");
2313 let mut failures = Vec::new();
2314 let mut unexecuted_setup: BTreeMap<String, (PathBuf, usize)> = BTreeMap::new();
2315 let mut uncheckable: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
2316 let sql_log = self.ui.sql_log();
2317 for st in statements {
2318 if uncheckable.contains(&st.connection) {
2319 continue;
2320 }
2321 let session = pool.session(&st.connection)?;
2322 let mut s = session.lock().unwrap();
2323 if !s.get()?.has(dre_protocol::CAP_CHECK) {
2324 let kind = pool.kind(&st.connection).unwrap_or_default();
2325 self.ui.warn(&format!(
2326 " not checkable: the `{kind}` source plugin (connection `{}`) can't check statements without running them",
2327 st.connection
2328 ));
2329 uncheckable.insert(st.connection.clone());
2330 continue;
2331 }
2332 let verb = if st.kind == StatementKind::TempCreate {
2333 ""
2334 } else {
2335 " (check)"
2336 };
2337 sql_log(
2338 &format!("{}:{}{verb}", st.file.display(), st.line),
2339 &protected_sql(&st.text, st.sensitive),
2340 );
2341 if st.kind == StatementKind::TempCreate {
2342 let result = {
2343 let (process, _log_scope) = s.statement(st.sensitive)?;
2344 process.execute(&st.text, None, |_, _| Ok(()))
2345 };
2346 if let Err(e) = result {
2347 failures.push(format!(
2348 "{}:{}: {}",
2349 st.file.display(),
2350 st.line,
2351 masked_source_error(&e, st.sensitive)
2352 ));
2353 }
2354 continue;
2355 }
2356 let result = {
2357 let (process, _log_scope) = s.statement(st.sensitive)?;
2358 process.check(&st.text)
2359 };
2360 if let Err(e) = result {
2361 let mut msg = format!(
2362 "{}:{}: {}",
2363 st.file.display(),
2364 st.line,
2365 masked_source_error(&e, st.sensitive)
2366 );
2367 if let Some((f, l)) = unexecuted_setup.get(&st.connection) {
2368 msg.push_str(&format!(
2369 " (may be a false positive: the setup statement at {}:{l} wasn't executed during the check)",
2370 f.display()
2371 ));
2372 }
2373 failures.push(msg);
2374 }
2375 if st.kind == StatementKind::Other {
2376 unexecuted_setup.insert(st.connection.clone(), (st.file.clone(), st.line));
2377 }
2378 }
2379 self.check_source_columns(&pool, &mut failures);
2380 if failures.is_empty() {
2381 Ok(())
2382 } else {
2383 Err(failures.join("\n "))
2384 }
2385 }
2386
2387 fn check_source_columns(&mut self, pool: &Arc<Pool>, failures: &mut Vec<String>) {
2390 type Checked = std::collections::BTreeSet<(PathBuf, String, String, String)>;
2392 static CHECKED: std::sync::LazyLock<Mutex<Checked>> = std::sync::LazyLock::new(Mutex::default);
2393 for q in self.parsed.queries.clone() {
2394 let Some(conn) = q.connection.clone() else {
2395 continue;
2396 };
2397 for key in &q.sources {
2398 let Some(r) = self.parsed.sources.get(key).cloned() else {
2399 continue;
2400 };
2401 let Some(declared) = self
2402 .project
2403 .sources
2404 .get(&r.source)
2405 .and_then(|s| s.table(&r.table))
2406 .map(|t| t.columns.clone())
2407 .filter(|c| !c.is_empty())
2408 else {
2409 continue;
2410 };
2411 let id = (
2412 self.project.root.clone(),
2413 key.clone(),
2414 conn.clone(),
2415 self.project.profiles.target_of(Role::Connection, &conn),
2416 );
2417 if !CHECKED.lock().unwrap().insert(id) {
2418 continue;
2419 }
2420 let quote = if r.quoting.any() {
2421 let kind = pool.kind(&conn).unwrap_or_default();
2422 match pool.connections.identifier_quote(&kind) {
2423 Ok(q) => q,
2424 Err(e) => {
2425 failures.push(format!("source `{key}`: can't quote its name: {e}"));
2426 continue;
2427 }
2428 }
2429 } else {
2430 None
2431 };
2432 let rel = r.relation(quote.as_deref());
2433 let runner = PoolRunner {
2434 pool: pool.clone(),
2435 default: Some(conn.clone()),
2436 log: self.ui.sql_log(),
2437 };
2438 let actual = match runner.columns(&format!("select * from {rel} where 1=0"), None) {
2439 Ok(c) => c,
2440 Err(e) => {
2441 failures.push(format!("source `{key}` ({rel} on connection `{conn}`): {e}"));
2442 continue;
2443 }
2444 };
2445 for c in &declared {
2446 let Some(a) = actual.iter().find(|a| a.name.eq_ignore_ascii_case(&c.name)) else {
2447 failures.push(format!(
2448 "source `{key}`: declared column `{}` isn't in {rel} on connection `{conn}` (it has: {})",
2449 c.name,
2450 actual.iter().map(|a| a.name.as_str()).collect::<Vec<_>>().join(", ")
2451 ));
2452 continue;
2453 };
2454 let Some(dt) = &c.data_type else { continue };
2455 match crate::coltypes::matches(dt, &a.data_type) {
2456 Some(true) => {}
2457 Some(false) => failures.push(format!(
2458 "source `{key}`: column `{}` is declared `{dt}`, but {rel} on connection `{conn}` returns {}",
2459 c.name, a.data_type
2460 )),
2461 None => self.ui.warn(&format!(
2462 " source `{key}`: column `{}`'s `data_type: {dt}` isn't a type DRE can compare, so only its presence was checked",
2463 c.name
2464 )),
2465 }
2466 }
2467 }
2468 }
2469 }
2470}
2471
2472struct Session {
2474 path: Result<crate::plugins::Located, String>,
2475 connection: Result<JsonMap<String, Json>, String>,
2476 cwd: PathBuf,
2477 log: LogSink,
2478 opaque_log: Arc<AtomicBool>,
2481 unmanaged: bool,
2483 read_only: bool,
2484 loaded: bool,
2486 proc_: Option<PluginProcess>,
2487}
2488
2489impl Session {
2490 fn new(
2491 path: Result<crate::plugins::Located, String>,
2492 connection: Result<JsonMap<String, Json>, String>,
2493 cwd: PathBuf,
2494 log: LogSink,
2495 unmanaged: bool,
2496 ) -> Session {
2497 let opaque_log = Arc::new(AtomicBool::new(false));
2498 let protect = Arc::clone(&opaque_log);
2499 let downstream = log;
2500 let log: LogSink = Arc::new(move |plugin, line| {
2501 let line = crate::secrets::mask_source_error(line, protect.load(Ordering::SeqCst));
2502 downstream(plugin, &line);
2503 });
2504 Session {
2505 path,
2506 connection,
2507 cwd,
2508 log,
2509 opaque_log,
2510 unmanaged,
2511 read_only: unmanaged,
2512 loaded: false,
2513 proc_: None,
2514 }
2515 }
2516
2517 fn want_read_only(&mut self, ro: bool) {
2518 if self.read_only != ro {
2519 self.close();
2521 self.read_only = ro;
2522 }
2523 }
2524
2525 fn get(&mut self) -> Result<&mut PluginProcess, String> {
2526 if self.proc_.is_none() {
2527 let plugin = self.path.clone()?;
2528 let connection = self.connection.clone()?;
2529 let mut p = plugin
2530 .start(self.log.clone(), Some(&self.cwd))
2531 .map_err(|e| masked_source_error(&e, false))?;
2532 let ro = self.unmanaged && self.read_only && p.has(CAP_READ_ONLY);
2533 p.open(connection, ro).map_err(|e| {
2534 let protected = masked_source_error(&e, false);
2535 format!("can't open the source connection: {protected}")
2536 })?;
2537 self.proc_ = Some(p);
2538 }
2539 Ok(self.proc_.as_mut().unwrap())
2540 }
2541
2542 fn statement(&mut self, sensitive: bool) -> Result<(&mut PluginProcess, OpaqueLogScope), String> {
2543 let opaque_log = Arc::clone(&self.opaque_log);
2544 let process = self.get()?;
2545 let scope = OpaqueLogScope::new(opaque_log, sensitive);
2546 Ok((process, scope))
2547 }
2548
2549 fn close(&mut self) {
2550 if let Some(p) = self.proc_.take() {
2551 let _ = p.close();
2552 }
2553 }
2554}
2555
2556struct OpaqueLogScope(Arc<AtomicBool>);
2557
2558impl OpaqueLogScope {
2559 fn new(active: Arc<AtomicBool>, sensitive: bool) -> Self {
2560 active.store(sensitive, Ordering::SeqCst);
2561 Self(active)
2562 }
2563}
2564
2565impl Drop for OpaqueLogScope {
2566 fn drop(&mut self) {
2567 self.0.store(false, Ordering::SeqCst);
2568 }
2569}
2570
2571fn nowhere_note(profile: &str, target: &str) -> String {
2573 format!(
2574 "destination `{profile}`: `{target}` delivers nowhere (`deliver: false`); output stays in target/"
2575 )
2576}
2577
2578struct Pool {
2581 connections: Arc<ProfileConnections>,
2582 cwd: PathBuf,
2583 log: LogSink,
2584 unmanaged: bool,
2585 sessions: Mutex<BTreeMap<String, Arc<Mutex<Session>>>>,
2586}
2587
2588impl Pool {
2589 fn new(connections: Arc<ProfileConnections>, cwd: PathBuf, log: LogSink, unmanaged: bool) -> Pool {
2590 Pool {
2591 connections,
2592 cwd,
2593 log,
2594 unmanaged,
2595 sessions: Mutex::default(),
2596 }
2597 }
2598
2599 fn target(&self, name: &str) -> Result<ProfileTarget, String> {
2601 let profiles = &self.connections.profiles;
2602 match profiles.entry(Role::Connection, name) {
2603 Entry::Use(o) => Ok(o.clone()),
2604 Entry::Unknown => Err(format!(
2605 "connection `{name}` isn't under `{}:` in {}",
2606 profiles.section_key(Role::Connection),
2607 profiles.path.display()
2608 )),
2609 Entry::Missing | Entry::Nowhere => Err(profiles.missing_entry(Role::Connection, name)),
2611 }
2612 }
2613
2614 fn kind(&self, name: &str) -> Result<String, String> {
2616 Ok(self.target(name)?.kind)
2617 }
2618
2619 fn session(&self, name: &str) -> Result<Arc<Mutex<Session>>, String> {
2620 if let Some(s) = self.sessions.lock().unwrap().get(name) {
2621 return Ok(s.clone());
2622 }
2623 let output = self.target(name)?;
2624 let id = crate::project::PluginId::new(PluginKind::Source, output.kind.clone());
2625 let path = crate::plugins::locate_in(&self.connections.root, &self.connections.plugins, &id)
2626 .map_err(String::from);
2627 let session = Arc::new(Mutex::new(Session::new(
2628 path,
2629 render_connection(&output),
2630 self.cwd.clone(),
2631 self.log.clone(),
2632 self.unmanaged,
2633 )));
2634 self.sessions
2635 .lock()
2636 .unwrap()
2637 .insert(name.to_string(), session.clone());
2638 Ok(session)
2639 }
2640
2641 fn close_all(&self) {
2642 for s in self.sessions.lock().unwrap().values() {
2643 if let Ok(mut s) = s.lock() {
2644 s.close();
2645 }
2646 }
2647 }
2648}
2649
2650struct PoolRunner {
2653 pool: Arc<Pool>,
2654 default: Option<String>,
2655 log: LogSink,
2656}
2657
2658impl PoolRunner {
2659 fn session(&self, profile: Option<&str>) -> Result<Arc<Mutex<Session>>, String> {
2660 let name = profile.or(self.default.as_deref()).ok_or(
2661 "no connection here: pass `profile=`, or give the report a `profile:` (this isn't a query's SQL)",
2662 )?;
2663 self.pool.session(name)
2664 }
2665}
2666
2667impl QueryRunner for PoolRunner {
2668 fn run_query(&self, sql: &str, max_rows: u64, profile: Option<&str>) -> Result<QueryRows, String> {
2669 let session = self.session(profile)?;
2670 let mut s = session.lock().map_err(|_| "session lock poisoned".to_string())?;
2671 let sensitive = crate::secrets::contains_secret(sql);
2672 if s.unmanaged
2675 && let Some(bad) = sqlsplit::split(sql)
2676 .into_iter()
2677 .find(|st| sqlsplit::classify(&st.text) != StatementKind::Read)
2678 {
2679 return Err(format!(
2680 "run_query() in an unmanaged report may only read, but got `{}`; give the report a YAML to declare it",
2681 masked_sql_summary(&bad.text, 60, sensitive)
2682 ));
2683 }
2684 (self.log)("run_query()", &protected_sql(sql, sensitive));
2685 let (p, _log_scope) = s.statement(sensitive)?;
2686 let mut out = QueryRows::default();
2687 let mut too_many = false;
2688 let exec = p
2689 .execute(sql, Some(max_rows + 1), |schema, batch| {
2690 if out.columns.is_empty() {
2691 out.columns = schema.fields().iter().map(|f| f.name().clone()).collect();
2692 }
2693 if out.rows.len() as u64 + batch.num_rows() as u64 > max_rows {
2694 too_many = true;
2695 return Ok(());
2696 }
2697 out.rows.extend(crate::values::batch_rows(&batch));
2698 Ok(())
2699 })
2700 .map_err(|e| format!("run_query() failed: {}", masked_source_error(&e, sensitive)))?;
2701 if too_many {
2702 return Err(format!(
2703 "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"
2704 ));
2705 }
2706 if let Execution::Result { schema, .. } = exec
2707 && out.columns.is_empty()
2708 {
2709 out.columns = schema.fields().iter().map(|f| f.name().clone()).collect();
2710 }
2711 Ok(out)
2712 }
2713
2714 fn columns(&self, sql: &str, profile: Option<&str>) -> Result<Vec<Column>, String> {
2715 let session = self.session(profile)?;
2716 let mut s = session.lock().map_err(|_| "session lock poisoned".to_string())?;
2717 let sensitive = crate::secrets::contains_secret(sql);
2718 (self.log)("columns()", &protected_sql(sql, sensitive));
2719 let (p, _log_scope) = s.statement(sensitive)?;
2720 let mut schema: Option<SchemaRef> = None;
2721 let exec = p
2722 .execute(sql, Some(1), |sch, _| {
2723 schema.get_or_insert_with(|| sch.clone());
2724 Ok(())
2725 })
2726 .map_err(|e| masked_source_error(&e, sensitive))?;
2727 if let Execution::Result { schema: sch, .. } = exec {
2728 schema.get_or_insert(sch);
2729 }
2730 let schema = schema.ok_or("the query returned no result set")?;
2731 Ok(schema
2732 .fields()
2733 .iter()
2734 .map(|f| Column {
2735 name: f.name().clone(),
2736 data_type: f.data_type().to_string(),
2737 })
2738 .collect())
2739 }
2740
2741 fn load(&self, name: &str, table: &Table) -> Result<Option<(String, Option<String>)>, String> {
2742 let session = self.session(None)?;
2743 let mut s = session.lock().map_err(|_| "session lock poisoned".to_string())?;
2744 s.want_read_only(false);
2746 if !s.get()?.has(CAP_LOAD) {
2747 return Ok(None);
2748 }
2749 let batch = table.to_batch()?;
2750 let schema = batch.schema();
2751 (self.log)(
2752 &format!("lookup `{name}`"),
2753 &format!(
2754 "-- {} rows loaded through the plugin's `load` request",
2755 batch.num_rows()
2756 ),
2757 );
2758 let r = s.get()?.load(name, &schema, [batch]).map_err(|e| e.to_string())?;
2759 s.loaded = true;
2760 Ok(Some(r))
2761 }
2762}
2763
2764fn render_json(renderer: &Renderer, file: &Path, v: &Json) -> Result<Json, RenderError> {
2766 Ok(match v {
2767 Json::String(s) => Json::String(renderer.render(file, s)?),
2768 Json::Array(a) => Json::Array(
2769 a.iter()
2770 .map(|v| render_json(renderer, file, v))
2771 .collect::<Result<_, _>>()?,
2772 ),
2773 Json::Object(o) => Json::Object(
2774 o.iter()
2775 .map(|(k, v)| Ok((k.clone(), render_json(renderer, file, v)?)))
2776 .collect::<Result<_, RenderError>>()?,
2777 ),
2778 other => other.clone(),
2779 })
2780}
2781
2782fn render_connection(output: &ProfileTarget) -> Result<JsonMap<String, Json>, String> {
2784 let mut env = minijinja::Environment::new();
2785 env.set_undefined_behavior(minijinja::UndefinedBehavior::Strict);
2786 env.add_function("env_var", |name: String, default: Option<String>| -> Result<String, minijinja::Error> {
2787 std::env::var(&name).ok().or(default).ok_or_else(|| {
2788 minijinja::Error::new(
2789 minijinja::ErrorKind::UndefinedError,
2790 format!("`env_var('{name}')`: environment variable `{name}` is not set and no default is given"),
2791 )
2792 })
2793 });
2794 fn walk(env: &minijinja::Environment<'_>, v: &Json) -> Result<Json, String> {
2795 Ok(match v {
2796 Json::String(s) if crate::preflight::is_templated(s) => Json::String(
2797 env.render_str(s, ())
2798 .map_err(|e| format!("profiles.yml: {}", e.detail().unwrap_or("render error")))?,
2799 ),
2800 Json::Array(a) => Json::Array(a.iter().map(|x| walk(env, x)).collect::<Result<_, _>>()?),
2801 Json::Object(o) => Json::Object(
2802 o.iter()
2803 .map(|(k, x)| Ok((k.clone(), walk(env, x)?)))
2804 .collect::<Result<_, String>>()?,
2805 ),
2806 other => other.clone(),
2807 })
2808 }
2809 match walk(&env, &Json::Object(output.fields.clone()))? {
2810 Json::Object(o) => Ok(o),
2811 _ => unreachable!(),
2812 }
2813}
2814
2815pub fn find_plugin(
2818 project: &Project,
2819 kind: PluginKind,
2820 name: &str,
2821) -> Result<crate::plugins::Located, crate::plugins::LocateError> {
2822 crate::plugins::locate(project, &crate::project::PluginId::new(kind, name))
2823}
2824
2825struct ProfileConnections {
2829 profiles: Profiles,
2831 root: PathBuf,
2832 plugins: Vec<crate::project::PluginRequirement>,
2833 log: LogSink,
2834}
2835
2836static DESCRIBED: std::sync::LazyLock<Mutex<BTreeMap<DescribeKey, Option<Described>>>> =
2839 std::sync::LazyLock::new(Mutex::default);
2840
2841type DescribeKey = (PathBuf, Role, String);
2843
2844#[derive(Clone)]
2845struct Described {
2846 secrets: Vec<String>,
2847 identifier_quote: Option<String>,
2848 capabilities: Vec<String>,
2849 message_limit: Option<u64>,
2851}
2852
2853const SECRET_NAME_WORDS: &[&str] = &["password", "secret", "token", "key", "credential"];
2854
2855impl ProfileConnections {
2856 fn view(&self, role: Role, name: &str) -> Result<Connection, String> {
2857 let target = self.profiles.target_of(role, name);
2858 let out: ProfileTarget = match self.profiles.entry(role, name) {
2859 Entry::Use(o) => o.clone(),
2860 Entry::Nowhere => {
2861 return Err(format!(
2862 "destination `{name}`: `{target}` delivers nowhere (`deliver: false`), so it has no settings"
2863 ));
2864 }
2865 Entry::Missing => return Err(self.profiles.missing_entry(role, name)),
2866 Entry::Unknown => {
2867 return Err(format!(
2868 "no {} profile `{name}` in {}",
2869 role.as_str(),
2870 self.profiles.path.display()
2871 ));
2872 }
2873 };
2874 let mut secrets: Vec<String> = out
2875 .fields
2876 .iter()
2877 .filter(|(_, v)| v.as_str().is_some_and(|s| s.contains(crate::secrets::PREFIX)))
2878 .map(|(k, _)| k.clone())
2879 .collect();
2880 match self.described(role, &out.kind) {
2881 Some(d) => secrets.extend(d.secrets),
2882 None => secrets.extend(
2883 out.fields
2884 .keys()
2885 .filter(|k| {
2886 let k = k.to_lowercase();
2887 SECRET_NAME_WORDS.iter().any(|w| k.contains(w))
2888 })
2889 .cloned(),
2890 ),
2891 }
2892 let fields = render_connection(&out)?;
2893 Ok(Connection {
2894 profile: name.to_string(),
2895 target,
2896 kind: out.kind.clone(),
2897 fields,
2898 secrets,
2899 })
2900 }
2901
2902 fn described(&self, role: Role, kind: &str) -> Option<Described> {
2904 let key = (self.root.clone(), role, kind.to_string());
2905 if let Some(v) = DESCRIBED.lock().unwrap().get(&key) {
2906 return v.clone();
2907 }
2908 let plugin_kind = match role {
2909 Role::Connection => PluginKind::Source,
2910 Role::Destination => PluginKind::Destination,
2911 };
2912 let asked = (|| {
2913 if kind == LOCAL_TYPE {
2914 return Some(Described {
2915 secrets: Vec::new(),
2916 identifier_quote: None,
2917 capabilities: Vec::new(),
2918 message_limit: None,
2919 });
2920 }
2921 let id = crate::project::PluginId::new(plugin_kind, kind);
2922 let plugin = crate::plugins::locate_in(&self.root, &self.plugins, &id).ok()?;
2923 let mut p = plugin.start(self.log.clone(), Some(&self.root)).ok()?;
2924 let capabilities = p.info().capabilities.clone();
2925 let d = p.description().ok();
2926 let _ = p.close();
2927 let d = d?;
2928 Some(Described {
2929 secrets: d
2930 .connection_fields
2931 .into_iter()
2932 .filter(|f| f.secret)
2933 .map(|f| f.name)
2934 .collect(),
2935 identifier_quote: d.identifier_quote,
2936 capabilities,
2937 message_limit: d.message_limit,
2938 })
2939 })();
2940 DESCRIBED.lock().unwrap().insert(key, asked.clone());
2941 asked
2942 }
2943}
2944
2945impl Connections for ProfileConnections {
2946 fn profile(&self, name: &str, role: Option<&str>) -> Result<Connection, String> {
2947 let role = match role {
2948 Some("connection") => Role::Connection,
2949 Some(_) => Role::Destination,
2950 None => {
2951 let c = self.profiles.get(Role::Connection, name).is_some();
2952 let d = self.profiles.get(Role::Destination, name).is_some();
2953 match (c, d) {
2954 (true, true) => {
2955 return Err(format!(
2956 "`{name}` is both a connection and a destination profile; say which with `profile('{name}', role='connection')` or `role='destination'`"
2957 ));
2958 }
2959 (true, false) => Role::Connection,
2960 _ => Role::Destination,
2961 }
2962 }
2963 };
2964 self.view(role, name)
2965 }
2966
2967 fn identifier_quote(&self, kind: &str) -> Result<Option<String>, String> {
2968 match self.described(Role::Connection, kind) {
2969 Some(d) => Ok(d.identifier_quote),
2970 None => Err(format!(
2971 "the `{kind}` source plugin couldn't be asked for its identifier quote character"
2972 )),
2973 }
2974 }
2975}
2976
2977fn check_sheet_name(n: &str) -> Result<(), String> {
2978 if n.is_empty() || n.chars().count() > 31 {
2979 return Err(format!("sheet name `{n}` must be 1 to 31 characters long"));
2980 }
2981 if let Some(c) = n.chars().find(|c| "[]:*?/\\".contains(*c)) {
2982 return Err(format!(
2983 "sheet name `{n}` contains `{c}`, which Excel doesn't allow in sheet names"
2984 ));
2985 }
2986 if n.starts_with('\'') || n.ends_with('\'') {
2987 return Err(format!("sheet name `{n}` can't start or end with an apostrophe"));
2988 }
2989 if n.eq_ignore_ascii_case("history") {
2990 return Err("`History` is reserved by Excel and can't be a sheet name".into());
2991 }
2992 Ok(())
2993}
2994
2995fn is_single_table(format: &str) -> bool {
2996 matches!(format, "csv" | "delimited" | "fixed_width" | "parquet")
2997}
2998
2999fn extension(format: &str) -> &str {
3000 match format {
3001 "delimited" | "fixed_width" => "txt",
3002 crate::project::MESSAGE_FORMAT => "md",
3003 f => f,
3004 }
3005}
3006
3007fn split_ext(name: &str) -> (&str, &str) {
3008 match name.rfind('.') {
3009 Some(i) if i > 0 => (&name[..i], &name[i..]),
3010 _ => (name, ""),
3011 }
3012}
3013
3014fn rel(root: &Path, p: &Path) -> PathBuf {
3015 crate::slash(p.strip_prefix(root).unwrap_or(p))
3016}
3017
3018pub fn rfc3339(t: chrono::DateTime<chrono::Utc>) -> String {
3020 t.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
3021}
3022
3023fn record_path(project: &Project, p: &Path) -> PathBuf {
3027 match p.strip_prefix(&project.root) {
3028 Ok(r) => crate::slash(r),
3029 Err(_) => rel(&project.target_dir, p),
3030 }
3031}
3032
3033impl BindingRun<'_> {
3034 fn summary_line(&self) -> String {
3036 let rows: u64 = self.produced.iter().map(|p| p.rows).sum();
3037 let mut parts = Vec::new();
3038 if !self.produced.is_empty() {
3039 let n = self.produced.len();
3040 parts.push(format!(
3041 "{n} result set{}, {} row{}",
3042 if n == 1 { "" } else { "s" },
3043 thousands(rows),
3044 if rows == 1 { "" } else { "s" }
3045 ));
3046 }
3047 let names: Vec<String> = self
3048 .files()
3049 .map(|(f, _)| f.file_name().unwrap_or_default().to_string_lossy().to_string())
3050 .collect();
3051 if !names.is_empty() {
3052 parts.push(format!("→ {}", names.join(", ")));
3053 }
3054 let delivered: Vec<&str> = self
3056 .outs
3057 .iter()
3058 .flat_map(|o| o.deliveries.iter())
3059 .filter_map(|d| d.get("location").and_then(Json::as_str))
3060 .collect();
3061 let notes: Vec<&str> = self
3062 .outs
3063 .iter()
3064 .filter(|o| !o.dests.is_empty())
3065 .filter_map(|o| o.delivery_note.as_deref())
3066 .collect();
3067 if !delivered.is_empty() {
3068 parts.push(format!("→ {}", delivered.join(", ")));
3069 } else if !notes.is_empty() && self.opts.preview.is_none() {
3070 parts.push(format!("({})", notes.join("; ")));
3071 }
3072 parts.join(" ")
3073 }
3074
3075 fn files(&self) -> impl Iterator<Item = &(PathBuf, Option<String>)> {
3077 self.outs.iter().flat_map(|o| o.files.iter())
3078 }
3079}
3080
3081pub fn thousands(n: u64) -> String {
3083 let s = n.to_string();
3084 let mut out = String::new();
3085 for (i, c) in s.chars().enumerate() {
3086 if i > 0 && (s.len() - i).is_multiple_of(3) {
3087 out.push(',');
3088 }
3089 out.push(c);
3090 }
3091 out
3092}
3093
3094fn human_bytes(n: u64) -> String {
3095 match n {
3096 n if n < 1024 => format!("{n} B"),
3097 n if n < 1024 * 1024 => format!("{:.1} KB", n as f64 / 1024.0),
3098 n => format!("{:.1} MB", n as f64 / (1024.0 * 1024.0)),
3099 }
3100}
3101
3102pub fn clean(t: &Path) -> std::io::Result<bool> {
3104 if t.exists() {
3105 std::fs::remove_dir_all(t)?;
3106 Ok(true)
3107 } else {
3108 Ok(false)
3109 }
3110}