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_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::parse::ParsedBinding;
28use crate::profiles::{Entry, LOCAL_TYPE, ProfileTarget, Profiles, Role};
29use crate::project::{Binding, PluginKind, Project, QueryEntry, Report};
30use crate::render::{
31 Column, Connection, Connections, Mode, QueryRows, QueryRunner, RenderError, Renderer, RendererConfig,
32};
33use crate::selector;
34use crate::sqlsplit::{self, StatementKind};
35
36#[derive(Debug, Clone, Default)]
38pub struct RunOptions {
39 pub selector: Option<String>,
40 pub set: Option<String>,
42 pub target: Option<String>,
44 pub profile: Option<String>,
46 pub vars: BTreeMap<String, String>,
47 pub output_name: Option<String>,
48 pub output_path: Option<String>,
49 pub dry_run: bool,
50 pub preview: Option<u64>,
52 pub accept_schema_change: bool,
53 pub interactive: bool,
55 pub date: Option<NaiveDate>,
57 pub scheduled_at: Option<chrono::DateTime<chrono::Utc>>,
60 pub live_check: bool,
62 pub schedule: Option<String>,
64 pub timezone: Option<String>,
66 pub manifest_checksum: Option<String>,
68}
69
70impl RunOptions {
71 pub fn params(&self, date: NaiveDate) -> Json {
73 json!({
74 "selector": self.selector,
75 "set": self.set,
76 "schedule": self.schedule,
77 "target": self.target,
78 "profile": self.profile,
79 "vars": self.vars,
80 "run_date": date.to_string(),
81 "scheduled_at": self.scheduled_at.map(rfc3339),
82 "timezone": self.timezone,
83 "output_name": self.output_name,
84 "output_path": self.output_path,
85 "dry_run": self.dry_run,
86 "preview": self.preview,
87 "accept_schema_change": self.accept_schema_change,
88 })
89 }
90}
91
92#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
94pub enum Level {
95 Debug,
96 Info,
97}
98
99pub trait Ui {
102 fn targets(&mut self, _targets: &RunTargets) {}
104 fn plan(&mut self, _bindings: usize) {}
106 fn binding_start(&mut self, _report: &str, _set: Option<&str>) {}
107 fn binding_vars(
109 &mut self,
110 _schedule: Option<&str>,
111 _schedule_vars: Option<&JsonMap<String, Json>>,
112 _vars: &JsonMap<String, Json>,
113 ) {
114 }
115 fn step(&mut self, level: Level, verb: &str, detail: &str, elapsed: Option<Duration>);
117 fn warn(&mut self, msg: &str);
118 fn binding_end(&mut self, _outcome: &BindingOutcome) {}
119 fn compiled(&mut self, _plan: &BindingPlan) {}
121 fn choose_set(&mut self, report: &str, sets: &[String]) -> Result<Option<String>, String>;
123 fn plugin_log(&self) -> LogSink;
125 fn sql_log(&self) -> LogSink {
127 Arc::new(|_, _| {})
128 }
129}
130
131#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
132#[serde(rename_all = "snake_case")]
133pub enum Status {
134 Success,
135 Error,
136 DryRun,
137 Checked,
138}
139
140#[derive(Debug, Clone, Serialize)]
141pub struct BindingOutcome {
142 pub report: String,
143 #[serde(skip_serializing_if = "Option::is_none")]
144 pub set: Option<String>,
145 pub binding: String,
146 pub status: Status,
147 #[serde(skip_serializing_if = "Option::is_none")]
148 pub error: Option<String>,
149 pub files: Vec<PathBuf>,
150 pub summary: String,
152 pub schedule: Option<String>,
154 pub schedule_vars: Option<JsonMap<String, Json>>,
155 pub vars: JsonMap<String, Json>,
157 #[serde(skip_serializing_if = "String::is_empty")]
159 pub timezone: String,
160 #[serde(skip)]
161 pub elapsed: Duration,
162}
163
164#[derive(Debug, Clone, Serialize)]
166pub struct BindingPlan {
167 pub report: String,
168 pub set: Option<String>,
169 pub compiled: Vec<PathBuf>,
171 pub profile: Option<String>,
173 pub queries: Vec<PlannedQuery>,
175 pub target: String,
176 pub format: String,
177 pub output: PathBuf,
180 pub destinations: Vec<PlannedDestination>,
181 pub schedules: Vec<String>,
182}
183
184#[derive(Debug, Clone, Serialize)]
185pub struct PlannedQuery {
186 pub query: String,
187 pub connection: String,
188 #[serde(rename = "type")]
189 pub kind: String,
190 pub target: String,
192 #[serde(skip_serializing_if = "Vec::is_empty")]
193 pub sources: Vec<String>,
194}
195
196#[derive(Debug, Clone, Serialize)]
197pub struct PlannedDestination {
198 pub profile: String,
199 pub kind: Option<String>,
201 pub target: Option<String>,
203 pub path: Option<String>,
204 pub delivers: bool,
206}
207
208#[derive(Debug, Default)]
209pub struct RunSummary {
210 pub outcomes: Vec<BindingOutcome>,
211 pub error: Option<String>,
213 pub missing_entry: bool,
215}
216
217impl RunSummary {
218 pub fn failed(&self) -> bool {
219 self.error.is_some() || self.outcomes.iter().any(|o| o.status == Status::Error)
220 }
221}
222
223pub fn run(project: &Project, opts: &RunOptions, ui: &mut dyn Ui) -> RunSummary {
224 let mut summary = RunSummary::default();
225 if let Err(e) = crate::target::ensure(&project.target_dir) {
226 summary.error = Some(format!(
227 "can't write to the target path {} (from {}): {e}",
228 project.target_dir.display(),
229 project.target_source
230 ));
231 return summary;
232 }
233 let planned: Vec<(&Report, Binding)> = match &opts.schedule {
234 Some(name) => {
235 if !project.schedules.iter().any(|e| &e.name == name) {
236 summary.error = Some(unknown_schedule(project, name));
237 return summary;
238 }
239 match schedule_bindings(project, name, opts) {
240 Ok(p) => p,
241 Err(e) => {
242 summary.error = Some(e);
243 return summary;
244 }
245 }
246 }
247 None => {
248 let reports: Vec<&Report> = match &opts.selector {
249 None => project.reports.iter().collect(),
250 Some(s) => match selector::resolve(project, s) {
251 Ok(r) if r.is_empty() => {
252 summary.error = Some(format!("selector `{s}` matches no report"));
253 return summary;
254 }
255 Ok(r) => r,
256 Err(e) => {
257 summary.error = Some(e.to_string());
258 return summary;
259 }
260 },
261 };
262 let mut planned = Vec::new();
265 for report in reports {
266 match choose_bindings(project, report, opts, ui) {
267 Ok(bs) => planned.extend(bs.into_iter().map(|b| (report, b))),
268 Err(e) => {
269 let outcome = BindingOutcome {
270 report: report.name.clone(),
271 set: None,
272 binding: "-".into(),
273 status: Status::Error,
274 error: Some(e),
275 files: Vec::new(),
276 summary: String::new(),
277 schedule: opts.schedule.clone(),
278 schedule_vars: None,
279 vars: JsonMap::new(),
280 timezone: String::new(),
281 elapsed: Duration::ZERO,
282 };
283 ui.binding_end(&outcome);
284 summary.outcomes.push(outcome);
285 }
286 }
287 }
288 planned
289 }
290 };
291 match run_targets(project, &planned, opts) {
293 Ok(t) => ui.targets(&t),
294 Err(e) => {
295 summary.error = Some(e);
296 summary.missing_entry = true;
297 return summary;
298 }
299 }
300 ui.plan(planned.len());
301 for (report, b) in &planned {
302 ui.binding_start(&report.name, b.set.as_deref());
303 let mut r = BindingRun::new(project, report, b, opts, ui);
304 let outcome = r.run();
305 ui.binding_end(&outcome);
306 summary.outcomes.push(outcome);
307 }
308 summary
309}
310
311#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
313pub struct UsedProfile {
314 pub role: &'static str,
316 pub profile: String,
317 pub target: String,
319 pub deliver: bool,
321}
322
323#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
325pub struct RunTargets {
326 pub name: String,
328 pub from: String,
330 pub profiles: Vec<UsedProfile>,
331}
332
333impl RunTargets {
334 pub fn line(&self) -> String {
336 let mut out = format!("{} ({})", self.name, self.from);
337 for p in self.profiles.iter().filter(|p| p.target != self.name) {
338 out.push_str(&format!("; {} `{}`: {}", p.role, p.profile, p.target));
339 }
340 out
341 }
342
343 pub fn mismatch(&self) -> Option<String> {
346 let first = &self.profiles.first()?.target;
347 if first == &self.name || self.profiles.iter().any(|p| &p.target != first) {
348 return None;
349 }
350 Some(format!(
351 "every profile is on `{first}` but the run's target is `{}`: pass `--target {first}` or set `{}`",
352 self.name,
353 crate::profiles::TARGET_ENV
354 ))
355 }
356}
357
358pub fn run_targets(
362 project: &Project,
363 planned: &[(&Report, Binding)],
364 opts: &RunOptions,
365) -> Result<RunTargets, String> {
366 let profiles = &project.profiles;
367 let schedule_vars = opts
368 .schedule
369 .as_ref()
370 .and_then(|n| project.schedules.iter().find(|e| &e.name == n))
371 .map(|e| e.vars.clone());
372 let inputs = crate::parse::Inputs {
373 target: project.target_name.clone(),
374 cli_vars: opts.vars.clone(),
375 date: opts.date,
376 scheduled_at: opts.scheduled_at,
377 timezone: opts.timezone.clone(),
378 schedule: opts.schedule.clone(),
379 started_at: None,
380 };
381 let mut used: Vec<(Role, String)> = Vec::new();
382 for (report, b) in planned {
383 let mut vars = b.vars.clone();
384 vars.extend(schedule_vars.clone().unwrap_or_default());
385 let parsed = crate::parse::binding(project, report, b, &vars, &inputs);
387 let names = parsed
388 .connections()
389 .into_iter()
390 .map(|c| (Role::Connection, c.to_string()))
391 .chain(
392 parsed
393 .destinations
394 .iter()
395 .flatten()
396 .filter(|d| !profiles.is_builtin_local(d))
397 .map(|d| (Role::Destination, d.clone())),
398 );
399 for u in names {
400 if !used.contains(&u) {
401 used.push(u);
402 }
403 }
404 }
405 let mut missing = Vec::new();
406 let mut out = Vec::new();
407 for (role, name) in &used {
408 let deliver = match profiles.entry(*role, name) {
409 Entry::Use(_) => true,
410 Entry::Nowhere => false,
411 Entry::Missing => {
412 missing.push(profiles.missing_entry(*role, name));
413 continue;
414 }
415 Entry::Unknown => continue,
417 };
418 out.push(UsedProfile {
419 role: role.as_str(),
420 profile: name.clone(),
421 target: profiles.target_of(*role, name),
422 deliver,
423 });
424 }
425 match missing.len() {
426 0 => Ok(RunTargets {
427 name: project.target_name.clone(),
428 from: project.target_from.to_string(),
429 profiles: out,
430 }),
431 1 => Err(format!("{}; nothing was run", missing[0])),
432 n => Err(format!(
433 "{n} profiles this run uses have no entry for it; nothing was run:\n {}",
434 missing.join("\n ")
435 )),
436 }
437}
438
439fn schedule_bindings<'a>(
442 project: &'a Project,
443 name: &str,
444 opts: &RunOptions,
445) -> Result<Vec<(&'a Report, Binding)>, String> {
446 let reports: Option<Vec<&str>> = match &opts.selector {
447 None => None,
448 Some(s) => Some(
449 selector::resolve(project, s)
450 .map_err(|e| e.to_string())?
451 .into_iter()
452 .map(|r| r.name.as_str())
453 .collect(),
454 ),
455 };
456 let set = opts.set.as_deref().filter(|s| *s != "all");
457 let all: Vec<(&Report, &Binding)> = project
458 .reports
459 .iter()
460 .flat_map(|r| {
461 r.bindings
462 .iter()
463 .filter(|b| b.schedules.iter().any(|s| s == name))
464 .map(move |b| (r, b))
465 })
466 .collect();
467 let picked: Vec<(&Report, Binding)> = all
468 .iter()
469 .filter(|(r, _)| reports.as_ref().is_none_or(|rs| rs.contains(&r.name.as_str())))
470 .filter(|(_, b)| set.is_none() || b.set.as_deref() == set)
471 .map(|(r, b)| {
472 let mut b = (*b).clone();
473 if let Some(p) = &opts.profile {
474 b.profile = Some(p.clone());
475 b.profile_at = Some(profile_flag_at());
476 }
477 (*r, b)
478 })
479 .collect();
480 if picked.is_empty() && (reports.is_some() || set.is_some()) {
481 let mut asked = Vec::new();
482 if let Some(s) = &opts.selector {
483 asked.push(format!("`{s}`"));
484 }
485 if let Some(s) = set {
486 asked.push(format!("Set `{s}`"));
487 }
488 let runs: Vec<String> = all
489 .iter()
490 .map(|(r, b)| match &b.set {
491 Some(s) => format!("{}/{s}", r.name),
492 None => r.name.clone(),
493 })
494 .collect();
495 return Err(format!(
496 "schedule `{name}` doesn't run {}; it runs: {}",
497 asked.join(" with "),
498 if runs.is_empty() {
499 "nothing".to_string()
500 } else {
501 runs.join(", ")
502 }
503 ));
504 }
505 Ok(picked)
506}
507
508pub fn unknown_schedule(project: &Project, name: &str) -> String {
510 let names: Vec<&str> = project.schedules.iter().map(|e| e.name.as_str()).collect();
511 if names.is_empty() {
512 format!("no schedule `{name}`: this project's schedules.yml declares none")
513 } else {
514 format!("no schedule `{name}`; valid names: {}", names.join(", "))
515 }
516}
517
518pub fn choose_bindings(
520 project: &Project,
521 report: &Report,
522 opts: &RunOptions,
523 ui: &mut dyn Ui,
524) -> Result<Vec<Binding>, String> {
525 let mut out: Vec<Binding> = match opts.set.as_deref() {
526 Some("all") => report.bindings.clone(),
527 Some(name) => match report.binding(name) {
528 Some(b) => vec![b.clone()],
529 None => vec![ad_hoc(project, report, name)],
530 },
531 None if !report.has_sets => report.bindings.clone(),
532 None => {
533 let names: Vec<String> = report.bindings.iter().filter_map(|b| b.set.clone()).collect();
534 let pick = report
535 .default_set
536 .clone()
537 .or_else(|| project.default_set.clone().filter(|d| names.contains(d)))
538 .or_else(|| (names.len() == 1).then(|| names[0].clone()));
539 match pick {
540 Some(p) => report.binding(&p).cloned().into_iter().collect(),
541 None if opts.interactive => match ui.choose_set(&report.name, &names)? {
542 Some(p) => report.binding(&p).cloned().into_iter().collect(),
543 None => report.bindings.clone(),
544 },
545 None => {
546 return Err(format!(
547 "report `{}` has several Sets ({}) and no `default_set`; declare `default_set:` or pass `--set <name>` or `--set all`",
548 report.name,
549 names.join(", ")
550 ));
551 }
552 }
553 }
554 };
555 if let Some(p) = &opts.profile {
556 for b in &mut out {
557 b.profile = Some(p.clone());
558 b.profile_at = Some(profile_flag_at());
559 }
560 }
561 Ok(out)
562}
563
564fn ad_hoc(project: &Project, report: &Report, name: &str) -> Binding {
567 let mut b = report.base.clone();
568 b.set = Some(name.to_string());
569 if let Some(reg) = project.sets.get(name) {
570 if let Some(p) = ®.profile {
571 b.profile = Some(p.clone());
572 b.profile_at = Some(crate::project::ProfileAt {
573 file: reg.file.clone(),
574 line: reg.line,
575 key: format!("`profile` of Set `{name}`"),
576 });
577 }
578 b.vars.extend(reg.vars.clone());
579 }
580 b
581}
582
583fn profile_flag_at() -> crate::project::ProfileAt {
585 crate::project::ProfileAt {
586 file: PathBuf::from("--profile"),
587 line: None,
588 key: "`--profile`".into(),
589 }
590}
591
592struct Produced {
594 query: String,
595 connection: String,
596 schema: SchemaRef,
597 rows: u64,
598 spool: PathBuf,
599 name: String,
600 anchor: Option<String>,
601 header: Option<bool>,
602 columns: BTreeMap<String, ColumnOptions>,
603}
604
605struct Statement {
606 query: String,
607 connection: String,
609 file: PathBuf,
610 line: usize,
611 text: String,
612 kind: StatementKind,
613 sensitive: bool,
616 tab: bool,
619}
620
621struct BindingRun<'a> {
622 project: &'a Project,
623 report: &'a Report,
624 b: &'a Binding,
625 vars: JsonMap<String, Json>,
627 rendered_vars: JsonMap<String, Json>,
629 schedule_vars: Option<JsonMap<String, Json>>,
630 opts: &'a RunOptions,
631 date: NaiveDate,
633 calendar: Calendar,
634 ui: &'a mut dyn Ui,
635 compiled_dir: PathBuf,
636 run_dir: PathBuf,
637 schema_dir: PathBuf,
638 target: String,
640 parsed: ParsedBinding,
642 pool: Option<Arc<Pool>>,
644 renderers: BTreeMap<Option<String>, Arc<Renderer>>,
646 started: Instant,
647 started_at: chrono::DateTime<chrono::Utc>,
648 produced: Vec<Produced>,
649 files: Vec<(PathBuf, Option<String>)>,
650 dests: Vec<RenderedDest>,
652 deliveries: Vec<Json>,
654 delivery_note: Option<String>,
655 drift: Vec<String>,
656 protected_snapshot: Option<Json>,
658}
659
660struct RenderedDest {
662 profile: String,
663 path: Option<String>,
664 options: JsonMap<String, Json>,
665}
666
667type Fail = String;
668
669fn masked_source_error(error: &HostError, sensitive: bool) -> String {
670 let error = error.to_string();
671 crate::secrets::mask_source_error(&error, sensitive).into_owned()
672}
673
674fn masked_sql_summary(sql: &str, max: usize, sensitive: bool) -> String {
675 if sensitive {
676 crate::secrets::MASK.to_string()
677 } else {
678 dre_protocol::util::summarize(&crate::secrets::mask(sql), max)
679 }
680}
681
682fn protected_sql(sql: &str, sensitive: bool) -> std::borrow::Cow<'_, str> {
683 if sensitive {
684 std::borrow::Cow::Borrowed(crate::secrets::MASK)
685 } else {
686 crate::secrets::mask(sql)
687 }
688}
689
690impl<'a> BindingRun<'a> {
691 fn new(
692 project: &'a Project,
693 report: &'a Report,
694 b: &'a Binding,
695 opts: &'a RunOptions,
696 ui: &'a mut dyn Ui,
697 ) -> Self {
698 let t = &project.target_dir;
699 let rel = Path::new(&report.name).join(b.dir_name());
700 let schedule = opts
701 .schedule
702 .as_ref()
703 .and_then(|n| project.schedules.iter().find(|e| &e.name == n));
704 let schedule_vars = schedule.map(|e| e.vars.clone());
705 let calendar: Calendar = crate::parse::calendar(
707 project,
708 report,
709 opts.timezone.as_deref(),
710 opts.schedule.as_deref(),
711 );
712 let date = crate::parse::run_date(&calendar, opts.date, opts.scheduled_at);
713 let mut vars = b.vars.clone();
715 vars.extend(schedule_vars.clone().unwrap_or_default());
716 let rendered_vars = {
717 let mut v = vars.clone();
718 v.extend(
719 opts.vars
720 .iter()
721 .map(|(k, x)| (k.clone(), Json::String(x.clone()))),
722 );
723 v
724 };
725 ui.binding_vars(opts.schedule.as_deref(), schedule_vars.as_ref(), &rendered_vars);
726 BindingRun {
727 project,
728 report,
729 b,
730 vars,
731 rendered_vars,
732 schedule_vars,
733 opts,
734 date,
735 calendar,
736 ui,
737 compiled_dir: t.join("compiled").join(&rel),
738 run_dir: t.join("run").join(&rel),
739 schema_dir: t.join("schema").join(&rel),
740 target: project.target_name.clone(),
741 parsed: ParsedBinding::default(),
742 pool: None,
743 renderers: BTreeMap::new(),
744 started: Instant::now(),
745 started_at: chrono::Utc::now(),
746 produced: Vec::new(),
747 files: Vec::new(),
748 dests: Vec::new(),
749 deliveries: Vec::new(),
750 delivery_note: None,
751 drift: Vec::new(),
752 protected_snapshot: None,
753 }
754 }
755
756 fn connections(&self) -> Arc<ProfileConnections> {
758 Arc::new(ProfileConnections {
759 profiles: self.project.profiles.clone(),
760 root: self.project.root.clone(),
761 plugins: self.project.plugins.clone(),
762 log: self.ui.plugin_log(),
763 })
764 }
765
766 fn renderer(&mut self, connection: Option<&str>) -> Result<Arc<Renderer>, Fail> {
769 let key = connection.map(str::to_string);
770 if let Some(r) = self.renderers.get(&key) {
771 return Ok(r.clone());
772 }
773 let pool = self.pool.clone().expect("the pool is made before rendering");
774 let name = key.clone().or_else(|| self.parsed.inherited.clone());
776 let source_type = match (&key, &name) {
777 (Some(n), _) => pool.kind(n)?,
778 (None, Some(n)) => pool.kind(n).unwrap_or_default(),
780 (None, None) => String::new(),
781 };
782 let context =
783 crate::parse::run_context(self.project, self.report, self.b, &self.inputs(), self.started_at);
784 let limited = Arc::new(crate::render::Limited::new(
785 context.clone(),
786 self.vars.clone(),
787 self.opts.vars.clone(),
788 ));
789 let r = Renderer::new(RendererConfig {
790 root: &self.project.root,
791 macros: &self.project.macros,
792 context,
793 vars: self.vars.clone(),
794 cli_vars: self.opts.vars.clone(),
795 runner: Some(Arc::new(PoolRunner {
796 pool: pool.clone(),
797 default: name.clone(),
798 log: self.ui.sql_log(),
799 })),
800 connections: Some(pool.connections.clone()),
801 mode: Mode::Run,
802 connection: name,
803 source_type,
804 sources: crate::parse::resolver(self.project, limited),
805 run_query_max_rows: self.project.run_query_max_rows,
806 sql: self.project.sql.clone(),
807 lookups: self.project.lookups.clone(),
808 lookup_inline_max_rows: self.project.lookup_inline_max_rows,
809 packages: self.project.packages.clone(),
810 project_name: self.project.name.clone(),
811 dispatch: self.project.dispatch.clone(),
812 })
813 .map_err(|e| e.to_string())?;
814 let r = Arc::new(r);
815 self.renderers.insert(key, r.clone());
816 Ok(r)
817 }
818
819 fn inputs(&self) -> crate::parse::Inputs {
821 crate::parse::Inputs {
822 target: self.target.clone(),
823 cli_vars: self.opts.vars.clone(),
824 date: self.opts.date,
825 scheduled_at: self.opts.scheduled_at,
826 timezone: self.opts.timezone.clone(),
827 schedule: self.opts.schedule.clone(),
828 started_at: Some(self.started_at),
829 }
830 }
831
832 fn connection_of(&self, query: &str) -> Result<String, Fail> {
834 self.parsed
835 .query(query)
836 .and_then(|q| q.connection.clone())
837 .ok_or_else(|| format!("query `{query}` has no connection"))
838 }
839
840 fn outcome(&self, status: Status, error: Option<String>) -> BindingOutcome {
841 BindingOutcome {
842 report: self.report.name.clone(),
843 set: self.b.set.clone(),
844 binding: self.b.dir_name().to_string(),
845 status,
846 error,
847 files: self.files.iter().map(|(p, _)| p.clone()).collect(),
848 summary: self.summary_line(),
849 schedule: self.opts.schedule.clone(),
850 schedule_vars: self.schedule_vars.clone(),
851 vars: self.rendered_vars.clone(),
852 timezone: self.calendar.tz.name().to_string(),
853 elapsed: self.started.elapsed(),
854 }
855 }
856
857 fn run(&mut self) -> BindingOutcome {
858 let dry = self.opts.dry_run || self.opts.live_check;
859 if !dry {
860 let _ = std::fs::remove_dir_all(&self.run_dir);
861 }
862 let result = self.run_inner();
863 let status = match (&result, dry, self.opts.live_check) {
864 (Err(_), _, _) => Status::Error,
865 (Ok(()), _, true) => Status::Checked,
866 (Ok(()), true, _) => Status::DryRun,
867 (Ok(()), false, _) => Status::Success,
868 };
869 let err = result.err();
870 if !dry && let Err(e) = self.write_results(&status, err.as_deref()) {
871 self.ui.warn(&format!("can't write run_results.json: {e}"));
872 }
873 let _ = std::fs::remove_dir_all(self.run_dir.join(".spool"));
875 self.outcome(status, err)
876 }
877
878 fn run_inner(&mut self) -> Result<(), Fail> {
879 self.parsed = crate::parse::binding(self.project, self.report, self.b, &self.vars, &self.inputs());
881 if !self.parsed.errors.is_empty() {
882 let msgs: Vec<String> = self
883 .parsed
884 .errors
885 .iter()
886 .map(|p| match p.line {
887 Some(l) => format!("{}:{l}: {}", p.file.display(), p.message),
888 None => format!("{}: {}", p.file.display(), p.message),
889 })
890 .collect();
891 return Err(msgs.join("\n "));
892 }
893 let pool = Arc::new(Pool::new(
896 self.connections(),
897 self.project.root.clone(),
898 self.ui.plugin_log(),
899 !self.report.managed,
900 ));
901 self.pool = Some(pool.clone());
902
903 if !self.report.managed {
904 self.ui.warn(&format!(
905 "`{}` is an unmanaged report ({}): for quick tests only — add a YAML to make it a managed report",
906 self.report.name,
907 self.report.file.display()
908 ));
909 }
910
911 std::fs::create_dir_all(&self.compiled_dir).map_err(|e| e.to_string())?;
913 let mut per_connection: BTreeMap<String, usize> = BTreeMap::new();
915 for q in &self.b.queries {
916 *per_connection.entry(self.connection_of(&q.query)?).or_default() += 1;
917 }
918 let interleave = !self.opts.dry_run && !self.opts.live_check && self.report.managed;
922 let mut statements = Vec::new();
923 let mut two_tabs: Option<String> = None;
925 if interleave {
926 self.dests = self.render_destinations()?;
927 let mut checked: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
928 for (c, n) in &per_connection {
929 if *n > 1 {
930 self.check_sessions(&pool, c, *n)?;
931 checked.insert(c.clone());
932 }
933 }
934 std::fs::create_dir_all(self.run_dir.join(".spool")).map_err(|e| e.to_string())?;
935 let mut i = 0;
936 let queries = self.b.queries.clone();
937 for q in &queries {
938 let conn = self.connection_of(&q.query)?;
939 let renderer = self.renderer(Some(&conn))?;
940 let sts = self.render_query(&renderer, q, &conn, &mut two_tabs)?;
941 for w in renderer.take_warnings() {
942 self.ui.warn(&w);
943 }
944 if let Some(e) = two_tabs.take() {
945 return Err(e);
946 }
947 if sts.len() > 1 && checked.insert(conn.clone()) {
948 self.check_sessions(&pool, &conn, sts.len())?;
949 }
950 let session = pool.session(&conn)?;
951 for st in &sts {
952 self.execute_statement(&session, i, st)?;
953 i += 1;
954 }
955 }
956 } else {
957 let queries = self.b.queries.clone();
958 for q in &queries {
959 let conn = self.connection_of(&q.query)?;
960 let renderer = self.renderer(Some(&conn))?;
961 statements.extend(self.render_query(&renderer, q, &conn, &mut two_tabs)?);
962 }
963 self.dests = self.render_destinations()?;
964 for r in self.renderers.values() {
965 for w in r.take_warnings() {
966 self.ui.warn(&w);
967 }
968 }
969
970 if !self.report.managed {
972 if let Some(bad) = statements.iter().find(|s| !s.kind.is_read_only_safe()) {
973 return Err(format!(
974 "{}:{}: 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",
975 bad.file.display(),
976 bad.line,
977 self.report.name,
978 masked_sql_summary(&bad.text, 60, bad.sensitive)
979 ));
980 }
981 for conn in per_connection.keys() {
982 let session = pool.session(conn)?;
983 let creates_temp = statements
984 .iter()
985 .any(|s| &s.connection == conn && s.kind == StatementKind::TempCreate)
986 || session.lock().unwrap().loaded;
987 session.lock().unwrap().want_read_only(!creates_temp);
988 }
989 }
990 if let Some(e) = two_tabs {
991 return Err(e);
992 }
993
994 if self.opts.dry_run {
995 let plan = self.plan();
996 self.ui.compiled(&plan);
997 return Ok(());
998 }
999
1000 for conn in per_connection.keys() {
1002 let n = statements.iter().filter(|s| &s.connection == conn).count();
1003 if n > 1 {
1004 self.check_sessions(&pool, conn, n)?;
1005 }
1006 }
1007
1008 if self.opts.live_check {
1009 return self.live_check(&statements);
1010 }
1011
1012 std::fs::create_dir_all(self.run_dir.join(".spool")).map_err(|e| e.to_string())?;
1014 for (i, st) in statements.iter().enumerate() {
1015 let session = pool.session(&st.connection)?;
1016 self.execute_statement(&session, i, st)?;
1017 }
1018 }
1019 pool.close_all();
1020
1021 self.name_result_sets()?;
1022
1023 let filename = self.file_names();
1025 self.format(&filename)?;
1026
1027 if self.opts.preview.is_none() {
1029 self.drift = self.schema_drift();
1030 if !self.drift.is_empty() {
1031 if self.opts.accept_schema_change {
1032 self.ui.warn(&format!(
1033 " schema changed since the last successful run (accepted): {}",
1034 self.drift.join("; ")
1035 ));
1036 } else {
1037 return Err(format!(
1038 "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",
1039 self.drift.join("; "),
1040 rel(&self.project.root, &self.run_dir).display()
1041 ));
1042 }
1043 }
1044 }
1045
1046 if self.opts.preview.is_some() {
1048 self.delivery_note = Some("preview: not delivered".into());
1049 let dir = rel(&self.project.root, &self.run_dir);
1050 self.ui.step(
1051 Level::Info,
1052 "Preview",
1053 &format!("not delivered; output stays in {}", dir.display()),
1054 None,
1055 );
1056 } else {
1057 self.deliver()?;
1058 }
1059
1060 if self.opts.preview.is_none()
1062 && let Err(error) = self.write_snapshot()
1063 {
1064 self.ui.warn(&format!(
1065 "delivery completed, but the schema snapshot could not be written: {error}; the next run may compare against an older schema"
1066 ));
1067 }
1068 Ok(())
1069 }
1070
1071 fn render_query(
1073 &mut self,
1074 renderer: &Renderer,
1075 q: &QueryEntry,
1076 connection: &str,
1077 two_tabs: &mut Option<String>,
1078 ) -> Result<Vec<Statement>, Fail> {
1079 let mut out = Vec::new();
1080 let src = std::fs::read_to_string(self.project.root.join(&q.path))
1081 .map_err(|e| format!("{}: {e}", q.path.display()))?;
1082 let _ = renderer.take_sources();
1083 let sql = renderer
1084 .render(&q.path, &src)
1085 .map_err(|e: RenderError| e.to_string())?;
1086 let known = self
1089 .parsed
1090 .query(&q.query)
1091 .map(|p| p.sources.clone())
1092 .unwrap_or_default();
1093 if let Some((s, t)) = renderer
1094 .take_sources()
1095 .into_iter()
1096 .find(|(s, t)| !known.contains(&format!("{s}.{t}")))
1097 {
1098 return Err(format!(
1099 "{}: `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",
1100 q.path.display(),
1101 q.query
1102 ));
1103 }
1104 let sensitive = crate::secrets::contains_secret(&sql);
1105 std::fs::write(
1106 self.compiled_dir.join(format!("{}.sql", q.query)),
1107 crate::secrets::mask(&sql).as_bytes(),
1108 )
1109 .map_err(|e| e.to_string())?;
1110 self.ui
1111 .step(Level::Debug, "Rendered", &q.path.display().to_string(), None);
1112 let parts = sqlsplit::split(&sql);
1115 if q.tab && parts.is_empty() {
1116 return Err(format!(
1117 "{}: `{}` makes a tab but has no statements; add a query, or `tab: false` if it's meant to be empty",
1118 q.path.display(),
1119 q.query
1120 ));
1121 }
1122 let n = parts.len();
1123 for (i, st) in parts.into_iter().enumerate() {
1124 let body = sqlsplit::strip_leading_comments(&st.text);
1125 let line = st.line + st.text[..st.text.len() - body.len()].matches('\n').count();
1126 let kind = sqlsplit::classify(&st.text);
1127 let last = i + 1 == n;
1128 if q.tab && !last && kind == StatementKind::Read && two_tabs.is_none() {
1129 *two_tabs = Some(format!(
1130 "{}:{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:`",
1131 q.path.display(),
1132 q.query
1133 ));
1134 }
1135 out.push(Statement {
1136 query: q.query.clone(),
1137 connection: connection.to_string(),
1138 file: q.path.clone(),
1139 line,
1140 kind,
1141 text: st.text,
1142 sensitive,
1143 tab: q.tab && last,
1144 });
1145 }
1146 Ok(out)
1147 }
1148
1149 fn check_sessions(&self, pool: &Pool, connection: &str, n: usize) -> Result<(), Fail> {
1151 let session = pool.session(connection)?;
1152 let mut s = session.lock().unwrap();
1153 if !s.get()?.has(CAP_SESSIONS) {
1154 let kind = pool.kind(connection).unwrap_or_default();
1155 return Err(format!(
1156 "this Binding runs {n} statements on connection `{connection}`, but the `{kind}` source plugin can't hold one session across them; nothing was run"
1157 ));
1158 }
1159 Ok(())
1160 }
1161
1162 fn execute_statement(
1164 &mut self,
1165 session: &Arc<Mutex<Session>>,
1166 i: usize,
1167 st: &Statement,
1168 ) -> Result<(), Fail> {
1169 let sql_log = self.ui.sql_log();
1170 sql_log(
1171 &format!("{}:{}", st.file.display(), st.line),
1172 &protected_sql(&st.text, st.sensitive),
1173 );
1174 let spool_path = self.run_dir.join(".spool").join(format!("{i}.arrow"));
1175 let mut writer: Option<FileWriter<File>> = None;
1176 let t = Instant::now();
1177 let exec = {
1178 let mut s = session.lock().unwrap();
1179 let (p, _log_scope) = s.statement(st.sensitive)?;
1180 p.execute(&st.text, self.opts.preview, |schema, batch| {
1181 if writer.is_none() {
1182 let f = File::create(&spool_path).map_err(|e| e.to_string())?;
1183 writer = Some(FileWriter::try_new(f, schema).map_err(|e| e.to_string())?);
1184 }
1185 writer.as_mut().unwrap().write(&batch).map_err(|e| e.to_string())
1186 })
1187 .map_err(|e| {
1188 format!(
1189 "{}:{}: {}",
1190 st.file.display(),
1191 st.line,
1192 masked_source_error(&e, st.sensitive)
1193 )
1194 })?
1195 };
1196 let what = match &exec {
1197 Execution::Result { rows, .. } => {
1198 format!("{} row{}", thousands(*rows), if *rows == 1 { "" } else { "s" })
1199 }
1200 Execution::NoResult {
1201 rows_affected: Some(n),
1202 } => format!("no result set ({n} affected)"),
1203 Execution::NoResult { .. } => "no result set".to_string(),
1204 };
1205 let at = format!("{}:{}", st.file.display(), st.line);
1206 self.ui.step(
1207 Level::Debug,
1208 "Executed",
1209 &format!("{at} {what}"),
1210 Some(t.elapsed()),
1211 );
1212 let (schema, rows) = match (st.tab, exec) {
1215 (true, Execution::Result { schema, rows }) => (schema, rows),
1216 (true, Execution::NoResult { .. }) => {
1217 return Err(format!(
1218 "{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",
1219 st.query
1220 ));
1221 }
1222 (false, _) => {
1223 drop(writer);
1224 let _ = std::fs::remove_file(&spool_path);
1225 return Ok(());
1226 }
1227 };
1228 {
1229 let mut w = match writer {
1230 Some(w) => w,
1231 None => FileWriter::try_new(File::create(&spool_path).map_err(|e| e.to_string())?, &schema)
1232 .map_err(|e| e.to_string())?,
1233 };
1234 w.finish().map_err(|e| e.to_string())?;
1235 self.produced.push(Produced {
1236 query: st.query.clone(),
1237 connection: st.connection.clone(),
1238 schema,
1239 rows,
1240 spool: spool_path,
1241 name: String::new(),
1242 anchor: None,
1243 header: None,
1244 columns: Default::default(),
1245 });
1246 }
1247 Ok(())
1248 }
1249
1250 fn render_destinations(&mut self) -> Result<Vec<RenderedDest>, Fail> {
1253 if self.b.output.destinations.is_empty() {
1254 return Ok(Vec::new());
1255 }
1256 let renderer = self.renderer(None)?;
1257 let connections = self.pool.as_ref().expect("pool").connections.clone();
1258 let mut out = Vec::new();
1259 for (i, d) in self.b.output.destinations.iter().enumerate() {
1260 let profile = self
1261 .parsed
1262 .destinations
1263 .get(i)
1264 .cloned()
1265 .flatten()
1266 .ok_or_else(|| format!("destination `{}` has no profile", d.profile))?;
1267 if matches!(
1269 self.project.profiles.entry(Role::Destination, &profile),
1270 Entry::Nowhere
1271 ) {
1272 out.push(RenderedDest {
1273 profile,
1274 path: None,
1275 options: JsonMap::new(),
1276 });
1277 continue;
1278 }
1279 renderer.set_destination(connections.profile(&profile, Some("destination")).ok());
1280 let rendered = (|| {
1281 let path = match &d.path {
1282 Some(p) => Some(
1283 renderer
1284 .render(&self.report.file, p)
1285 .map_err(|e| format!("output path: {e}"))?,
1286 ),
1287 None => None,
1288 };
1289 let mut options = JsonMap::new();
1290 for (k, v) in &d.options {
1291 let v = render_json(&renderer, &self.report.file, v)
1292 .map_err(|e| format!("destination `{profile}` option `{k}`: {e}"))?;
1293 options.insert(k.clone(), v);
1294 }
1295 Ok::<_, Fail>(RenderedDest {
1296 profile: profile.clone(),
1297 path,
1298 options,
1299 })
1300 })();
1301 renderer.set_destination(None);
1302 out.push(rendered?);
1303 }
1304 Ok(out)
1305 }
1306
1307 fn name_result_sets(&mut self) -> Result<(), Fail> {
1309 let entries: BTreeMap<&str, &QueryEntry> =
1310 self.b.queries.iter().map(|q| (q.query.as_str(), q)).collect();
1311 for p in &mut self.produced {
1312 let q = entries[p.query.as_str()];
1313 p.name = q.tab_name.clone().unwrap_or_else(|| q.query.clone());
1314 p.anchor = q.anchor.clone();
1315 p.header = q.header;
1316 p.columns = q.columns.clone();
1317 }
1318 if self.b.output.format == "xlsx" {
1319 let mut seen: BTreeMap<String, String> = BTreeMap::new();
1320 for p in &self.produced {
1321 check_sheet_name(&p.name)?;
1322 if let Some(prev) = seen.insert(p.name.to_lowercase(), p.query.clone()) {
1323 return Err(format!(
1324 "sheet name `{}` is used twice (by `{prev}` and `{}`); Excel sheet names must be unique",
1325 p.name, p.query
1326 ));
1327 }
1328 }
1329 }
1330 Ok(())
1331 }
1332
1333 fn plan(&mut self) -> BindingPlan {
1335 let file = self.file_names();
1336 let compiled = self
1337 .b
1338 .queries
1339 .iter()
1340 .map(|q| {
1341 rel(
1342 &self.project.root,
1343 &self.compiled_dir.join(format!("{}.sql", q.query)),
1344 )
1345 })
1346 .collect();
1347 let destinations = self
1348 .dests
1349 .iter()
1350 .map(|d| {
1351 let out = self.project.profiles.target(Role::Destination, &d.profile);
1352 PlannedDestination {
1353 profile: d.profile.clone(),
1354 kind: out.map(|o| o.kind.clone()),
1355 target: Some(self.project.profiles.target_of(Role::Destination, &d.profile)),
1356 path: d.path.clone(),
1357 delivers: out.is_some(),
1358 }
1359 })
1360 .collect();
1361 let pool = self.pool.clone().expect("pool");
1362 let queries = self
1363 .parsed
1364 .queries
1365 .iter()
1366 .map(|q| {
1367 let connection = q.connection.clone().unwrap_or_default();
1368 PlannedQuery {
1369 query: q.query.clone(),
1370 kind: pool.kind(&connection).unwrap_or_default(),
1371 target: self.project.profiles.target_of(Role::Connection, &connection),
1372 connection,
1373 sources: q.sources.clone(),
1374 }
1375 })
1376 .collect();
1377 BindingPlan {
1378 report: self.report.name.clone(),
1379 set: self.b.set.clone(),
1380 compiled,
1381 profile: self.parsed.inherited.clone(),
1382 queries,
1383 target: self.target.clone(),
1384 format: self.b.output.format.clone(),
1385 output: rel(&self.project.root, &self.run_dir.join(file)),
1386 destinations,
1387 schedules: self.b.schedules.clone(),
1388 }
1389 }
1390
1391 fn file_names(&mut self) -> String {
1395 for d in self.dests.iter_mut().filter(|d| d.path.is_some()) {
1396 if let Some(p) = &self.opts.output_path {
1397 d.path = Some(p.clone());
1398 }
1399 if let (Some(n), Some(r)) = (&self.opts.output_name, d.path.as_mut()) {
1400 *r = match r.rfind(['/', '\\']) {
1401 Some(i) => format!("{}{n}", &r[..=i]),
1402 None => n.clone(),
1403 };
1404 }
1405 }
1406 let ext = match &self.b.output.extension {
1407 Some(e) => e.as_str(),
1408 None => extension(&self.b.output.format),
1409 };
1410 let from_remote = self
1411 .dests
1412 .iter()
1413 .find_map(|d| d.path.as_deref())
1414 .and_then(|r| r.rsplit(['/', '\\']).next())
1415 .filter(|s| !s.is_empty())
1416 .map(str::to_string);
1417 self.opts
1418 .output_name
1419 .clone()
1420 .or(from_remote)
1421 .unwrap_or_else(|| match ext {
1422 "" => self.report.name.clone(),
1423 ext => format!("{}.{ext}", self.report.name),
1424 })
1425 }
1426
1427 fn format(&mut self, filename: &str) -> Result<(), Fail> {
1428 std::fs::create_dir_all(&self.run_dir).map_err(|e| e.to_string())?;
1429 if self.produced.is_empty() {
1430 self.ui
1431 .warn(" no query makes a tab (every query has `tab: false`); no output file was written");
1432 return Ok(());
1433 }
1434 let out = &self.b.output;
1435 let plugin = find_plugin(self.project, PluginKind::Format, &out.format)?;
1436 let mut p = plugin
1437 .start(self.ui.plugin_log(), Some(&self.project.root))
1438 .map_err(|e| e.to_string())?;
1439 let multi = out.format == "xlsx" || out.template.is_some() || !is_single_table(&out.format);
1440 let groups: Vec<(String, Vec<usize>)> = if multi || self.produced.len() == 1 {
1441 vec![(filename.to_string(), (0..self.produced.len()).collect())]
1442 } else {
1443 let (stem, ext) = split_ext(filename);
1444 let mut g = Vec::new();
1445 for (i, r) in self.produced.iter().enumerate() {
1446 if r.name.contains(['/', '\\']) {
1447 return Err(format!("`{}` can't be used in a file name", r.name));
1448 }
1449 g.push((format!("{stem}_{}{ext}", r.name), vec![i]));
1450 }
1451 g
1452 };
1453 let template = match &out.template {
1454 Some(t) => Some(self.template_payload(t)?),
1455 None => None,
1456 };
1457 for (name, idxs) in groups {
1458 let t = Instant::now();
1459 let path = self.run_dir.join(&name);
1460 let metas: Vec<ResultSetMeta> = idxs
1461 .iter()
1462 .map(|&i| {
1463 let r = &self.produced[i];
1464 ResultSetMeta {
1465 name: r.name.clone(),
1466 query: r.query.clone(),
1467 result_index: 1,
1468 anchor: r.anchor.clone(),
1469 header: r.header,
1470 columns: r.columns.clone(),
1471 }
1472 })
1473 .collect();
1474 p.write_begin(
1475 &path.to_string_lossy(),
1476 &out.format,
1477 out.options.clone(),
1478 metas,
1479 template.clone(),
1480 )
1481 .map_err(|e| e.to_string())?;
1482 for &i in &idxs {
1483 let r = &self.produced[i];
1484 let reader = FileReader::try_new(File::open(&r.spool).map_err(|e| e.to_string())?, None)
1485 .map_err(|e| e.to_string())?;
1486 let schema = r.schema.clone();
1487 let mut any = false;
1489 for b in reader {
1490 let b = b.map_err(|e| e.to_string())?;
1491 any = true;
1492 p.send_batch(&b)
1493 .map_err(|e| format!("{} format: {e}", out.format))?;
1494 }
1495 if !any {
1496 p.send_batch(&RecordBatch::new_empty(schema))
1497 .map_err(|e| format!("{} format: {e}", out.format))?;
1498 }
1499 p.send(&dre_protocol::msg::Request::ResultSetEnd {})
1500 .map_err(|e| format!("{} format: {e}", out.format))?;
1501 }
1502 let (files, warnings) = p
1503 .write_finish_with_warnings()
1504 .map_err(|e| format!("{} format: {e}", out.format))?;
1505 for w in warnings {
1506 self.ui.warn(&format!(" {} format: {w}", out.format));
1507 }
1508 for f in files {
1509 let f = PathBuf::from(f);
1510 let size = std::fs::metadata(&f).map(|m| m.len()).unwrap_or(0);
1511 let shown = rel(&self.project.root, &f);
1512 self.ui.step(
1513 Level::Debug,
1514 "Wrote",
1515 &format!("{} ({})", shown.display(), human_bytes(size)),
1516 Some(t.elapsed()),
1517 );
1518 self.files.push((f, None));
1519 }
1520 }
1521 let _ = p.close();
1522 Ok(())
1523 }
1524
1525 fn template_payload(&mut self, t: &crate::project::Template) -> Result<Json, Fail> {
1526 let root = &self.project.root;
1527 let file = [root.join(&t.file), root.join("templates").join(&t.file)]
1528 .into_iter()
1529 .find(|p| p.is_file())
1530 .ok_or_else(|| format!("template file `{}` doesn't exist", t.file))?;
1531 let renderer = self.renderer(None)?;
1532 let mut values = JsonMap::new();
1533 for b in &t.bindings {
1534 if let (Some(cell), Some(v)) = (&b.cell, &b.value) {
1535 let rendered = renderer
1536 .render(&self.report.file, v)
1537 .map_err(|e| format!("template value: {e}"))?;
1538 values.insert(format!("{}!{}", b.sheet, cell), Json::String(rendered));
1539 }
1540 }
1541 Ok(json!({"file": file.to_string_lossy(), "bindings": t.bindings, "values": values}))
1542 }
1543
1544 fn deliver(&mut self) -> Result<(), Fail> {
1547 if self.dests.is_empty() {
1548 self.delivery_note = Some("no destination declared; output stays in target/".into());
1549 let dir = rel(&self.project.root, &self.run_dir);
1550 self.ui.step(
1551 Level::Debug,
1552 "Kept",
1553 &format!("no destination declared: output stays in {}", dir.display()),
1554 None,
1555 );
1556 return Ok(());
1557 }
1558 if self.files.is_empty() {
1559 return Ok(());
1560 }
1561 let dests = std::mem::take(&mut self.dests);
1562 let mut failures = Vec::new();
1563 let mut nowhere = Vec::new();
1564 for d in &dests {
1565 let profiles = &self.project.profiles;
1566 let target = profiles.target_of(Role::Destination, &d.profile);
1567 let kind = profiles
1569 .target(Role::Destination, &d.profile)
1570 .map(|o| o.kind.clone());
1571 let (status, location, error) = match self.deliver_one(d) {
1572 Ok(Some(loc)) => ("delivered", Some(loc), None),
1573 Ok(None) => {
1574 nowhere.push(nowhere_note(&d.profile, &target));
1575 ("not_delivered", None, None)
1576 }
1577 Err(e) => {
1578 failures.push(e.clone());
1579 ("failed", None, Some(e))
1580 }
1581 };
1582 let mut record = json!({"profile": d.profile, "type": kind, "target": target, "status": status});
1583 if let Some(l) = location {
1584 record["location"] = json!(l);
1585 }
1586 if let Some(e) = error {
1587 record["error"] = json!(e);
1588 }
1589 self.deliveries.push(record);
1590 }
1591 self.dests = dests;
1592 if self.files.iter().all(|(_, d)| d.is_none()) && !nowhere.is_empty() {
1593 self.delivery_note = Some(nowhere.join("; "));
1594 }
1595 match failures.len() {
1596 0 => Ok(()),
1597 1 if self.dests.len() == 1 => Err(failures.remove(0)),
1598 n => Err(format!(
1599 "{n} of {} destinations failed: {}",
1600 self.dests.len(),
1601 failures.join("; ")
1602 )),
1603 }
1604 }
1605
1606 fn deliver_one(&mut self, d: &RenderedDest) -> Result<Option<String>, Fail> {
1609 let profiles = &self.project.profiles;
1610 let out = match profiles.entry(Role::Destination, &d.profile) {
1611 Entry::Use(o) => o,
1612 Entry::Nowhere => {
1613 let note = nowhere_note(&d.profile, &profiles.target_of(Role::Destination, &d.profile));
1614 self.ui.step(Level::Info, "Kept", ¬e, None);
1615 return Ok(None);
1616 }
1617 Entry::Missing => return Err(profiles.missing_entry(Role::Destination, &d.profile)),
1619 Entry::Unknown => {
1620 return Err(format!(
1621 "destination profile `{}` isn't in profiles.yml",
1622 d.profile
1623 ));
1624 }
1625 };
1626 let kind = out.kind.clone();
1627 let connection = render_connection(out)?;
1628 let targets: Vec<DeliveryFile> = self
1629 .files
1630 .iter()
1631 .map(|(f, _)| {
1632 let name = f.file_name().unwrap().to_string_lossy().to_string();
1633 let r = d.path.as_deref().map(|r| {
1634 if self.files.len() == 1 {
1635 r.to_string()
1636 } else {
1637 match r.rfind(['/', '\\']) {
1638 Some(i) => format!("{}{name}", &r[..=i]),
1639 None => name.clone(),
1640 }
1641 }
1642 });
1643 DeliveryFile {
1644 local_path: f.to_string_lossy().to_string(),
1645 remote_path: r,
1646 }
1647 })
1648 .collect();
1649 let mut locations = Vec::new();
1650 if kind == LOCAL_TYPE {
1651 if let Some(k) = d.options.keys().next() {
1652 return Err(format!(
1653 "the local destination takes no options, but `{}` has `{k}`; check the key's spelling",
1654 d.profile
1655 ));
1656 }
1657 for (i, f) in targets.iter().enumerate() {
1658 let t = Instant::now();
1659 let r = f
1660 .remote_path
1661 .as_ref()
1662 .ok_or("the local destination needs `output.destination.path`")?;
1663 let dst = self.project.root.join(r);
1664 if let Some(parent) = dst.parent() {
1665 std::fs::create_dir_all(parent)
1666 .map_err(|e| format!("can't create {}: {e}", parent.display()))?;
1667 }
1668 std::fs::copy(&f.local_path, &dst).map_err(|e| {
1669 format!(
1670 "delivery to {} failed: {e}; the output is still in target/",
1671 dst.display()
1672 )
1673 })?;
1674 let loc = dst.to_string_lossy().to_string();
1675 self.ui.step(Level::Debug, "Delivered", &loc, Some(t.elapsed()));
1676 self.files[i].1.get_or_insert_with(|| loc.clone());
1677 locations.push(loc);
1678 }
1679 return Ok(Some(locations.join(", ")));
1680 }
1681 let failed = |e: dre_protocol::host::HostError| {
1682 format!("delivery through `{kind}` failed: {e}; the output is still in target/")
1683 };
1684 let plugin = find_plugin(self.project, PluginKind::Destination, &kind)?;
1685 let mut p = plugin
1686 .start(self.ui.plugin_log(), Some(&self.project.root))
1687 .map_err(|e| e.to_string())?;
1688 let batches: Vec<Vec<usize>> = if targets.len() > 1 && p.has(CAP_MULTI_FILE) {
1689 vec![(0..targets.len()).collect()]
1690 } else {
1691 (0..targets.len()).map(|i| vec![i]).collect()
1692 };
1693 for batch in batches {
1694 let t = Instant::now();
1695 let files: Vec<DeliveryFile> = batch.iter().map(|&i| targets[i].clone()).collect();
1696 let loc = p
1697 .deliver_files(&files, connection.clone(), d.options.clone())
1698 .map_err(failed)?;
1699 self.ui.step(Level::Debug, "Delivered", &loc, Some(t.elapsed()));
1700 for i in batch {
1701 self.files[i].1.get_or_insert_with(|| loc.clone());
1702 }
1703 locations.push(loc);
1704 }
1705 let _ = p.close();
1706 Ok(Some(locations.join(", ")))
1707 }
1708
1709 fn snapshot(&self) -> Json {
1710 let sets: Vec<Json> = self
1711 .produced
1712 .iter()
1713 .map(|p| {
1714 json!({
1715 "name": p.name,
1716 "columns": p.schema.fields().iter().map(|f| json!({
1717 "name": f.name(),
1718 "type": f.data_type().to_string(),
1719 "type_shape": crate::schema::type_shape(f.data_type()),
1720 })).collect::<Vec<_>>(),
1721 })
1722 })
1723 .collect();
1724 json!({ "result_sets": sets })
1725 }
1726
1727 fn schema_drift(&mut self) -> Vec<String> {
1728 let path = self.schema_dir.join("last_success.json");
1729 let mut current = self.snapshot();
1730 let Ok(prev) = std::fs::read_to_string(&path) else {
1731 crate::schema::redact(&mut current);
1732 self.protected_snapshot = Some(current);
1733 return Vec::new();
1734 };
1735 let Ok(mut prev) = serde_json::from_str::<Json>(&prev) else {
1736 crate::schema::redact(&mut current);
1737 self.protected_snapshot = Some(current);
1738 return Vec::new();
1739 };
1740 let original = prev.clone();
1741 crate::schema::redact(&mut prev);
1742 crate::schema::redact(&mut current);
1743 if crate::secrets::enabled() {
1744 crate::schema::align_protection(&mut prev, &mut current);
1745 }
1746 if prev != original {
1747 let protected =
1748 serde_json::to_string_pretty(&prev).expect("serializing a serde_json::Value cannot fail");
1749 if let Err(error) = std::fs::write(&path, protected) {
1750 self.ui.warn(&format!(
1751 "can't protect the existing schema snapshot at {}: {error}; continuing without rewriting it",
1752 path.display()
1753 ));
1754 }
1755 }
1756 let drift = crate::schema::drift(&prev, ¤t);
1757 self.protected_snapshot = Some(current);
1758 drift
1759 }
1760
1761 fn write_snapshot(&self) -> std::io::Result<()> {
1762 std::fs::create_dir_all(&self.schema_dir)?;
1763 let snapshot = self.protected_snapshot.clone().unwrap_or_else(|| {
1764 let mut snapshot = self.snapshot();
1765 crate::schema::redact(&mut snapshot);
1766 snapshot
1767 });
1768 std::fs::write(
1769 self.schema_dir.join("last_success.json"),
1770 serde_json::to_string_pretty(&snapshot)?,
1771 )
1772 }
1773
1774 fn write_results(&self, status: &Status, error: Option<&str>) -> std::io::Result<()> {
1775 std::fs::create_dir_all(&self.run_dir)?;
1776 let outputs: Vec<Json> = self
1777 .files
1778 .iter()
1779 .map(|(f, delivered)| {
1780 json!({
1781 "path": record_path(self.project, f),
1782 "size": std::fs::metadata(f).map(|m| m.len()).unwrap_or(0),
1783 "delivered_to": delivered,
1784 })
1785 })
1786 .collect();
1787 let results = json!({
1788 "report": self.report.name,
1789 "set": self.b.set,
1790 "binding": self.b.dir_name(),
1791 "managed": self.report.managed,
1792 "profile": self.parsed.inherited,
1793 "connections": self.parsed.connections(),
1794 "target": self.target,
1795 "schedule": self.opts.schedule,
1796 "schedule_vars": self.schedule_vars,
1797 "vars": self.rendered_vars,
1798 "run_date": self.date.to_string(),
1799 "scheduled_at": self.opts.scheduled_at.map(rfc3339),
1800 "timezone": self.calendar.tz.name(),
1801 "params": self.opts.params(self.date),
1802 "status": status,
1803 "error": error,
1804 "preview": self.opts.preview.is_some(),
1805 "row_limit": self.opts.preview,
1806 "started_at": self.started_at.to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
1807 "duration_ms": self.started.elapsed().as_millis() as u64,
1808 "result_sets": self.produced.iter().map(|p| json!({
1809 "name": p.name,
1810 "query": p.query,
1811 "connection": p.connection,
1812 "rows": p.rows,
1813 "columns": p.schema.fields().iter().map(|f| f.name().clone()).collect::<Vec<_>>(),
1814 })).collect::<Vec<_>>(),
1815 "outputs": outputs,
1816 "delivery": self.delivery_note,
1817 "deliveries": self.deliveries,
1818 "schema_drift": self.drift,
1819 "target_path": self.project.target_dir,
1820 "manifest_checksum": self.opts.manifest_checksum,
1821 });
1822 std::fs::write(
1823 self.run_dir.join("run_results.json"),
1824 (crate::secrets::to_json_pretty(&results)? + "\n").as_bytes(),
1825 )
1826 }
1827
1828 fn live_check(&mut self, statements: &[Statement]) -> Result<(), Fail> {
1831 let pool = self.pool.clone().expect("pool");
1832 let mut failures = Vec::new();
1833 let mut unexecuted_setup: BTreeMap<String, (PathBuf, usize)> = BTreeMap::new();
1834 let mut uncheckable: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
1835 let sql_log = self.ui.sql_log();
1836 for st in statements {
1837 if uncheckable.contains(&st.connection) {
1838 continue;
1839 }
1840 let session = pool.session(&st.connection)?;
1841 let mut s = session.lock().unwrap();
1842 if !s.get()?.has(dre_protocol::CAP_CHECK) {
1843 let kind = pool.kind(&st.connection).unwrap_or_default();
1844 self.ui.warn(&format!(
1845 " not checkable: the `{kind}` source plugin (connection `{}`) can't check statements without running them",
1846 st.connection
1847 ));
1848 uncheckable.insert(st.connection.clone());
1849 continue;
1850 }
1851 let verb = if st.kind == StatementKind::TempCreate {
1852 ""
1853 } else {
1854 " (check)"
1855 };
1856 sql_log(
1857 &format!("{}:{}{verb}", st.file.display(), st.line),
1858 &protected_sql(&st.text, st.sensitive),
1859 );
1860 if st.kind == StatementKind::TempCreate {
1861 let result = {
1862 let (process, _log_scope) = s.statement(st.sensitive)?;
1863 process.execute(&st.text, None, |_, _| Ok(()))
1864 };
1865 if let Err(e) = result {
1866 failures.push(format!(
1867 "{}:{}: {}",
1868 st.file.display(),
1869 st.line,
1870 masked_source_error(&e, st.sensitive)
1871 ));
1872 }
1873 continue;
1874 }
1875 let result = {
1876 let (process, _log_scope) = s.statement(st.sensitive)?;
1877 process.check(&st.text)
1878 };
1879 if let Err(e) = result {
1880 let mut msg = format!(
1881 "{}:{}: {}",
1882 st.file.display(),
1883 st.line,
1884 masked_source_error(&e, st.sensitive)
1885 );
1886 if let Some((f, l)) = unexecuted_setup.get(&st.connection) {
1887 msg.push_str(&format!(
1888 " (may be a false positive: the setup statement at {}:{l} wasn't executed during the check)",
1889 f.display()
1890 ));
1891 }
1892 failures.push(msg);
1893 }
1894 if st.kind == StatementKind::Other {
1895 unexecuted_setup.insert(st.connection.clone(), (st.file.clone(), st.line));
1896 }
1897 }
1898 self.check_source_columns(&pool, &mut failures);
1899 if failures.is_empty() {
1900 Ok(())
1901 } else {
1902 Err(failures.join("\n "))
1903 }
1904 }
1905
1906 fn check_source_columns(&mut self, pool: &Arc<Pool>, failures: &mut Vec<String>) {
1909 type Checked = std::collections::BTreeSet<(PathBuf, String, String, String)>;
1911 static CHECKED: std::sync::LazyLock<Mutex<Checked>> = std::sync::LazyLock::new(Mutex::default);
1912 for q in self.parsed.queries.clone() {
1913 let Some(conn) = q.connection.clone() else {
1914 continue;
1915 };
1916 for key in &q.sources {
1917 let Some(r) = self.parsed.sources.get(key).cloned() else {
1918 continue;
1919 };
1920 let Some(declared) = self
1921 .project
1922 .sources
1923 .get(&r.source)
1924 .and_then(|s| s.table(&r.table))
1925 .map(|t| t.columns.clone())
1926 .filter(|c| !c.is_empty())
1927 else {
1928 continue;
1929 };
1930 let id = (
1931 self.project.root.clone(),
1932 key.clone(),
1933 conn.clone(),
1934 self.project.profiles.target_of(Role::Connection, &conn),
1935 );
1936 if !CHECKED.lock().unwrap().insert(id) {
1937 continue;
1938 }
1939 let quote = if r.quoting.any() {
1940 let kind = pool.kind(&conn).unwrap_or_default();
1941 match pool.connections.identifier_quote(&kind) {
1942 Ok(q) => q,
1943 Err(e) => {
1944 failures.push(format!("source `{key}`: can't quote its name: {e}"));
1945 continue;
1946 }
1947 }
1948 } else {
1949 None
1950 };
1951 let rel = r.relation(quote.as_deref());
1952 let runner = PoolRunner {
1953 pool: pool.clone(),
1954 default: Some(conn.clone()),
1955 log: self.ui.sql_log(),
1956 };
1957 let actual = match runner.columns(&format!("select * from {rel} where 1=0"), None) {
1958 Ok(c) => c,
1959 Err(e) => {
1960 failures.push(format!("source `{key}` ({rel} on connection `{conn}`): {e}"));
1961 continue;
1962 }
1963 };
1964 for c in &declared {
1965 let Some(a) = actual.iter().find(|a| a.name.eq_ignore_ascii_case(&c.name)) else {
1966 failures.push(format!(
1967 "source `{key}`: declared column `{}` isn't in {rel} on connection `{conn}` (it has: {})",
1968 c.name,
1969 actual.iter().map(|a| a.name.as_str()).collect::<Vec<_>>().join(", ")
1970 ));
1971 continue;
1972 };
1973 let Some(dt) = &c.data_type else { continue };
1974 match crate::coltypes::matches(dt, &a.data_type) {
1975 Some(true) => {}
1976 Some(false) => failures.push(format!(
1977 "source `{key}`: column `{}` is declared `{dt}`, but {rel} on connection `{conn}` returns {}",
1978 c.name, a.data_type
1979 )),
1980 None => self.ui.warn(&format!(
1981 " source `{key}`: column `{}`'s `data_type: {dt}` isn't a type DRE can compare, so only its presence was checked",
1982 c.name
1983 )),
1984 }
1985 }
1986 }
1987 }
1988 }
1989}
1990
1991struct Session {
1993 path: Result<crate::plugins::Located, String>,
1994 connection: Result<JsonMap<String, Json>, String>,
1995 cwd: PathBuf,
1996 log: LogSink,
1997 opaque_log: Arc<AtomicBool>,
2000 unmanaged: bool,
2002 read_only: bool,
2003 loaded: bool,
2005 proc_: Option<PluginProcess>,
2006}
2007
2008impl Session {
2009 fn new(
2010 path: Result<crate::plugins::Located, String>,
2011 connection: Result<JsonMap<String, Json>, String>,
2012 cwd: PathBuf,
2013 log: LogSink,
2014 unmanaged: bool,
2015 ) -> Session {
2016 let opaque_log = Arc::new(AtomicBool::new(false));
2017 let protect = Arc::clone(&opaque_log);
2018 let downstream = log;
2019 let log: LogSink = Arc::new(move |plugin, line| {
2020 let line = crate::secrets::mask_source_error(line, protect.load(Ordering::SeqCst));
2021 downstream(plugin, &line);
2022 });
2023 Session {
2024 path,
2025 connection,
2026 cwd,
2027 log,
2028 opaque_log,
2029 unmanaged,
2030 read_only: unmanaged,
2031 loaded: false,
2032 proc_: None,
2033 }
2034 }
2035
2036 fn want_read_only(&mut self, ro: bool) {
2037 if self.read_only != ro {
2038 self.close();
2040 self.read_only = ro;
2041 }
2042 }
2043
2044 fn get(&mut self) -> Result<&mut PluginProcess, String> {
2045 if self.proc_.is_none() {
2046 let plugin = self.path.clone()?;
2047 let connection = self.connection.clone()?;
2048 let mut p = plugin
2049 .start(self.log.clone(), Some(&self.cwd))
2050 .map_err(|e| masked_source_error(&e, false))?;
2051 let ro = self.unmanaged && self.read_only && p.has(CAP_READ_ONLY);
2052 p.open(connection, ro).map_err(|e| {
2053 let protected = masked_source_error(&e, false);
2054 format!("can't open the source connection: {protected}")
2055 })?;
2056 self.proc_ = Some(p);
2057 }
2058 Ok(self.proc_.as_mut().unwrap())
2059 }
2060
2061 fn statement(&mut self, sensitive: bool) -> Result<(&mut PluginProcess, OpaqueLogScope), String> {
2062 let opaque_log = Arc::clone(&self.opaque_log);
2063 let process = self.get()?;
2064 let scope = OpaqueLogScope::new(opaque_log, sensitive);
2065 Ok((process, scope))
2066 }
2067
2068 fn close(&mut self) {
2069 if let Some(p) = self.proc_.take() {
2070 let _ = p.close();
2071 }
2072 }
2073}
2074
2075struct OpaqueLogScope(Arc<AtomicBool>);
2076
2077impl OpaqueLogScope {
2078 fn new(active: Arc<AtomicBool>, sensitive: bool) -> Self {
2079 active.store(sensitive, Ordering::SeqCst);
2080 Self(active)
2081 }
2082}
2083
2084impl Drop for OpaqueLogScope {
2085 fn drop(&mut self) {
2086 self.0.store(false, Ordering::SeqCst);
2087 }
2088}
2089
2090fn nowhere_note(profile: &str, target: &str) -> String {
2092 format!(
2093 "destination `{profile}`: `{target}` delivers nowhere (`deliver: false`); output stays in target/"
2094 )
2095}
2096
2097struct Pool {
2100 connections: Arc<ProfileConnections>,
2101 cwd: PathBuf,
2102 log: LogSink,
2103 unmanaged: bool,
2104 sessions: Mutex<BTreeMap<String, Arc<Mutex<Session>>>>,
2105}
2106
2107impl Pool {
2108 fn new(connections: Arc<ProfileConnections>, cwd: PathBuf, log: LogSink, unmanaged: bool) -> Pool {
2109 Pool {
2110 connections,
2111 cwd,
2112 log,
2113 unmanaged,
2114 sessions: Mutex::default(),
2115 }
2116 }
2117
2118 fn target(&self, name: &str) -> Result<ProfileTarget, String> {
2120 let profiles = &self.connections.profiles;
2121 match profiles.entry(Role::Connection, name) {
2122 Entry::Use(o) => Ok(o.clone()),
2123 Entry::Unknown => Err(format!(
2124 "connection `{name}` isn't under `{}:` in {}",
2125 profiles.section_key(Role::Connection),
2126 profiles.path.display()
2127 )),
2128 Entry::Missing | Entry::Nowhere => Err(profiles.missing_entry(Role::Connection, name)),
2130 }
2131 }
2132
2133 fn kind(&self, name: &str) -> Result<String, String> {
2135 Ok(self.target(name)?.kind)
2136 }
2137
2138 fn session(&self, name: &str) -> Result<Arc<Mutex<Session>>, String> {
2139 if let Some(s) = self.sessions.lock().unwrap().get(name) {
2140 return Ok(s.clone());
2141 }
2142 let output = self.target(name)?;
2143 let id = crate::project::PluginId::new(PluginKind::Source, output.kind.clone());
2144 let path = crate::plugins::locate_in(&self.connections.root, &self.connections.plugins, &id)
2145 .map_err(String::from);
2146 let session = Arc::new(Mutex::new(Session::new(
2147 path,
2148 render_connection(&output),
2149 self.cwd.clone(),
2150 self.log.clone(),
2151 self.unmanaged,
2152 )));
2153 self.sessions
2154 .lock()
2155 .unwrap()
2156 .insert(name.to_string(), session.clone());
2157 Ok(session)
2158 }
2159
2160 fn close_all(&self) {
2161 for s in self.sessions.lock().unwrap().values() {
2162 if let Ok(mut s) = s.lock() {
2163 s.close();
2164 }
2165 }
2166 }
2167}
2168
2169struct PoolRunner {
2172 pool: Arc<Pool>,
2173 default: Option<String>,
2174 log: LogSink,
2175}
2176
2177impl PoolRunner {
2178 fn session(&self, profile: Option<&str>) -> Result<Arc<Mutex<Session>>, String> {
2179 let name = profile.or(self.default.as_deref()).ok_or(
2180 "no connection here: pass `profile=`, or give the report a `profile:` (this isn't a query's SQL)",
2181 )?;
2182 self.pool.session(name)
2183 }
2184}
2185
2186impl QueryRunner for PoolRunner {
2187 fn run_query(&self, sql: &str, max_rows: u64, profile: Option<&str>) -> Result<QueryRows, String> {
2188 let session = self.session(profile)?;
2189 let mut s = session.lock().map_err(|_| "session lock poisoned".to_string())?;
2190 let sensitive = crate::secrets::contains_secret(sql);
2191 if s.unmanaged
2194 && let Some(bad) = sqlsplit::split(sql)
2195 .into_iter()
2196 .find(|st| sqlsplit::classify(&st.text) != StatementKind::Read)
2197 {
2198 return Err(format!(
2199 "run_query() in an unmanaged report may only read, but got `{}`; give the report a YAML to declare it",
2200 masked_sql_summary(&bad.text, 60, sensitive)
2201 ));
2202 }
2203 (self.log)("run_query()", &protected_sql(sql, sensitive));
2204 let (p, _log_scope) = s.statement(sensitive)?;
2205 let mut out = QueryRows::default();
2206 let mut too_many = false;
2207 let exec = p
2208 .execute(sql, Some(max_rows + 1), |schema, batch| {
2209 if out.columns.is_empty() {
2210 out.columns = schema.fields().iter().map(|f| f.name().clone()).collect();
2211 }
2212 if out.rows.len() as u64 + batch.num_rows() as u64 > max_rows {
2213 too_many = true;
2214 return Ok(());
2215 }
2216 out.rows.extend(crate::values::batch_rows(&batch));
2217 Ok(())
2218 })
2219 .map_err(|e| format!("run_query() failed: {}", masked_source_error(&e, sensitive)))?;
2220 if too_many {
2221 return Err(format!(
2222 "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"
2223 ));
2224 }
2225 if let Execution::Result { schema, .. } = exec
2226 && out.columns.is_empty()
2227 {
2228 out.columns = schema.fields().iter().map(|f| f.name().clone()).collect();
2229 }
2230 Ok(out)
2231 }
2232
2233 fn columns(&self, sql: &str, profile: Option<&str>) -> Result<Vec<Column>, String> {
2234 let session = self.session(profile)?;
2235 let mut s = session.lock().map_err(|_| "session lock poisoned".to_string())?;
2236 let sensitive = crate::secrets::contains_secret(sql);
2237 (self.log)("columns()", &protected_sql(sql, sensitive));
2238 let (p, _log_scope) = s.statement(sensitive)?;
2239 let mut schema: Option<SchemaRef> = None;
2240 let exec = p
2241 .execute(sql, Some(1), |sch, _| {
2242 schema.get_or_insert_with(|| sch.clone());
2243 Ok(())
2244 })
2245 .map_err(|e| masked_source_error(&e, sensitive))?;
2246 if let Execution::Result { schema: sch, .. } = exec {
2247 schema.get_or_insert(sch);
2248 }
2249 let schema = schema.ok_or("the query returned no result set")?;
2250 Ok(schema
2251 .fields()
2252 .iter()
2253 .map(|f| Column {
2254 name: f.name().clone(),
2255 data_type: f.data_type().to_string(),
2256 })
2257 .collect())
2258 }
2259
2260 fn load(&self, name: &str, table: &Table) -> Result<Option<(String, Option<String>)>, String> {
2261 let session = self.session(None)?;
2262 let mut s = session.lock().map_err(|_| "session lock poisoned".to_string())?;
2263 s.want_read_only(false);
2265 if !s.get()?.has(CAP_LOAD) {
2266 return Ok(None);
2267 }
2268 let batch = table.to_batch()?;
2269 let schema = batch.schema();
2270 (self.log)(
2271 &format!("lookup `{name}`"),
2272 &format!(
2273 "-- {} rows loaded through the plugin's `load` request",
2274 batch.num_rows()
2275 ),
2276 );
2277 let r = s.get()?.load(name, &schema, [batch]).map_err(|e| e.to_string())?;
2278 s.loaded = true;
2279 Ok(Some(r))
2280 }
2281}
2282
2283fn render_json(renderer: &Renderer, file: &Path, v: &Json) -> Result<Json, RenderError> {
2285 Ok(match v {
2286 Json::String(s) => Json::String(renderer.render(file, s)?),
2287 Json::Array(a) => Json::Array(
2288 a.iter()
2289 .map(|v| render_json(renderer, file, v))
2290 .collect::<Result<_, _>>()?,
2291 ),
2292 Json::Object(o) => Json::Object(
2293 o.iter()
2294 .map(|(k, v)| Ok((k.clone(), render_json(renderer, file, v)?)))
2295 .collect::<Result<_, RenderError>>()?,
2296 ),
2297 other => other.clone(),
2298 })
2299}
2300
2301fn render_connection(output: &ProfileTarget) -> Result<JsonMap<String, Json>, String> {
2303 let mut env = minijinja::Environment::new();
2304 env.set_undefined_behavior(minijinja::UndefinedBehavior::Strict);
2305 env.add_function("env_var", |name: String, default: Option<String>| -> Result<String, minijinja::Error> {
2306 std::env::var(&name).ok().or(default).ok_or_else(|| {
2307 minijinja::Error::new(
2308 minijinja::ErrorKind::UndefinedError,
2309 format!("`env_var('{name}')`: environment variable `{name}` is not set and no default is given"),
2310 )
2311 })
2312 });
2313 fn walk(env: &minijinja::Environment<'_>, v: &Json) -> Result<Json, String> {
2314 Ok(match v {
2315 Json::String(s) if crate::preflight::is_templated(s) => Json::String(
2316 env.render_str(s, ())
2317 .map_err(|e| format!("profiles.yml: {}", e.detail().unwrap_or("render error")))?,
2318 ),
2319 Json::Array(a) => Json::Array(a.iter().map(|x| walk(env, x)).collect::<Result<_, _>>()?),
2320 Json::Object(o) => Json::Object(
2321 o.iter()
2322 .map(|(k, x)| Ok((k.clone(), walk(env, x)?)))
2323 .collect::<Result<_, String>>()?,
2324 ),
2325 other => other.clone(),
2326 })
2327 }
2328 match walk(&env, &Json::Object(output.fields.clone()))? {
2329 Json::Object(o) => Ok(o),
2330 _ => unreachable!(),
2331 }
2332}
2333
2334pub fn find_plugin(
2337 project: &Project,
2338 kind: PluginKind,
2339 name: &str,
2340) -> Result<crate::plugins::Located, crate::plugins::LocateError> {
2341 crate::plugins::locate(project, &crate::project::PluginId::new(kind, name))
2342}
2343
2344struct ProfileConnections {
2348 profiles: Profiles,
2350 root: PathBuf,
2351 plugins: Vec<crate::project::PluginRequirement>,
2352 log: LogSink,
2353}
2354
2355static DESCRIBED: std::sync::LazyLock<Mutex<BTreeMap<DescribeKey, Option<Described>>>> =
2358 std::sync::LazyLock::new(Mutex::default);
2359
2360type DescribeKey = (PathBuf, Role, String);
2362
2363#[derive(Clone)]
2364struct Described {
2365 secrets: Vec<String>,
2366 identifier_quote: Option<String>,
2367}
2368
2369const SECRET_NAME_WORDS: &[&str] = &["password", "secret", "token", "key", "credential"];
2370
2371impl ProfileConnections {
2372 fn view(&self, role: Role, name: &str) -> Result<Connection, String> {
2373 let target = self.profiles.target_of(role, name);
2374 let out: ProfileTarget = match self.profiles.entry(role, name) {
2375 Entry::Use(o) => o.clone(),
2376 Entry::Nowhere => {
2377 return Err(format!(
2378 "destination `{name}`: `{target}` delivers nowhere (`deliver: false`), so it has no settings"
2379 ));
2380 }
2381 Entry::Missing => return Err(self.profiles.missing_entry(role, name)),
2382 Entry::Unknown => {
2383 return Err(format!(
2384 "no {} profile `{name}` in {}",
2385 role.as_str(),
2386 self.profiles.path.display()
2387 ));
2388 }
2389 };
2390 let mut secrets: Vec<String> = out
2391 .fields
2392 .iter()
2393 .filter(|(_, v)| v.as_str().is_some_and(|s| s.contains(crate::secrets::PREFIX)))
2394 .map(|(k, _)| k.clone())
2395 .collect();
2396 match self.described(role, &out.kind) {
2397 Some(d) => secrets.extend(d.secrets),
2398 None => secrets.extend(
2399 out.fields
2400 .keys()
2401 .filter(|k| {
2402 let k = k.to_lowercase();
2403 SECRET_NAME_WORDS.iter().any(|w| k.contains(w))
2404 })
2405 .cloned(),
2406 ),
2407 }
2408 let fields = render_connection(&out)?;
2409 Ok(Connection {
2410 profile: name.to_string(),
2411 target,
2412 kind: out.kind.clone(),
2413 fields,
2414 secrets,
2415 })
2416 }
2417
2418 fn described(&self, role: Role, kind: &str) -> Option<Described> {
2420 let key = (self.root.clone(), role, kind.to_string());
2421 if let Some(v) = DESCRIBED.lock().unwrap().get(&key) {
2422 return v.clone();
2423 }
2424 let plugin_kind = match role {
2425 Role::Connection => PluginKind::Source,
2426 Role::Destination => PluginKind::Destination,
2427 };
2428 let asked = (|| {
2429 if kind == LOCAL_TYPE {
2430 return Some(Described {
2431 secrets: Vec::new(),
2432 identifier_quote: None,
2433 });
2434 }
2435 let id = crate::project::PluginId::new(plugin_kind, kind);
2436 let plugin = crate::plugins::locate_in(&self.root, &self.plugins, &id).ok()?;
2437 let mut p = plugin.start(self.log.clone(), Some(&self.root)).ok()?;
2438 let d = p.description().ok();
2439 let _ = p.close();
2440 let d = d?;
2441 Some(Described {
2442 secrets: d
2443 .connection_fields
2444 .into_iter()
2445 .filter(|f| f.secret)
2446 .map(|f| f.name)
2447 .collect(),
2448 identifier_quote: d.identifier_quote,
2449 })
2450 })();
2451 DESCRIBED.lock().unwrap().insert(key, asked.clone());
2452 asked
2453 }
2454}
2455
2456impl Connections for ProfileConnections {
2457 fn profile(&self, name: &str, role: Option<&str>) -> Result<Connection, String> {
2458 let role = match role {
2459 Some("connection") => Role::Connection,
2460 Some(_) => Role::Destination,
2461 None => {
2462 let c = self.profiles.get(Role::Connection, name).is_some();
2463 let d = self.profiles.get(Role::Destination, name).is_some();
2464 match (c, d) {
2465 (true, true) => {
2466 return Err(format!(
2467 "`{name}` is both a connection and a destination profile; say which with `profile('{name}', role='connection')` or `role='destination'`"
2468 ));
2469 }
2470 (true, false) => Role::Connection,
2471 _ => Role::Destination,
2472 }
2473 }
2474 };
2475 self.view(role, name)
2476 }
2477
2478 fn identifier_quote(&self, kind: &str) -> Result<Option<String>, String> {
2479 match self.described(Role::Connection, kind) {
2480 Some(d) => Ok(d.identifier_quote),
2481 None => Err(format!(
2482 "the `{kind}` source plugin couldn't be asked for its identifier quote character"
2483 )),
2484 }
2485 }
2486}
2487
2488fn check_sheet_name(n: &str) -> Result<(), String> {
2489 if n.is_empty() || n.chars().count() > 31 {
2490 return Err(format!("sheet name `{n}` must be 1 to 31 characters long"));
2491 }
2492 if let Some(c) = n.chars().find(|c| "[]:*?/\\".contains(*c)) {
2493 return Err(format!(
2494 "sheet name `{n}` contains `{c}`, which Excel doesn't allow in sheet names"
2495 ));
2496 }
2497 if n.starts_with('\'') || n.ends_with('\'') {
2498 return Err(format!("sheet name `{n}` can't start or end with an apostrophe"));
2499 }
2500 if n.eq_ignore_ascii_case("history") {
2501 return Err("`History` is reserved by Excel and can't be a sheet name".into());
2502 }
2503 Ok(())
2504}
2505
2506fn is_single_table(format: &str) -> bool {
2507 matches!(format, "csv" | "delimited" | "fixed_width" | "parquet")
2508}
2509
2510fn extension(format: &str) -> &str {
2511 match format {
2512 "delimited" | "fixed_width" => "txt",
2513 f => f,
2514 }
2515}
2516
2517fn split_ext(name: &str) -> (&str, &str) {
2518 match name.rfind('.') {
2519 Some(i) if i > 0 => (&name[..i], &name[i..]),
2520 _ => (name, ""),
2521 }
2522}
2523
2524fn rel(root: &Path, p: &Path) -> PathBuf {
2525 crate::slash(p.strip_prefix(root).unwrap_or(p))
2526}
2527
2528pub fn rfc3339(t: chrono::DateTime<chrono::Utc>) -> String {
2530 t.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
2531}
2532
2533fn record_path(project: &Project, p: &Path) -> PathBuf {
2537 match p.strip_prefix(&project.root) {
2538 Ok(r) => crate::slash(r),
2539 Err(_) => rel(&project.target_dir, p),
2540 }
2541}
2542
2543impl BindingRun<'_> {
2544 fn summary_line(&self) -> String {
2546 let rows: u64 = self.produced.iter().map(|p| p.rows).sum();
2547 let mut parts = Vec::new();
2548 if !self.produced.is_empty() {
2549 let n = self.produced.len();
2550 parts.push(format!(
2551 "{n} result set{}, {} row{}",
2552 if n == 1 { "" } else { "s" },
2553 thousands(rows),
2554 if rows == 1 { "" } else { "s" }
2555 ));
2556 }
2557 let names: Vec<String> = self
2558 .files
2559 .iter()
2560 .map(|(f, _)| f.file_name().unwrap_or_default().to_string_lossy().to_string())
2561 .collect();
2562 if !names.is_empty() {
2563 parts.push(format!("→ {}", names.join(", ")));
2564 }
2565 let delivered: Vec<&str> = self
2567 .deliveries
2568 .iter()
2569 .filter_map(|d| d.get("location").and_then(Json::as_str))
2570 .collect();
2571 if !delivered.is_empty() {
2572 parts.push(format!("→ {}", delivered.join(", ")));
2573 } else if let Some(n) = &self.delivery_note
2574 && self.opts.preview.is_none()
2575 && !self.b.output.destinations.is_empty()
2576 {
2577 parts.push(format!("({n})"));
2578 }
2579 parts.join(" ")
2580 }
2581}
2582
2583pub fn thousands(n: u64) -> String {
2585 let s = n.to_string();
2586 let mut out = String::new();
2587 for (i, c) in s.chars().enumerate() {
2588 if i > 0 && (s.len() - i).is_multiple_of(3) {
2589 out.push(',');
2590 }
2591 out.push(c);
2592 }
2593 out
2594}
2595
2596fn human_bytes(n: u64) -> String {
2597 match n {
2598 n if n < 1024 => format!("{n} B"),
2599 n if n < 1024 * 1024 => format!("{:.1} KB", n as f64 / 1024.0),
2600 n => format!("{:.1} MB", n as f64 / (1024.0 * 1024.0)),
2601 }
2602}
2603
2604pub fn clean(t: &Path) -> std::io::Result<bool> {
2606 if t.exists() {
2607 std::fs::remove_dir_all(t)?;
2608 Ok(true)
2609 } else {
2610 Ok(false)
2611 }
2612}