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