rbt/core/
frontmatter.rs

1//! Staging SQL frontmatter: bronze scan contract and compile-time path checks.
2
3use anyhow::{bail, Context, Result};
4use serde::{Deserialize, Serialize};
5use std::fmt;
6use std::path::{Path, PathBuf};
7
8/// How `rbt compile` treats missing/unresolvable bronze `scan_path` entries.
9#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
10#[serde(rename_all = "snake_case")]
11pub enum BronzeCheckMode {
12    /// Skip filesystem checks (DAG structure only).
13    Off,
14    /// Emit warnings; compile still succeeds (default for `compile`).
15    #[default]
16    Warn,
17    /// Missing or invalid bronze sources fail compile.
18    Fail,
19}
20
21impl BronzeCheckMode {
22    pub fn as_str(self) -> &'static str {
23        match self {
24            Self::Off => "off",
25            Self::Warn => "warn",
26            Self::Fail => "fail",
27        }
28    }
29}
30
31impl fmt::Display for BronzeCheckMode {
32    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
33        f.write_str(self.as_str())
34    }
35}
36
37/// Supported bronze file formats for staging lake scans.
38#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
39#[serde(rename_all = "snake_case")]
40pub enum SourceFormat {
41    /// Newline-delimited JSON (also accepts alias `ndjson`).
42    #[serde(alias = "ndjson")]
43    Jsonl,
44    /// Single JSON document or JSON array of objects.
45    Json,
46    Parquet,
47    Csv,
48    /// Arrow IPC file (random-access footer).
49    #[serde(alias = "arrow", alias = "arrow_file", alias = "ipc")]
50    ArrowIpc,
51    /// Arrow IPC stream (append-friendly / WAL-style).
52    #[serde(alias = "arrow_stream", alias = "ipc_stream")]
53    ArrowIpcStream,
54    /// Line-oriented application / server logs.
55    Log,
56    /// Line-oriented text (llms.txt, docs dumps, structured line files).
57    Txt,
58    /// TOML tables / array-of-tables as rows.
59    Toml,
60    /// Length-delimited or whole-file protobuf blobs (opaque bronze).
61    ///
62    /// Each file becomes one row: `_source_path` (Utf8) + `payload` (Binary).
63    /// Typed decode of domain messages is a later step (Rust models / schema registry).
64    #[serde(alias = "pb", alias = "proto")]
65    Protobuf,
66}
67
68impl SourceFormat {
69    pub fn as_str(self) -> &'static str {
70        match self {
71            Self::Jsonl => "jsonl",
72            Self::Json => "json",
73            Self::Parquet => "parquet",
74            Self::Csv => "csv",
75            Self::ArrowIpc => "arrow_ipc",
76            Self::ArrowIpcStream => "arrow_ipc_stream",
77            Self::Log => "log",
78            Self::Txt => "txt",
79            Self::Toml => "toml",
80            Self::Protobuf => "protobuf",
81        }
82    }
83
84    /// Prefer DataFusion listing / external table registration when true.
85    pub fn prefers_datafusion_listing(self) -> bool {
86        matches!(self, Self::Parquet | Self::Csv | Self::Json | Self::Jsonl)
87    }
88
89    /// Infer format from a file extension (without the dot).
90    pub fn from_extension(ext: &str) -> Option<Self> {
91        match ext.to_ascii_lowercase().as_str() {
92            "jsonl" | "ndjson" => Some(Self::Jsonl),
93            "json" => Some(Self::Json),
94            "parquet" | "pq" => Some(Self::Parquet),
95            "csv" | "tsv" => Some(Self::Csv),
96            "arrow" | "arrows" | "ipc" | "feather" => Some(Self::ArrowIpc),
97            "arrows_stream" | "ipc_stream" => Some(Self::ArrowIpcStream),
98            "log" => Some(Self::Log),
99            "txt" | "text" | "md" => Some(Self::Txt),
100            "toml" => Some(Self::Toml),
101            "pb" | "protobuf" | "protobin" => Some(Self::Protobuf),
102            _ => None,
103        }
104    }
105
106    /// Parse free-form frontmatter / CLI format strings.
107    pub fn parse(s: &str) -> Result<Self> {
108        let key = s.trim().to_ascii_lowercase().replace('-', "_");
109        match key.as_str() {
110            "jsonl" | "ndjson" | "json_lines" => Ok(Self::Jsonl),
111            "json" => Ok(Self::Json),
112            "parquet" | "pq" => Ok(Self::Parquet),
113            "csv" | "tsv" => Ok(Self::Csv),
114            "arrow_ipc" | "arrow" | "arrow_file" | "ipc" | "feather" => Ok(Self::ArrowIpc),
115            "arrow_ipc_stream" | "arrow_stream" | "ipc_stream" => Ok(Self::ArrowIpcStream),
116            "log" => Ok(Self::Log),
117            "txt" | "text" => Ok(Self::Txt),
118            "toml" => Ok(Self::Toml),
119            "protobuf" | "pb" | "proto" | "protobin" => Ok(Self::Protobuf),
120            other => bail!(
121                "Unknown source_format '{}'. Expected one of: jsonl, json, parquet, csv, arrow_ipc, arrow_ipc_stream, log, txt, toml, protobuf",
122                other
123            ),
124        }
125    }
126}
127
128impl fmt::Display for SourceFormat {
129    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
130        f.write_str(self.as_str())
131    }
132}
133
134/// Declared data-quality tests for a model (run after materialization when present).
135#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
136pub struct ModelTests {
137    /// Columns that must have zero nulls.
138    #[serde(default)]
139    pub not_null: Option<Vec<String>>,
140    /// Single column unique, or multi-column composite unique when len > 1.
141    #[serde(default)]
142    pub unique: Option<Vec<String>>,
143    /// Map of column → allowed string values.
144    #[serde(default)]
145    pub accepted_values: Option<std::collections::HashMap<String, Vec<String>>>,
146    /// When true (default), failed tests abort `rbt run` for that model.
147    #[serde(default)]
148    pub fail_on_error: Option<bool>,
149}
150
151impl ModelTests {
152    pub fn is_empty(&self) -> bool {
153        self.not_null.as_ref().map(|v| v.is_empty()).unwrap_or(true)
154            && self.unique.as_ref().map(|v| v.is_empty()).unwrap_or(true)
155            && self
156                .accepted_values
157                .as_ref()
158                .map(|m| m.is_empty())
159                .unwrap_or(true)
160    }
161
162    pub fn should_fail_on_error(&self) -> bool {
163        self.fail_on_error.unwrap_or(true)
164    }
165}
166
167/// Per-column documentation for humans and AI agents.
168///
169/// * `description` — short label (1–2 lines)
170/// * `context` — longer intent, units, lineage, caveats (agent-oriented)
171#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
172pub struct ColumnMeta {
173    #[serde(default)]
174    pub description: Option<String>,
175    #[serde(default)]
176    pub context: Option<String>,
177    /// Optional logical type hint (`utf8`, `int64`, `float64`, `timestamp`, …).
178    #[serde(default)]
179    pub dtype: Option<String>,
180    /// Optional unit (`USD`, `shares`, `ratio`, `ns_epoch`, …).
181    #[serde(default)]
182    pub unit: Option<String>,
183}
184
185/// YAML frontmatter embedded in model SQL files (`---` … `---`).
186///
187/// Used on staging, transforms, and marts. Scan-related fields only apply when
188/// a bronze scan contract (`scan_path`) is present.
189#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
190pub struct StagingFrontmatter {
191    /// Human-readable model purpose (docs + future catalog).
192    #[serde(default)]
193    pub description: Option<String>,
194    /// Longer model-level context for AI agents (why it exists, consumers, caveats).
195    #[serde(default)]
196    pub context: Option<String>,
197    /// Column-level description + context map (name → meta).
198    #[serde(default)]
199    pub columns: Option<std::collections::BTreeMap<String, ColumnMeta>>,
200
201    /// Explicit bronze format. If omitted, inferred from `scan_path` extension.
202    #[serde(default)]
203    pub source_format: Option<SourceFormat>,
204    /// File or directory to scan (project-relative, absolute, or `$root/...` template).
205    #[serde(default)]
206    pub scan_path: Option<String>,
207    /// Filename / relative-path glob(s) under `scan_path` (OR semantics).
208    ///
209    /// Examples: `crawlplan.parquet`, `**/raw_snoop/crawlplan.parquet`, `*.jsonl`.
210    /// Empty / omitted = all files matching `source_format`.
211    /// Accepts a single string or a YAML list.
212    ///
213    /// **Pushdown note:** any non-empty `path_glob` forces the scan→MemTable bronze path
214    /// (DataFusion directory listing / predicate pushdown is **not** used for that source),
215    /// because listing providers cannot apply rbt's filename globs or hive path injection.
216    #[serde(default, deserialize_with = "deserialize_string_or_vec")]
217    pub path_glob: Option<Vec<String>>,
218    /// Optional hive-style partition keys (path injection + future pruning).
219    #[serde(default)]
220    pub partition_by: Option<Vec<String>>,
221    /// jshift / projection field paths for selective JSON(L) extract.
222    #[serde(default)]
223    pub paths: Option<Vec<String>>,
224    /// Override catalog/schema name for registration (default: first `source()` name).
225    #[serde(default)]
226    pub source_name: Option<String>,
227    /// Override table name for registration (default: first `source()` table).
228    #[serde(default)]
229    pub source_table: Option<String>,
230    /// TOML: key of the array-of-tables to expand into rows (default: auto-detect).
231    #[serde(default)]
232    pub toml_rows_key: Option<String>,
233    /// When true, use scan→MemTable path even for formats that support DF listing.
234    #[serde(default)]
235    pub force_scan: Option<bool>,
236    /// Only scan hive-partitioned files whose path segments match these values.
237    /// Example: `{ timeframe: "1m" }` keeps `.../timeframe=1m/...` and skips `timeframe=1d`.
238    #[serde(default)]
239    pub require_partitions: Option<std::collections::HashMap<String, String>>,
240    /// Inject `_source_path` (Utf8) with the absolute file path for each row.
241    /// Enables "latest chunk wins" dedupe via `ORDER BY _source_path DESC`.
242    #[serde(default)]
243    pub inject_source_path: Option<bool>,
244
245    /// Logical grain of the model (e.g. `[symbol, timestamp_ns]`).
246    #[serde(default)]
247    pub grain: Option<Vec<String>>,
248    /// Primary uniqueness contract (usually same as grain for staging facts).
249    #[serde(default)]
250    pub unique_key: Option<Vec<String>>,
251    /// Free-form tags for selection / docs.
252    #[serde(default)]
253    pub tags: Option<Vec<String>>,
254    /// Materialization hint: `table` | `view` | `incremental_append` (engine may ignore for now).
255    #[serde(default)]
256    pub materialization: Option<String>,
257    /// Post-materialization assertions.
258    #[serde(default)]
259    pub tests: Option<ModelTests>,
260    /// Opaque metadata for tools and agents (strings, lists, nested maps OK).
261    #[serde(default)]
262    pub meta: Option<std::collections::BTreeMap<String, serde_yaml::Value>>,
263}
264
265impl StagingFrontmatter {
266    /// Resolve format from explicit field or path extension.
267    pub fn resolve_format(&self) -> Result<SourceFormat> {
268        if let Some(fmt) = self.source_format {
269            return Ok(fmt);
270        }
271        let path = self
272            .scan_path
273            .as_deref()
274            .context("frontmatter missing both source_format and scan_path")?;
275        // Strip globs for extension sniffing: `foo/*.jsonl` → look at last segment
276        let candidate = path.rsplit('/').next().unwrap_or(path);
277        let candidate = candidate.trim_matches(|c| c == '*' || c == '?');
278        if let Some(ext) = Path::new(candidate).extension().and_then(|e| e.to_str()) {
279            if let Some(fmt) = SourceFormat::from_extension(ext) {
280                return Ok(fmt);
281            }
282        }
283        // Directory paths: no extension — require explicit format
284        bail!(
285            "Cannot infer source_format from scan_path '{}'; set source_format explicitly",
286            path
287        );
288    }
289
290    pub fn has_scan_contract(&self) -> bool {
291        self.scan_path
292            .as_ref()
293            .map(|s| !s.trim().is_empty())
294            .unwrap_or(false)
295    }
296}
297
298/// Severity of a bronze compile diagnostic.
299#[derive(Debug, Clone, Copy, PartialEq, Eq)]
300pub enum DiagnosticSeverity {
301    Warning,
302    Error,
303}
304
305/// One compile-time finding about bronze frontmatter / scan paths.
306#[derive(Debug, Clone, PartialEq, Eq)]
307pub struct BronzeDiagnostic {
308    pub model: String,
309    pub severity: DiagnosticSeverity,
310    pub code: &'static str,
311    pub message: String,
312}
313
314impl fmt::Display for BronzeDiagnostic {
315    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
316        let level = match self.severity {
317            DiagnosticSeverity::Warning => "warning",
318            DiagnosticSeverity::Error => "error",
319        };
320        write!(
321            f,
322            "{}[{}] model={}: {}",
323            level, self.code, self.model, self.message
324        )
325    }
326}
327
328/// Result of bronze path validation during compile.
329#[derive(Debug, Clone, Default)]
330pub struct BronzeValidationReport {
331    pub diagnostics: Vec<BronzeDiagnostic>,
332}
333
334impl BronzeValidationReport {
335    pub fn warning_count(&self) -> usize {
336        self.diagnostics
337            .iter()
338            .filter(|d| d.severity == DiagnosticSeverity::Warning)
339            .count()
340    }
341
342    pub fn error_count(&self) -> usize {
343        self.diagnostics
344            .iter()
345            .filter(|d| d.severity == DiagnosticSeverity::Error)
346            .count()
347    }
348
349    pub fn has_errors(&self) -> bool {
350        self.error_count() > 0
351    }
352}
353
354/// Resolve `scan_path` against the project root (no named roots). Prefer
355/// [`crate::core::paths::resolve_project_path`] when `roots:` are in play.
356pub fn resolve_scan_path(project_dir: &Path, scan_path: &str) -> PathBuf {
357    crate::core::paths::resolve_project_path(project_dir, scan_path, &Default::default())
358        .unwrap_or_else(|_| project_dir.to_path_buf())
359}
360
361pub use crate::core::paths::is_remote_uri;
362
363/// Deserialize either a single string or a sequence into `Option<Vec<String>>`.
364fn deserialize_string_or_vec<'de, D>(
365    deserializer: D,
366) -> std::result::Result<Option<Vec<String>>, D::Error>
367where
368    D: serde::Deserializer<'de>,
369{
370    use serde::de::{self, SeqAccess, Visitor};
371    use std::fmt;
372
373    struct StringOrVec;
374    impl<'de> Visitor<'de> for StringOrVec {
375        type Value = Option<Vec<String>>;
376
377        fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result {
378            f.write_str("a string or list of strings")
379        }
380
381        fn visit_none<E: de::Error>(self) -> std::result::Result<Self::Value, E> {
382            Ok(None)
383        }
384
385        fn visit_unit<E: de::Error>(self) -> std::result::Result<Self::Value, E> {
386            Ok(None)
387        }
388
389        fn visit_str<E: de::Error>(self, v: &str) -> std::result::Result<Self::Value, E> {
390            Ok(Some(vec![v.to_string()]))
391        }
392
393        fn visit_string<E: de::Error>(self, v: String) -> std::result::Result<Self::Value, E> {
394            Ok(Some(vec![v]))
395        }
396
397        fn visit_seq<A: SeqAccess<'de>>(
398            self,
399            mut seq: A,
400        ) -> std::result::Result<Self::Value, A::Error> {
401            let mut out = Vec::new();
402            while let Some(s) = seq.next_element::<String>()? {
403                out.push(s);
404            }
405            Ok(Some(out))
406        }
407    }
408
409    deserializer.deserialize_any(StringOrVec)
410}
411
412/// Whether a resolved local scan path currently exists (file or directory).
413/// Remote URIs are treated as "exists" for compile (runtime / object-store later).
414pub fn scan_path_exists(project_dir: &Path, scan_path: &str) -> bool {
415    scan_path_exists_with_roots(project_dir, scan_path, &std::collections::HashMap::new())
416}
417
418/// Like [`scan_path_exists`] but expands `$root` templates from project config.
419pub fn scan_path_exists_with_roots(
420    project_dir: &Path,
421    scan_path: &str,
422    roots: &std::collections::HashMap<String, String>,
423) -> bool {
424    if is_remote_uri(scan_path.trim()) {
425        return true;
426    }
427    let Ok(resolved) = crate::core::paths::resolve_project_path(project_dir, scan_path, roots)
428    else {
429        return false;
430    };
431    // Support simple trailing globs: `dir/*.jsonl` → check parent dir
432    let check = strip_simple_glob(&resolved);
433    check.exists()
434}
435
436fn strip_simple_glob(path: &Path) -> PathBuf {
437    let s = path.to_string_lossy();
438    if s.contains('*') || s.contains('?') {
439        if let Some(parent) = path.parent() {
440            return parent.to_path_buf();
441        }
442    }
443    path.to_path_buf()
444}
445
446#[cfg(test)]
447mod tests {
448    use super::*;
449
450    #[test]
451    fn format_from_extension_and_parse() {
452        assert_eq!(
453            SourceFormat::from_extension("jsonl"),
454            Some(SourceFormat::Jsonl)
455        );
456        assert_eq!(
457            SourceFormat::from_extension("toml"),
458            Some(SourceFormat::Toml)
459        );
460        assert_eq!(SourceFormat::from_extension("log"), Some(SourceFormat::Log));
461        assert_eq!(
462            SourceFormat::parse("arrow-ipc").unwrap(),
463            SourceFormat::ArrowIpc
464        );
465        assert_eq!(SourceFormat::parse("ndjson").unwrap(), SourceFormat::Jsonl);
466    }
467
468    #[test]
469    fn resolve_format_from_path() {
470        let fm = StagingFrontmatter {
471            scan_path: Some("lake/bronze/raw.jsonl".into()),
472            ..Default::default()
473        };
474        assert_eq!(fm.resolve_format().unwrap(), SourceFormat::Jsonl);
475    }
476
477    #[test]
478    fn remote_uri_exists_for_compile() {
479        assert!(scan_path_exists(
480            Path::new("/tmp"),
481            "s3://bucket/bronze/x.jsonl"
482        ));
483    }
484}
rbt/core/frontmatter.rs

rbt/core/
frontmatter.rs
