use std::collections::BTreeMap;
use std::fs::File;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use arrow::array::RecordBatch;
use arrow::datatypes::SchemaRef;
use arrow::ipc::reader::FileReader;
use arrow::ipc::writer::FileWriter;
use chrono::NaiveDate;
use dre_protocol::host::{Execution, LogSink, PluginProcess};
use dre_protocol::msg::{ColumnOptions, DeliveryFile, ResultSetMeta};
use dre_protocol::{CAP_LOAD, CAP_MULTI_FILE, CAP_READ_ONLY, CAP_SESSIONS};
use serde::Serialize;
use serde_json::{Map as JsonMap, Value as Json, json};
use crate::dates::Calendar;
use crate::lookups::Table;
use crate::profiles::{BUILTIN_LOCAL, LOCAL_TYPE, ProfileTarget, Profiles, Role};
use crate::project::{Binding, PluginKind, Project, QueryEntry, Report};
use crate::render::{
Column, Connection, Connections, QueryRows, QueryRunner, RenderError, Renderer, RendererConfig,
RunContext,
};
use crate::selector;
use crate::sqlsplit::{self, StatementKind};
#[derive(Debug, Clone, Default)]
pub struct RunOptions {
pub selector: Option<String>,
pub set: Option<String>,
pub target: Option<String>,
pub profile: Option<String>,
pub vars: BTreeMap<String, String>,
pub output_name: Option<String>,
pub output_path: Option<String>,
pub dry_run: bool,
pub preview: Option<u64>,
pub accept_schema_change: bool,
pub interactive: bool,
pub date: Option<NaiveDate>,
pub live_check: bool,
pub schedule: Option<String>,
pub timezone: Option<String>,
pub manifest_checksum: Option<String>,
}
impl RunOptions {
pub fn params(&self, date: NaiveDate) -> Json {
json!({
"selector": self.selector,
"set": self.set,
"schedule": self.schedule,
"target": self.target,
"profile": self.profile,
"vars": self.vars,
"run_date": date.to_string(),
"timezone": self.timezone,
"output_name": self.output_name,
"output_path": self.output_path,
"dry_run": self.dry_run,
"preview": self.preview,
"accept_schema_change": self.accept_schema_change,
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum Level {
Debug,
Info,
}
pub trait Ui {
fn plan(&mut self, _bindings: usize) {}
fn binding_start(&mut self, _report: &str, _set: Option<&str>) {}
fn binding_vars(
&mut self,
_schedule: Option<&str>,
_schedule_vars: Option<&JsonMap<String, Json>>,
_vars: &JsonMap<String, Json>,
) {
}
fn step(&mut self, level: Level, verb: &str, detail: &str, elapsed: Option<Duration>);
fn warn(&mut self, msg: &str);
fn binding_end(&mut self, _outcome: &BindingOutcome) {}
fn compiled(&mut self, _plan: &BindingPlan) {}
fn choose_set(&mut self, report: &str, sets: &[String]) -> Result<Option<String>, String>;
fn plugin_log(&self) -> LogSink;
fn sql_log(&self) -> LogSink {
Arc::new(|_, _| {})
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum Status {
Success,
Error,
DryRun,
Checked,
}
#[derive(Debug, Clone, Serialize)]
pub struct BindingOutcome {
pub report: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub set: Option<String>,
pub binding: String,
pub status: Status,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
pub files: Vec<PathBuf>,
pub summary: String,
pub schedule: Option<String>,
pub schedule_vars: Option<JsonMap<String, Json>>,
pub vars: JsonMap<String, Json>,
#[serde(skip_serializing_if = "String::is_empty")]
pub timezone: String,
#[serde(skip)]
pub elapsed: Duration,
}
#[derive(Debug, Clone, Serialize)]
pub struct BindingPlan {
pub report: String,
pub set: Option<String>,
pub compiled: Vec<PathBuf>,
pub profile: String,
pub source_type: String,
pub target: String,
pub format: String,
pub output: PathBuf,
pub destinations: Vec<PlannedDestination>,
pub schedules: Vec<String>,
}
#[derive(Debug, Clone, Serialize)]
pub struct PlannedDestination {
pub profile: String,
pub kind: Option<String>,
pub target: Option<String>,
pub path: Option<String>,
pub delivers: bool,
}
#[derive(Debug, Default)]
pub struct RunSummary {
pub outcomes: Vec<BindingOutcome>,
pub error: Option<String>,
}
impl RunSummary {
pub fn failed(&self) -> bool {
self.error.is_some() || self.outcomes.iter().any(|o| o.status == Status::Error)
}
}
pub fn run(project: &Project, opts: &RunOptions, ui: &mut dyn Ui) -> RunSummary {
let mut summary = RunSummary::default();
if let Err(e) = crate::target::ensure(&project.target_dir) {
summary.error = Some(format!(
"can't write to the target path {} (from {}): {e}",
project.target_dir.display(),
project.target_source
));
return summary;
}
if let Some(name) = &opts.schedule {
if !project.schedules.iter().any(|e| &e.name == name) {
summary.error = Some(unknown_schedule(project, name));
return summary;
}
let planned: Vec<(&Report, Binding)> = project
.reports
.iter()
.flat_map(|r| {
r.bindings
.iter()
.filter(|b| b.schedules.contains(name))
.map(move |b| {
let mut b = b.clone();
if let Some(p) = &opts.profile {
b.profile = Some(p.clone());
}
(r, b)
})
})
.collect();
ui.plan(planned.len());
for (report, b) in &planned {
ui.binding_start(&report.name, b.set.as_deref());
let mut r = BindingRun::new(project, report, b, opts, ui);
let outcome = r.run();
ui.binding_end(&outcome);
summary.outcomes.push(outcome);
}
return summary;
}
let reports: Vec<&Report> = match &opts.selector {
None => project.reports.iter().collect(),
Some(s) => match selector::resolve(project, s) {
Ok(r) if r.is_empty() => {
summary.error = Some(format!("selector `{s}` matches no report"));
return summary;
}
Ok(r) => r,
Err(e) => {
summary.error = Some(e.to_string());
return summary;
}
},
};
let mut planned: Vec<(&Report, Binding)> = Vec::new();
for report in reports {
match choose_bindings(project, report, opts, ui) {
Ok(bs) => planned.extend(bs.into_iter().map(|b| (report, b))),
Err(e) => {
let outcome = BindingOutcome {
report: report.name.clone(),
set: None,
binding: "-".into(),
status: Status::Error,
error: Some(e),
files: Vec::new(),
summary: String::new(),
schedule: opts.schedule.clone(),
schedule_vars: None,
vars: JsonMap::new(),
timezone: String::new(),
elapsed: Duration::ZERO,
};
ui.binding_end(&outcome);
summary.outcomes.push(outcome);
}
}
}
ui.plan(planned.len());
for (report, b) in &planned {
ui.binding_start(&report.name, b.set.as_deref());
let mut r = BindingRun::new(project, report, b, opts, ui);
let outcome = r.run();
ui.binding_end(&outcome);
summary.outcomes.push(outcome);
}
summary
}
pub fn unknown_schedule(project: &Project, name: &str) -> String {
let names: Vec<&str> = project.schedules.iter().map(|e| e.name.as_str()).collect();
if names.is_empty() {
format!("no schedule `{name}`: this project's schedules.yml declares none")
} else {
format!("no schedule `{name}`; valid names: {}", names.join(", "))
}
}
pub fn choose_bindings(
project: &Project,
report: &Report,
opts: &RunOptions,
ui: &mut dyn Ui,
) -> Result<Vec<Binding>, String> {
let mut out: Vec<Binding> = match opts.set.as_deref() {
Some("all") => report.bindings.clone(),
Some(name) => match report.binding(name) {
Some(b) => vec![b.clone()],
None => vec![ad_hoc(project, report, name)],
},
None if !report.has_sets => report.bindings.clone(),
None => {
let names: Vec<String> = report.bindings.iter().filter_map(|b| b.set.clone()).collect();
let pick = report
.default_set
.clone()
.or_else(|| project.default_set.clone().filter(|d| names.contains(d)))
.or_else(|| (names.len() == 1).then(|| names[0].clone()));
match pick {
Some(p) => report.binding(&p).cloned().into_iter().collect(),
None if opts.interactive => match ui.choose_set(&report.name, &names)? {
Some(p) => report.binding(&p).cloned().into_iter().collect(),
None => report.bindings.clone(),
},
None => {
return Err(format!(
"report `{}` has several Sets ({}) and no `default_set`; declare `default_set:` or pass `--set <name>` or `--set all`",
report.name,
names.join(", ")
));
}
}
}
};
if let Some(p) = &opts.profile {
for b in &mut out {
b.profile = Some(p.clone());
}
}
Ok(out)
}
fn ad_hoc(project: &Project, report: &Report, name: &str) -> Binding {
let mut b = report.base.clone();
b.set = Some(name.to_string());
if let Some(reg) = project.sets.get(name) {
if let Some(p) = ®.profile {
b.profile = Some(p.clone());
}
b.vars.extend(reg.vars.clone());
}
b
}
struct Produced {
query: String,
schema: SchemaRef,
rows: u64,
spool: PathBuf,
name: String,
anchor: Option<String>,
header: Option<bool>,
columns: BTreeMap<String, ColumnOptions>,
}
struct Statement {
query: String,
file: PathBuf,
line: usize,
text: String,
kind: StatementKind,
tab: bool,
}
struct BindingRun<'a> {
project: &'a Project,
report: &'a Report,
b: &'a Binding,
vars: JsonMap<String, Json>,
rendered_vars: JsonMap<String, Json>,
schedule_vars: Option<JsonMap<String, Json>>,
opts: &'a RunOptions,
date: NaiveDate,
calendar: Calendar,
ui: &'a mut dyn Ui,
compiled_dir: PathBuf,
run_dir: PathBuf,
schema_dir: PathBuf,
target: String,
source_type: String,
started: Instant,
started_at: chrono::DateTime<chrono::Utc>,
produced: Vec<Produced>,
files: Vec<(PathBuf, Option<String>)>,
dests: Vec<RenderedDest>,
deliveries: Vec<Json>,
delivery_note: Option<String>,
drift: Vec<String>,
}
struct RenderedDest {
profile: String,
path: Option<String>,
options: JsonMap<String, Json>,
}
type Fail = String;
impl<'a> BindingRun<'a> {
fn new(
project: &'a Project,
report: &'a Report,
b: &'a Binding,
opts: &'a RunOptions,
ui: &'a mut dyn Ui,
) -> Self {
let t = &project.target_dir;
let rel = Path::new(&report.name).join(b.dir_name());
let schedule = opts
.schedule
.as_ref()
.and_then(|n| project.schedules.iter().find(|e| &e.name == n));
let schedule_vars = schedule.map(|e| e.vars.clone());
let tz = opts
.timezone
.as_ref()
.or(schedule.and_then(|e| e.timezone.as_ref()))
.or(report.timezone.as_ref())
.and_then(|t| crate::dates::parse_tz(t).ok())
.unwrap_or(chrono_tz::Tz::UTC);
let calendar = Calendar {
tz,
week_start: project.week_start,
numbering: project.week_numbering,
};
let date = opts.date.unwrap_or_else(|| calendar.today());
let mut vars = b.vars.clone();
vars.extend(schedule_vars.clone().unwrap_or_default());
let rendered_vars = {
let mut v = vars.clone();
v.extend(
opts.vars
.iter()
.map(|(k, x)| (k.clone(), Json::String(x.clone()))),
);
v
};
ui.binding_vars(opts.schedule.as_deref(), schedule_vars.as_ref(), &rendered_vars);
BindingRun {
project,
report,
b,
vars,
rendered_vars,
schedule_vars,
opts,
date,
calendar,
ui,
compiled_dir: t.join("compiled").join(&rel),
run_dir: t.join("run").join(&rel),
schema_dir: t.join("schema").join(&rel),
target: String::new(),
source_type: String::new(),
started: Instant::now(),
started_at: chrono::Utc::now(),
produced: Vec::new(),
files: Vec::new(),
dests: Vec::new(),
deliveries: Vec::new(),
delivery_note: None,
drift: Vec::new(),
}
}
fn connections(&self, source: Option<&str>) -> Arc<dyn Connections> {
Arc::new(ProfileConnections {
profiles: self.project.profiles.clone(),
source: source.map(str::to_string),
target: self.opts.target.clone(),
root: self.project.root.clone(),
plugins: self.project.plugins.clone(),
log: self.ui.plugin_log(),
})
}
fn outcome(&self, status: Status, error: Option<String>) -> BindingOutcome {
BindingOutcome {
report: self.report.name.clone(),
set: self.b.set.clone(),
binding: self.b.dir_name().to_string(),
status,
error,
files: self.files.iter().map(|(p, _)| p.clone()).collect(),
summary: self.summary_line(),
schedule: self.opts.schedule.clone(),
schedule_vars: self.schedule_vars.clone(),
vars: self.rendered_vars.clone(),
timezone: self.calendar.tz.name().to_string(),
elapsed: self.started.elapsed(),
}
}
fn run(&mut self) -> BindingOutcome {
let dry = self.opts.dry_run || self.opts.live_check;
if !dry {
let _ = std::fs::remove_dir_all(&self.run_dir);
}
let result = self.run_inner();
let status = match (&result, dry, self.opts.live_check) {
(Err(_), _, _) => Status::Error,
(Ok(()), _, true) => Status::Checked,
(Ok(()), true, _) => Status::DryRun,
(Ok(()), false, _) => Status::Success,
};
let err = result.err();
if !dry && let Err(e) = self.write_results(&status, err.as_deref()) {
self.ui.warn(&format!("can't write run_results.json: {e}"));
}
let _ = std::fs::remove_dir_all(self.run_dir.join(".spool"));
self.outcome(status, err)
}
fn run_inner(&mut self) -> Result<(), Fail> {
let profile_name = self
.b
.profile
.clone()
.ok_or("no source profile resolves for this Binding")?;
let profiles = &self.project.profiles;
let profile = profiles
.get(Role::Source, &profile_name)
.ok_or_else(|| format!("source profile `{profile_name}` isn't in profiles.yml"))?;
let target = self.opts.target.clone().unwrap_or_else(|| profile.target.clone());
let (_, output) = profiles
.target(Role::Source, &profile_name, Some(&target))
.ok_or_else(|| {
format!(
"source profile `{profile_name}` has no `{target}` target (it has: {})",
profile.targets.keys().cloned().collect::<Vec<_>>().join(", ")
)
})?;
self.target = target.clone();
self.source_type = output.kind.clone();
let source_path = find_plugin(self.project, PluginKind::Source, &output.kind).map_err(String::from);
let connection = render_connection(output);
let session = Arc::new(Mutex::new(Session::new(
source_path,
connection,
self.project.root.clone(),
self.ui.plugin_log(),
!self.report.managed,
)));
if !self.report.managed {
self.ui.warn(&format!(
"`{}` is an unmanaged report ({}): for quick tests only — add a YAML to make it a managed report",
self.report.name,
self.report.file.display()
));
}
let renderer = Renderer::new(RendererConfig {
root: &self.project.root,
macros: &self.project.macros,
context: RunContext {
report: self.report.name.clone(),
set: self.b.set.clone(),
target: target.clone(),
profile: profile_name.clone(),
source_type: output.kind.clone(),
schedule: self.opts.schedule.clone(),
date: self.date,
now: self.started_at,
calendar: self.calendar,
},
vars: self.vars.clone(),
cli_vars: self.opts.vars.clone(),
runner: Some(Arc::new(SessionRunner(session.clone(), self.ui.sql_log()))),
connections: Some(self.connections(Some(&profile_name))),
run_query_max_rows: self.project.run_query_max_rows,
sql: self.project.sql.clone(),
lookups: self.project.lookups.clone(),
lookup_inline_max_rows: self.project.lookup_inline_max_rows,
packages: self.project.packages.clone(),
project_name: self.project.name.clone(),
dispatch: self.project.dispatch.clone(),
})
.map_err(|e| e.to_string())?;
std::fs::create_dir_all(&self.compiled_dir).map_err(|e| e.to_string())?;
let interleave = !self.opts.dry_run && !self.opts.live_check && self.report.managed;
let mut statements = Vec::new();
let mut two_tabs: Option<String> = None;
if interleave {
self.dests = self.render_destinations(&renderer)?;
if self.b.queries.len() > 1 {
self.check_sessions(&session, self.b.queries.len(), &output.kind)?;
}
std::fs::create_dir_all(self.run_dir.join(".spool")).map_err(|e| e.to_string())?;
let mut i = 0;
for q in &self.b.queries {
let sts = self.render_query(&renderer, q, &mut two_tabs)?;
for w in renderer.take_warnings() {
self.ui.warn(&w);
}
if let Some(e) = two_tabs.take() {
return Err(e);
}
if i == 0 && sts.len() > 1 {
self.check_sessions(&session, sts.len(), &output.kind)?;
}
for st in &sts {
self.execute_statement(&session, i, st)?;
i += 1;
}
}
} else {
for q in &self.b.queries {
statements.extend(self.render_query(&renderer, q, &mut two_tabs)?);
}
self.dests = self.render_destinations(&renderer)?;
for w in renderer.take_warnings() {
self.ui.warn(&w);
}
if !self.report.managed {
if let Some(bad) = statements.iter().find(|s| !s.kind.is_read_only_safe()) {
return Err(format!(
"{}:{}: 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",
bad.file.display(),
bad.line,
self.report.name,
dre_protocol::util::summarize(&bad.text, 60)
));
}
let creates_temp = statements.iter().any(|s| s.kind == StatementKind::TempCreate)
|| session.lock().unwrap().loaded;
session.lock().unwrap().want_read_only(!creates_temp);
}
if let Some(e) = two_tabs {
return Err(e);
}
if self.opts.dry_run {
let plan = self.plan(&profile_name, &output.kind);
self.ui.compiled(&plan);
return Ok(());
}
if statements.len() > 1 {
self.check_sessions(&session, statements.len(), &output.kind)?;
}
if self.opts.live_check {
return self.live_check(&session, &statements, &output.kind);
}
std::fs::create_dir_all(self.run_dir.join(".spool")).map_err(|e| e.to_string())?;
for (i, st) in statements.iter().enumerate() {
self.execute_statement(&session, i, st)?;
}
}
if let Ok(mut s) = session.lock() {
s.close();
}
self.name_result_sets()?;
let filename = self.file_names();
self.format(&filename)?;
if self.opts.preview.is_none() {
self.drift = self.schema_drift();
if !self.drift.is_empty() {
if self.opts.accept_schema_change {
self.ui.warn(&format!(
" schema changed since the last successful run (accepted): {}",
self.drift.join("; ")
));
} else {
return Err(format!(
"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",
self.drift.join("; "),
rel(&self.project.root, &self.run_dir).display()
));
}
}
}
if self.opts.preview.is_some() {
self.delivery_note = Some("preview: not delivered".into());
let dir = rel(&self.project.root, &self.run_dir);
self.ui.step(
Level::Info,
"Preview",
&format!("not delivered; output stays in {}", dir.display()),
None,
);
} else {
self.deliver()?;
}
if self.opts.preview.is_none() {
self.write_snapshot()
.map_err(|e| format!("can't write the schema snapshot: {e}"))?;
}
Ok(())
}
fn render_query(
&mut self,
renderer: &Renderer,
q: &QueryEntry,
two_tabs: &mut Option<String>,
) -> Result<Vec<Statement>, Fail> {
let mut out = Vec::new();
let src = std::fs::read_to_string(self.project.root.join(&q.path))
.map_err(|e| format!("{}: {e}", q.path.display()))?;
let sql = renderer
.render(&q.path, &src)
.map_err(|e: RenderError| e.to_string())?;
std::fs::write(
self.compiled_dir.join(format!("{}.sql", q.query)),
crate::secrets::mask(&sql).as_bytes(),
)
.map_err(|e| e.to_string())?;
self.ui
.step(Level::Debug, "Rendered", &q.path.display().to_string(), None);
let parts = sqlsplit::split(&sql);
if q.tab && parts.is_empty() {
return Err(format!(
"{}: `{}` makes a tab but has no statements; add a query, or `tab: false` if it's meant to be empty",
q.path.display(),
q.query
));
}
let n = parts.len();
for (i, st) in parts.into_iter().enumerate() {
let body = sqlsplit::strip_leading_comments(&st.text);
let line = st.line + st.text[..st.text.len() - body.len()].matches('\n').count();
let kind = sqlsplit::classify(&st.text);
let last = i + 1 == n;
if q.tab && !last && kind == StatementKind::Read && two_tabs.is_none() {
*two_tabs = Some(format!(
"{}:{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:`",
q.path.display(),
q.query
));
}
out.push(Statement {
query: q.query.clone(),
file: q.path.clone(),
line,
kind,
text: st.text,
tab: q.tab && last,
});
}
Ok(out)
}
fn check_sessions(&self, session: &Arc<Mutex<Session>>, n: usize, kind: &str) -> Result<(), Fail> {
let mut s = session.lock().unwrap();
if !s.get()?.has(CAP_SESSIONS) {
return Err(format!(
"this Binding runs {n} statements, but the `{kind}` source plugin can't hold one session across them; nothing was run"
));
}
Ok(())
}
fn execute_statement(
&mut self,
session: &Arc<Mutex<Session>>,
i: usize,
st: &Statement,
) -> Result<(), Fail> {
let sql_log = self.ui.sql_log();
sql_log(&format!("{}:{}", st.file.display(), st.line), &st.text);
let spool_path = self.run_dir.join(".spool").join(format!("{i}.arrow"));
let mut writer: Option<FileWriter<File>> = None;
let t = Instant::now();
let exec = {
let mut s = session.lock().unwrap();
let p = s.get()?;
p.execute(&st.text, self.opts.preview, |schema, batch| {
if writer.is_none() {
let f = File::create(&spool_path).map_err(|e| e.to_string())?;
writer = Some(FileWriter::try_new(f, schema).map_err(|e| e.to_string())?);
}
writer.as_mut().unwrap().write(&batch).map_err(|e| e.to_string())
})
.map_err(|e| format!("{}:{}: {e}", st.file.display(), st.line))?
};
let what = match &exec {
Execution::Result { rows, .. } => {
format!("{} row{}", thousands(*rows), if *rows == 1 { "" } else { "s" })
}
Execution::NoResult {
rows_affected: Some(n),
} => format!("no result set ({n} affected)"),
Execution::NoResult { .. } => "no result set".to_string(),
};
let at = format!("{}:{}", st.file.display(), st.line);
self.ui.step(
Level::Debug,
"Executed",
&format!("{at} {what}"),
Some(t.elapsed()),
);
let (schema, rows) = match (st.tab, exec) {
(true, Execution::Result { schema, rows }) => (schema, rows),
(true, Execution::NoResult { .. }) => {
return Err(format!(
"{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",
st.query
));
}
(false, _) => {
drop(writer);
let _ = std::fs::remove_file(&spool_path);
return Ok(());
}
};
{
let mut w = match writer {
Some(w) => w,
None => FileWriter::try_new(File::create(&spool_path).map_err(|e| e.to_string())?, &schema)
.map_err(|e| e.to_string())?,
};
w.finish().map_err(|e| e.to_string())?;
self.produced.push(Produced {
query: st.query.clone(),
schema,
rows,
spool: spool_path,
name: String::new(),
anchor: None,
header: None,
columns: Default::default(),
});
}
Ok(())
}
fn render_destinations(&self, renderer: &Renderer) -> Result<Vec<RenderedDest>, Fail> {
self.b
.output
.destinations
.iter()
.map(|d| {
let path = match &d.path {
Some(p) => Some(
renderer
.render(&self.report.file, p)
.map_err(|e| format!("output path: {e}"))?,
),
None => None,
};
let mut options = JsonMap::new();
for (k, v) in &d.options {
let v = render_json(renderer, &self.report.file, v)
.map_err(|e| format!("destination `{}` option `{k}`: {e}", d.profile))?;
options.insert(k.clone(), v);
}
Ok(RenderedDest {
profile: d.profile.clone(),
path,
options,
})
})
.collect()
}
fn name_result_sets(&mut self) -> Result<(), Fail> {
let entries: BTreeMap<&str, &QueryEntry> =
self.b.queries.iter().map(|q| (q.query.as_str(), q)).collect();
for p in &mut self.produced {
let q = entries[p.query.as_str()];
p.name = q.tab_name.clone().unwrap_or_else(|| q.query.clone());
p.anchor = q.anchor.clone();
p.header = q.header;
p.columns = q.columns.clone();
}
if self.b.output.format == "xlsx" {
let mut seen: BTreeMap<String, String> = BTreeMap::new();
for p in &self.produced {
check_sheet_name(&p.name)?;
if let Some(prev) = seen.insert(p.name.to_lowercase(), p.query.clone()) {
return Err(format!(
"sheet name `{}` is used twice (by `{prev}` and `{}`); Excel sheet names must be unique",
p.name, p.query
));
}
}
}
Ok(())
}
fn plan(&mut self, profile: &str, source_type: &str) -> BindingPlan {
let file = self.file_names();
let compiled = self
.b
.queries
.iter()
.map(|q| {
rel(
&self.project.root,
&self.compiled_dir.join(format!("{}.sql", q.query)),
)
})
.collect();
let destinations = self
.dests
.iter()
.map(|d| {
let resolved = self.dest_output(&d.profile);
PlannedDestination {
profile: d.profile.clone(),
kind: resolved.as_ref().map(|(_, o)| o.kind.clone()),
target: resolved.map(|(t, _)| t).or_else(|| self.dest_target(&d.profile)),
path: d.path.clone(),
delivers: self.dest_output(&d.profile).is_some(),
}
})
.collect();
BindingPlan {
report: self.report.name.clone(),
set: self.b.set.clone(),
compiled,
profile: profile.to_string(),
source_type: source_type.to_string(),
target: self.target.clone(),
format: self.b.output.format.clone(),
output: rel(&self.project.root, &self.run_dir.join(file)),
destinations,
schedules: self.b.schedules.clone(),
}
}
fn file_names(&mut self) -> String {
for d in self.dests.iter_mut().filter(|d| d.path.is_some()) {
if let Some(p) = &self.opts.output_path {
d.path = Some(p.clone());
}
if let (Some(n), Some(r)) = (&self.opts.output_name, d.path.as_mut()) {
*r = match r.rfind(['/', '\\']) {
Some(i) => format!("{}{n}", &r[..=i]),
None => n.clone(),
};
}
}
let ext = match &self.b.output.extension {
Some(e) => e.as_str(),
None => extension(&self.b.output.format),
};
let from_remote = self
.dests
.iter()
.find_map(|d| d.path.as_deref())
.and_then(|r| r.rsplit(['/', '\\']).next())
.filter(|s| !s.is_empty())
.map(str::to_string);
self.opts
.output_name
.clone()
.or(from_remote)
.unwrap_or_else(|| match ext {
"" => self.report.name.clone(),
ext => format!("{}.{ext}", self.report.name),
})
}
fn format(&mut self, filename: &str) -> Result<(), Fail> {
std::fs::create_dir_all(&self.run_dir).map_err(|e| e.to_string())?;
if self.produced.is_empty() {
self.ui
.warn(" no query makes a tab (every query has `tab: false`); no output file was written");
return Ok(());
}
let out = &self.b.output;
let plugin = find_plugin(self.project, PluginKind::Format, &out.format)?;
let mut p = plugin
.start(self.ui.plugin_log(), Some(&self.project.root))
.map_err(|e| e.to_string())?;
let multi = out.format == "xlsx" || out.template.is_some() || !is_single_table(&out.format);
let groups: Vec<(String, Vec<usize>)> = if multi || self.produced.len() == 1 {
vec![(filename.to_string(), (0..self.produced.len()).collect())]
} else {
let (stem, ext) = split_ext(filename);
let mut g = Vec::new();
for (i, r) in self.produced.iter().enumerate() {
if r.name.contains(['/', '\\']) {
return Err(format!("`{}` can't be used in a file name", r.name));
}
g.push((format!("{stem}_{}{ext}", r.name), vec![i]));
}
g
};
let template = match &out.template {
Some(t) => Some(self.template_payload(t)?),
None => None,
};
for (name, idxs) in groups {
let t = Instant::now();
let path = self.run_dir.join(&name);
let metas: Vec<ResultSetMeta> = idxs
.iter()
.map(|&i| {
let r = &self.produced[i];
ResultSetMeta {
name: r.name.clone(),
query: r.query.clone(),
result_index: 1,
anchor: r.anchor.clone(),
header: r.header,
columns: r.columns.clone(),
}
})
.collect();
p.write_begin(
&path.to_string_lossy(),
&out.format,
out.options.clone(),
metas,
template.clone(),
)
.map_err(|e| e.to_string())?;
for &i in &idxs {
let r = &self.produced[i];
let reader = FileReader::try_new(File::open(&r.spool).map_err(|e| e.to_string())?, None)
.map_err(|e| e.to_string())?;
let schema = r.schema.clone();
let mut any = false;
for b in reader {
let b = b.map_err(|e| e.to_string())?;
any = true;
p.send_batch(&b)
.map_err(|e| format!("{} format: {e}", out.format))?;
}
if !any {
p.send_batch(&RecordBatch::new_empty(schema))
.map_err(|e| format!("{} format: {e}", out.format))?;
}
p.send(&dre_protocol::msg::Request::ResultSetEnd {})
.map_err(|e| format!("{} format: {e}", out.format))?;
}
let (files, warnings) = p
.write_finish_with_warnings()
.map_err(|e| format!("{} format: {e}", out.format))?;
for w in warnings {
self.ui.warn(&format!(" {} format: {w}", out.format));
}
for f in files {
let f = PathBuf::from(f);
let size = std::fs::metadata(&f).map(|m| m.len()).unwrap_or(0);
let shown = rel(&self.project.root, &f);
self.ui.step(
Level::Debug,
"Wrote",
&format!("{} ({})", shown.display(), human_bytes(size)),
Some(t.elapsed()),
);
self.files.push((f, None));
}
}
let _ = p.close();
Ok(())
}
fn template_payload(&self, t: &crate::project::Template) -> Result<Json, Fail> {
let root = &self.project.root;
let file = [root.join(&t.file), root.join("templates").join(&t.file)]
.into_iter()
.find(|p| p.is_file())
.ok_or_else(|| format!("template file `{}` doesn't exist", t.file))?;
let renderer = Renderer::new(RendererConfig {
root,
macros: &self.project.macros,
context: RunContext {
report: self.report.name.clone(),
set: self.b.set.clone(),
target: self.target.clone(),
profile: self.b.profile.clone().unwrap_or_default(),
source_type: self.source_type.clone(),
schedule: self.opts.schedule.clone(),
date: self.date,
now: self.started_at,
calendar: self.calendar,
},
vars: self.vars.clone(),
cli_vars: self.opts.vars.clone(),
runner: None,
connections: Some(self.connections(self.b.profile.as_deref())),
run_query_max_rows: self.project.run_query_max_rows,
sql: self.project.sql.clone(),
lookups: self.project.lookups.clone(),
lookup_inline_max_rows: self.project.lookup_inline_max_rows,
packages: self.project.packages.clone(),
project_name: self.project.name.clone(),
dispatch: self.project.dispatch.clone(),
})
.map_err(|e| e.to_string())?;
let mut values = JsonMap::new();
for b in &t.bindings {
if let (Some(cell), Some(v)) = (&b.cell, &b.value) {
let rendered = renderer
.render(&self.report.file, v)
.map_err(|e| format!("template value: {e}"))?;
values.insert(format!("{}!{}", b.sheet, cell), Json::String(rendered));
}
}
Ok(json!({"file": file.to_string_lossy(), "bindings": t.bindings, "values": values}))
}
fn deliver(&mut self) -> Result<(), Fail> {
if self.dests.is_empty() {
self.delivery_note = Some("no destination declared; output stays in target/".into());
let dir = rel(&self.project.root, &self.run_dir);
self.ui.step(
Level::Debug,
"Kept",
&format!("no destination declared: output stays in {}", dir.display()),
None,
);
return Ok(());
}
if self.files.is_empty() {
return Ok(());
}
let dests = std::mem::take(&mut self.dests);
let mut failures = Vec::new();
let mut skipped = Vec::new();
for d in &dests {
let kind = self
.dest_output(&d.profile)
.map(|(_, o)| o.kind.clone())
.or_else(|| {
let p = self.project.profiles.get(Role::Destination, &d.profile)?;
p.targets
.get(&p.target)
.or(p.targets.values().next())
.map(|o| o.kind.clone())
});
let (status, location, error) = match self.deliver_one(d) {
Ok(Some(loc)) => ("delivered", Some(loc), None),
Ok(None) => {
skipped.push(self.skip_note(&d.profile));
("skipped", None, None)
}
Err(e) => {
failures.push(e.clone());
("failed", None, Some(e))
}
};
let mut record = json!({"profile": d.profile, "type": kind, "status": status});
if let Some(l) = location {
record["location"] = json!(l);
}
if let Some(e) = error {
record["error"] = json!(e);
}
self.deliveries.push(record);
}
self.dests = dests;
if self.files.iter().all(|(_, d)| d.is_none()) && !skipped.is_empty() {
self.delivery_note = Some(skipped.join("; "));
}
match failures.len() {
0 => Ok(()),
1 if self.dests.len() == 1 => Err(failures.remove(0)),
n => Err(format!(
"{n} of {} destinations failed: {}",
self.dests.len(),
failures.join("; ")
)),
}
}
fn dest_output(&self, profile: &str) -> Option<(String, &ProfileTarget)> {
if self.project.profiles.is_builtin_local(profile) {
return Some((self.target.clone(), &BUILTIN_LOCAL));
}
let dtarget = self.dest_target(profile)?;
self.project
.profiles
.target(Role::Destination, profile, Some(&dtarget))
.map(|(_, o)| (dtarget, o))
}
fn dest_target(&self, profile: &str) -> Option<String> {
let p = self.project.profiles.get(Role::Destination, profile)?;
Some(self.opts.target.clone().unwrap_or_else(|| p.target.clone()))
}
fn skip_note(&self, profile: &str) -> String {
let dtarget = self.dest_target(profile).unwrap_or_default();
format!(
"destination profile `{profile}` has no `{dtarget}` target: not delivered, output stays in target/"
)
}
fn deliver_one(&mut self, d: &RenderedDest) -> Result<Option<String>, Fail> {
if self.project.profiles.get(Role::Destination, &d.profile).is_none()
&& !self.project.profiles.is_builtin_local(&d.profile)
{
return Err(format!(
"destination profile `{}` isn't in profiles.yml",
d.profile
));
}
let Some((_, out)) = self.dest_output(&d.profile) else {
let note = self.skip_note(&d.profile);
self.ui.step(Level::Info, "Kept", ¬e, None);
return Ok(None);
};
let kind = out.kind.clone();
let connection = render_connection(out)?;
let targets: Vec<DeliveryFile> = self
.files
.iter()
.map(|(f, _)| {
let name = f.file_name().unwrap().to_string_lossy().to_string();
let r = d.path.as_deref().map(|r| {
if self.files.len() == 1 {
r.to_string()
} else {
match r.rfind(['/', '\\']) {
Some(i) => format!("{}{name}", &r[..=i]),
None => name.clone(),
}
}
});
DeliveryFile {
local_path: f.to_string_lossy().to_string(),
remote_path: r,
}
})
.collect();
let mut locations = Vec::new();
if kind == LOCAL_TYPE {
if let Some(k) = d.options.keys().next() {
return Err(format!(
"the local destination takes no options, but `{}` has `{k}`; check the key's spelling",
d.profile
));
}
for (i, f) in targets.iter().enumerate() {
let t = Instant::now();
let r = f
.remote_path
.as_ref()
.ok_or("the local destination needs `output.destination.path`")?;
let dst = self.project.root.join(r);
if let Some(parent) = dst.parent() {
std::fs::create_dir_all(parent)
.map_err(|e| format!("can't create {}: {e}", parent.display()))?;
}
std::fs::copy(&f.local_path, &dst).map_err(|e| {
format!(
"delivery to {} failed: {e}; the output is still in target/",
dst.display()
)
})?;
let loc = dst.to_string_lossy().to_string();
self.ui.step(Level::Debug, "Delivered", &loc, Some(t.elapsed()));
self.files[i].1.get_or_insert_with(|| loc.clone());
locations.push(loc);
}
return Ok(Some(locations.join(", ")));
}
let failed = |e: dre_protocol::host::HostError| {
format!("delivery through `{kind}` failed: {e}; the output is still in target/")
};
let plugin = find_plugin(self.project, PluginKind::Destination, &kind)?;
let mut p = plugin
.start(self.ui.plugin_log(), Some(&self.project.root))
.map_err(|e| e.to_string())?;
let batches: Vec<Vec<usize>> = if targets.len() > 1 && p.has(CAP_MULTI_FILE) {
vec![(0..targets.len()).collect()]
} else {
(0..targets.len()).map(|i| vec![i]).collect()
};
for batch in batches {
let t = Instant::now();
let files: Vec<DeliveryFile> = batch.iter().map(|&i| targets[i].clone()).collect();
let loc = p
.deliver_files(&files, connection.clone(), d.options.clone())
.map_err(failed)?;
self.ui.step(Level::Debug, "Delivered", &loc, Some(t.elapsed()));
for i in batch {
self.files[i].1.get_or_insert_with(|| loc.clone());
}
locations.push(loc);
}
let _ = p.close();
Ok(Some(locations.join(", ")))
}
fn snapshot(&self) -> Json {
let sets: Vec<Json> = self
.produced
.iter()
.map(|p| {
json!({
"name": p.name,
"columns": p.schema.fields().iter().map(|f| json!({"name": f.name(), "type": f.data_type().to_string()})).collect::<Vec<_>>(),
})
})
.collect();
json!({ "result_sets": sets })
}
fn schema_drift(&self) -> Vec<String> {
let path = self.schema_dir.join("last_success.json");
let Ok(prev) = std::fs::read_to_string(&path) else {
return Vec::new();
};
let Ok(prev) = serde_json::from_str::<Json>(&prev) else {
return Vec::new();
};
let cols = |v: &Json| -> BTreeMap<String, Vec<(String, String)>> {
v["result_sets"]
.as_array()
.into_iter()
.flatten()
.map(|s| {
let c = s["columns"]
.as_array()
.into_iter()
.flatten()
.map(|c| {
(
c["name"].as_str().unwrap_or("").to_string(),
c["type"].as_str().unwrap_or("").to_string(),
)
})
.collect();
(s["name"].as_str().unwrap_or("").to_string(), c)
})
.collect()
};
let (old, new) = (cols(&prev), cols(&self.snapshot()));
let mut drift = Vec::new();
for (name, oc) in &old {
let Some(nc) = new.get(name) else {
drift.push(format!("result set `{name}` is gone"));
continue;
};
let om: BTreeMap<_, _> = oc.iter().cloned().collect();
let nm: BTreeMap<_, _> = nc.iter().cloned().collect();
for (c, t) in &om {
match nm.get(c) {
None => drift.push(format!("`{name}`: column `{c}` removed")),
Some(t2) if t2 != t => {
drift.push(format!("`{name}`: column `{c}` changed type from {t} to {t2}"))
}
_ => {}
}
}
for c in nm.keys().filter(|c| !om.contains_key(*c)) {
drift.push(format!("`{name}`: column `{c}` added"));
}
}
for name in new.keys().filter(|n| !old.contains_key(*n)) {
drift.push(format!("result set `{name}` is new"));
}
drift
}
fn write_snapshot(&self) -> std::io::Result<()> {
std::fs::create_dir_all(&self.schema_dir)?;
std::fs::write(
self.schema_dir.join("last_success.json"),
serde_json::to_string_pretty(&self.snapshot())?,
)
}
fn write_results(&self, status: &Status, error: Option<&str>) -> std::io::Result<()> {
std::fs::create_dir_all(&self.run_dir)?;
let outputs: Vec<Json> = self
.files
.iter()
.map(|(f, delivered)| {
json!({
"path": record_path(self.project, f),
"size": std::fs::metadata(f).map(|m| m.len()).unwrap_or(0),
"delivered_to": delivered,
})
})
.collect();
let results = json!({
"report": self.report.name,
"set": self.b.set,
"binding": self.b.dir_name(),
"managed": self.report.managed,
"profile": self.b.profile,
"target": if self.target.is_empty() { Json::Null } else { json!(self.target) },
"schedule": self.opts.schedule,
"schedule_vars": self.schedule_vars,
"vars": self.rendered_vars,
"run_date": self.date.to_string(),
"timezone": self.calendar.tz.name(),
"params": self.opts.params(self.date),
"status": status,
"error": error,
"preview": self.opts.preview.is_some(),
"row_limit": self.opts.preview,
"started_at": self.started_at.to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
"duration_ms": self.started.elapsed().as_millis() as u64,
"result_sets": self.produced.iter().map(|p| json!({
"name": p.name,
"query": p.query,
"rows": p.rows,
"columns": p.schema.fields().iter().map(|f| f.name().clone()).collect::<Vec<_>>(),
})).collect::<Vec<_>>(),
"outputs": outputs,
"delivery": self.delivery_note,
"deliveries": self.deliveries,
"schema_drift": self.drift,
"target_path": self.project.target_dir,
"manifest_checksum": self.opts.manifest_checksum,
});
std::fs::write(
self.run_dir.join("run_results.json"),
(crate::secrets::to_json_pretty(&results)? + "\n").as_bytes(),
)
}
fn live_check(
&mut self,
session: &Arc<Mutex<Session>>,
statements: &[Statement],
kind: &str,
) -> Result<(), Fail> {
let mut s = session.lock().unwrap();
let p = s.get()?;
if !p.has(dre_protocol::CAP_CHECK) {
self.ui.warn(&format!(
" not checkable: the `{kind}` source plugin can't check statements without running them"
));
return Ok(());
}
let mut failures = Vec::new();
let mut unexecuted_setup: Option<(PathBuf, usize)> = None;
let sql_log = self.ui.sql_log();
for st in statements {
let verb = if st.kind == StatementKind::TempCreate {
""
} else {
" (check)"
};
sql_log(&format!("{}:{}{verb}", st.file.display(), st.line), &st.text);
if st.kind == StatementKind::TempCreate {
if let Err(e) = p.execute(&st.text, None, |_, _| Ok(())) {
failures.push(format!("{}:{}: {e}", st.file.display(), st.line));
}
continue;
}
match p.check(&st.text) {
Ok(()) => {}
Err(e) => {
let mut msg = format!("{}:{}: {e}", st.file.display(), st.line);
if let Some((f, l)) = &unexecuted_setup {
msg.push_str(&format!(
" (may be a false positive: the setup statement at {}:{l} wasn't executed during the check)",
f.display()
));
}
failures.push(msg);
}
}
if st.kind == StatementKind::Other {
unexecuted_setup = Some((st.file.clone(), st.line));
}
}
if failures.is_empty() {
Ok(())
} else {
Err(failures.join("\n "))
}
}
}
struct Session {
path: Result<crate::plugins::Located, String>,
connection: Result<JsonMap<String, Json>, String>,
cwd: PathBuf,
log: LogSink,
unmanaged: bool,
read_only: bool,
loaded: bool,
proc_: Option<PluginProcess>,
}
impl Session {
fn new(
path: Result<crate::plugins::Located, String>,
connection: Result<JsonMap<String, Json>, String>,
cwd: PathBuf,
log: LogSink,
unmanaged: bool,
) -> Session {
Session {
path,
connection,
cwd,
log,
unmanaged,
read_only: unmanaged,
loaded: false,
proc_: None,
}
}
fn want_read_only(&mut self, ro: bool) {
if self.read_only != ro {
self.close();
self.read_only = ro;
}
}
fn get(&mut self) -> Result<&mut PluginProcess, String> {
if self.proc_.is_none() {
let plugin = self.path.clone()?;
let connection = self.connection.clone()?;
let mut p = plugin
.start(self.log.clone(), Some(&self.cwd))
.map_err(|e| e.to_string())?;
let ro = self.unmanaged && self.read_only && p.has(CAP_READ_ONLY);
p.open(connection, ro)
.map_err(|e| format!("can't open the source connection: {e}"))?;
self.proc_ = Some(p);
}
Ok(self.proc_.as_mut().unwrap())
}
fn close(&mut self) {
if let Some(p) = self.proc_.take() {
let _ = p.close();
}
}
}
struct SessionRunner(Arc<Mutex<Session>>, LogSink);
impl QueryRunner for SessionRunner {
fn run_query(&self, sql: &str, max_rows: u64) -> Result<QueryRows, String> {
let mut s = self.0.lock().map_err(|_| "session lock poisoned".to_string())?;
if s.unmanaged
&& let Some(bad) = sqlsplit::split(sql)
.into_iter()
.find(|st| sqlsplit::classify(&st.text) != StatementKind::Read)
{
return Err(format!(
"run_query() in an unmanaged report may only read, but got `{}`; give the report a YAML to declare it",
dre_protocol::util::summarize(&bad.text, 60)
));
}
(self.1)("run_query()", sql);
let p = s.get()?;
let mut out = QueryRows::default();
let mut too_many = false;
let exec = p
.execute(sql, Some(max_rows + 1), |schema, batch| {
if out.columns.is_empty() {
out.columns = schema.fields().iter().map(|f| f.name().clone()).collect();
}
if out.rows.len() as u64 + batch.num_rows() as u64 > max_rows {
too_many = true;
return Ok(());
}
out.rows.extend(crate::values::batch_rows(&batch));
Ok(())
})
.map_err(|e| format!("run_query() failed: {e}"))?;
if too_many {
return Err(format!(
"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"
));
}
if let Execution::Result { schema, .. } = exec
&& out.columns.is_empty()
{
out.columns = schema.fields().iter().map(|f| f.name().clone()).collect();
}
Ok(out)
}
fn columns(&self, sql: &str) -> Result<Vec<Column>, String> {
let mut s = self.0.lock().map_err(|_| "session lock poisoned".to_string())?;
(self.1)("columns()", sql);
let p = s.get()?;
let mut schema: Option<SchemaRef> = None;
let exec = p
.execute(sql, Some(1), |sch, _| {
schema.get_or_insert_with(|| sch.clone());
Ok(())
})
.map_err(|e| e.to_string())?;
if let Execution::Result { schema: sch, .. } = exec {
schema.get_or_insert(sch);
}
let schema = schema.ok_or("the query returned no result set")?;
Ok(schema
.fields()
.iter()
.map(|f| Column {
name: f.name().clone(),
data_type: f.data_type().to_string(),
})
.collect())
}
fn load(&self, name: &str, table: &Table) -> Result<Option<(String, Option<String>)>, String> {
let mut s = self.0.lock().map_err(|_| "session lock poisoned".to_string())?;
s.want_read_only(false);
if !s.get()?.has(CAP_LOAD) {
return Ok(None);
}
let batch = table.to_batch()?;
let schema = batch.schema();
(self.1)(
&format!("lookup `{name}`"),
&format!(
"-- {} rows loaded through the plugin's `load` request",
batch.num_rows()
),
);
let r = s.get()?.load(name, &schema, [batch]).map_err(|e| e.to_string())?;
s.loaded = true;
Ok(Some(r))
}
}
fn render_json(renderer: &Renderer, file: &Path, v: &Json) -> Result<Json, RenderError> {
Ok(match v {
Json::String(s) => Json::String(renderer.render(file, s)?),
Json::Array(a) => Json::Array(
a.iter()
.map(|v| render_json(renderer, file, v))
.collect::<Result<_, _>>()?,
),
Json::Object(o) => Json::Object(
o.iter()
.map(|(k, v)| Ok((k.clone(), render_json(renderer, file, v)?)))
.collect::<Result<_, RenderError>>()?,
),
other => other.clone(),
})
}
fn render_connection(output: &ProfileTarget) -> Result<JsonMap<String, Json>, String> {
let mut env = minijinja::Environment::new();
env.set_undefined_behavior(minijinja::UndefinedBehavior::Strict);
env.add_function("env_var", |name: String, default: Option<String>| -> Result<String, minijinja::Error> {
std::env::var(&name).ok().or(default).ok_or_else(|| {
minijinja::Error::new(
minijinja::ErrorKind::UndefinedError,
format!("`env_var('{name}')`: environment variable `{name}` is not set and no default is given"),
)
})
});
fn walk(env: &minijinja::Environment<'_>, v: &Json) -> Result<Json, String> {
Ok(match v {
Json::String(s) if crate::preflight::is_templated(s) => Json::String(
env.render_str(s, ())
.map_err(|e| format!("profiles.yml: {}", e.detail().unwrap_or("render error")))?,
),
Json::Array(a) => Json::Array(a.iter().map(|x| walk(env, x)).collect::<Result<_, _>>()?),
Json::Object(o) => Json::Object(
o.iter()
.map(|(k, x)| Ok((k.clone(), walk(env, x)?)))
.collect::<Result<_, String>>()?,
),
other => other.clone(),
})
}
match walk(&env, &Json::Object(output.fields.clone()))? {
Json::Object(o) => Ok(o),
_ => unreachable!(),
}
}
pub fn find_plugin(
project: &Project,
kind: PluginKind,
name: &str,
) -> Result<crate::plugins::Located, crate::plugins::LocateError> {
crate::plugins::locate(project, &crate::project::PluginId::new(kind, name))
}
struct ProfileConnections {
profiles: Profiles,
source: Option<String>,
target: Option<String>,
root: PathBuf,
plugins: Vec<crate::project::PluginRequirement>,
log: LogSink,
}
static DESCRIBED: std::sync::LazyLock<Mutex<BTreeMap<DescribeKey, SecretFields>>> =
std::sync::LazyLock::new(Mutex::default);
type DescribeKey = (PathBuf, Role, String);
type SecretFields = Option<Vec<String>>;
const SECRET_NAME_WORDS: &[&str] = &["password", "secret", "token", "key", "credential"];
impl ProfileConnections {
fn view(&self, role: Role, name: &str) -> Result<Connection, String> {
let (target, out): (String, ProfileTarget) = match self.profiles.get(role, name) {
Some(p) => {
let t = self
.target
.as_ref()
.filter(|t| p.targets.contains_key(t.as_str()))
.unwrap_or(&p.target);
let o = p
.targets
.get(t)
.ok_or_else(|| format!("profile `{name}` has no `{t}` target"))?;
(t.clone(), o.clone())
}
None if role == Role::Destination && self.profiles.is_builtin_local(name) => {
("default".into(), BUILTIN_LOCAL.clone())
}
None => {
return Err(format!(
"no {} profile `{name}` in {}",
role.as_str(),
self.profiles.path.display()
));
}
};
let mut secrets: Vec<String> = out
.fields
.iter()
.filter(|(_, v)| v.as_str().is_some_and(|s| s.contains(crate::secrets::PREFIX)))
.map(|(k, _)| k.clone())
.collect();
match self.described(role, &out.kind) {
Some(named) => secrets.extend(named),
None => secrets.extend(
out.fields
.keys()
.filter(|k| {
let k = k.to_lowercase();
SECRET_NAME_WORDS.iter().any(|w| k.contains(w))
})
.cloned(),
),
}
let fields = render_connection(&out)?;
Ok(Connection {
profile: name.to_string(),
target,
kind: out.kind.clone(),
fields,
secrets,
})
}
fn described(&self, role: Role, kind: &str) -> Option<Vec<String>> {
let key = (self.root.clone(), role, kind.to_string());
if let Some(v) = DESCRIBED.lock().unwrap().get(&key) {
return v.clone();
}
let plugin_kind = match role {
Role::Source => PluginKind::Source,
Role::Destination => PluginKind::Destination,
};
let asked = (|| {
if kind == LOCAL_TYPE {
return Some(Vec::new());
}
let id = crate::project::PluginId::new(plugin_kind, kind);
let plugin = crate::plugins::locate_in(&self.root, &self.plugins, &id).ok()?;
let mut p = plugin.start(self.log.clone(), Some(&self.root)).ok()?;
let fields = p.describe().ok();
let _ = p.close();
Some(fields?.into_iter().filter(|f| f.secret).map(|f| f.name).collect())
})();
DESCRIBED.lock().unwrap().insert(key, asked.clone());
asked
}
}
impl Connections for ProfileConnections {
fn source(&self) -> Result<Connection, String> {
let name = self
.source
.as_deref()
.ok_or("no source profile resolves for this Binding")?;
self.view(Role::Source, name)
}
fn profile(&self, name: &str, role: Option<&str>) -> Result<Connection, String> {
let role = match role {
Some("source") => Role::Source,
Some(_) => Role::Destination,
None => {
let s = self.profiles.get(Role::Source, name).is_some();
let d = self.profiles.get(Role::Destination, name).is_some();
match (s, d) {
(true, true) => {
return Err(format!(
"`{name}` is both a source and a destination profile; say which with `profile('{name}', role='source')` or `role='destination'`"
));
}
(true, false) => Role::Source,
_ => Role::Destination,
}
}
};
self.view(role, name)
}
}
fn check_sheet_name(n: &str) -> Result<(), String> {
if n.is_empty() || n.chars().count() > 31 {
return Err(format!("sheet name `{n}` must be 1 to 31 characters long"));
}
if let Some(c) = n.chars().find(|c| "[]:*?/\\".contains(*c)) {
return Err(format!(
"sheet name `{n}` contains `{c}`, which Excel doesn't allow in sheet names"
));
}
if n.starts_with('\'') || n.ends_with('\'') {
return Err(format!("sheet name `{n}` can't start or end with an apostrophe"));
}
if n.eq_ignore_ascii_case("history") {
return Err("`History` is reserved by Excel and can't be a sheet name".into());
}
Ok(())
}
fn is_single_table(format: &str) -> bool {
matches!(format, "csv" | "delimited" | "fixed_width" | "parquet")
}
fn extension(format: &str) -> &str {
match format {
"delimited" | "fixed_width" => "txt",
f => f,
}
}
fn split_ext(name: &str) -> (&str, &str) {
match name.rfind('.') {
Some(i) if i > 0 => (&name[..i], &name[i..]),
_ => (name, ""),
}
}
fn rel(root: &Path, p: &Path) -> PathBuf {
crate::slash(p.strip_prefix(root).unwrap_or(p))
}
fn record_path(project: &Project, p: &Path) -> PathBuf {
match p.strip_prefix(&project.root) {
Ok(r) => crate::slash(r),
Err(_) => rel(&project.target_dir, p),
}
}
impl BindingRun<'_> {
fn summary_line(&self) -> String {
let rows: u64 = self.produced.iter().map(|p| p.rows).sum();
let mut parts = Vec::new();
if !self.produced.is_empty() {
let n = self.produced.len();
parts.push(format!(
"{n} result set{}, {} row{}",
if n == 1 { "" } else { "s" },
thousands(rows),
if rows == 1 { "" } else { "s" }
));
}
let names: Vec<String> = self
.files
.iter()
.map(|(f, _)| f.file_name().unwrap_or_default().to_string_lossy().to_string())
.collect();
if !names.is_empty() {
parts.push(format!("→ {}", names.join(", ")));
}
let delivered: Vec<&str> = self
.deliveries
.iter()
.filter_map(|d| d.get("location").and_then(Json::as_str))
.collect();
if !delivered.is_empty() {
parts.push(format!("→ {}", delivered.join(", ")));
} else if let Some(n) = &self.delivery_note
&& self.opts.preview.is_none()
&& !self.b.output.destinations.is_empty()
{
parts.push(format!("({n})"));
}
parts.join(" ")
}
}
pub fn thousands(n: u64) -> String {
let s = n.to_string();
let mut out = String::new();
for (i, c) in s.chars().enumerate() {
if i > 0 && (s.len() - i).is_multiple_of(3) {
out.push(',');
}
out.push(c);
}
out
}
fn human_bytes(n: u64) -> String {
match n {
n if n < 1024 => format!("{n} B"),
n if n < 1024 * 1024 => format!("{:.1} KB", n as f64 / 1024.0),
n => format!("{:.1} MB", n as f64 / (1024.0 * 1024.0)),
}
}
pub fn clean(t: &Path) -> std::io::Result<bool> {
if t.exists() {
std::fs::remove_dir_all(t)?;
Ok(true)
} else {
Ok(false)
}
}