rbt/core/
dag.rs

1use super::frontmatter::{
2    BronzeCheckMode, BronzeDiagnostic, BronzeValidationReport, DiagnosticSeverity,
3    StagingFrontmatter,
4};
5use super::parser::{DependencyRef, SqlModelParser};
6use anyhow::{bail, Result};
7use petgraph::algo::{is_cyclic_directed, toposort};
8use petgraph::graph::{DiGraph, NodeIndex};
9use serde::{Deserialize, Serialize};
10use std::collections::{HashMap, HashSet};
11use std::path::Path;
12
13/// Model materialization strategy in Apache Iceberg & data lakes.
14#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
15pub enum Materialization {
16    View,
17    Table,
18    IncrementalAppend,
19    IncrementalMerge,
20    /// Native zero-copy metadata table clone (zero disk byte duplication)
21    ZeroCopyClone,
22}
23
24/// Configurable model output format supported by the engine.
25#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
26pub enum OutputFormat {
27    /// Output directly as Parquet file(s)
28    #[default]
29    Parquet,
30    /// Output as JSONL (jshift zero-copy compatible)
31    Jsonl,
32    /// Output as CSV
33    Csv,
34    /// Output directly into Apache Iceberg catalog table
35    Iceberg,
36    /// Dual-write output: local Parquet files AND Apache Iceberg catalog registration
37    ParquetAndIceberg,
38    /// Native zero-copy metadata pointer clone
39    ZeroCopyClone,
40}
41
42/// DAG Layer classification following project architecture conventions.
43#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
44pub enum ModelLayer {
45    Staging,
46    Transform,
47    Mart,
48}
49
50impl ModelLayer {
51    pub fn from_name(name: &str) -> Self {
52        if name.starts_with("stg_") {
53            Self::Staging
54        } else if name.starts_with("tf_") || name.starts_with("int_") {
55            Self::Transform
56        } else if name.starts_with("dim_")
57            || name.starts_with("fact_")
58            || name.starts_with("obt_")
59            || name.starts_with("fct_")
60        {
61            Self::Mart
62        } else {
63            Self::Transform
64        }
65    }
66}
67
68/// Pipeline Model Node definition.
69#[derive(Debug, Clone, Serialize, Deserialize)]
70pub struct ModelNode {
71    pub name: String,
72    pub description: Option<String>,
73    pub raw_sql: String,
74    pub compiled_sql: String,
75    pub materialization: Materialization,
76    pub output_format: OutputFormat,
77    pub output_path: Option<String>,
78    pub dependencies: Vec<DependencyRef>,
79    pub layer: ModelLayer,
80    pub frontmatter: Option<StagingFrontmatter>,
81}
82
83/// Complete pipeline DAG with topological execution tiering and cycle detection.
84#[derive(Debug, Default)]
85pub struct ModelDag {
86    pub graph: DiGraph<ModelNode, ()>,
87    pub node_map: HashMap<String, NodeIndex>,
88}
89
90impl ModelDag {
91    pub fn new() -> Self {
92        Self::default()
93    }
94
95    /// Registers a raw model SQL definition into the DAG with a default Parquet format.
96    pub fn add_model(
97        &mut self,
98        name: impl Into<String>,
99        raw_sql: &str,
100        materialization: Materialization,
101        catalog_prefix: &str,
102    ) -> Result<NodeIndex> {
103        self.add_model_with_format(
104            name,
105            raw_sql,
106            materialization,
107            OutputFormat::Parquet,
108            None,
109            catalog_prefix,
110        )
111    }
112
113    /// Registers a raw model SQL definition with explicit `OutputFormat` and optional output path.
114    pub fn add_model_with_format(
115        &mut self,
116        name: impl Into<String>,
117        raw_sql: &str,
118        materialization: Materialization,
119        output_format: OutputFormat,
120        output_path: Option<String>,
121        catalog_prefix: &str,
122    ) -> Result<NodeIndex> {
123        let name = name.into();
124        let (frontmatter, pure_sql) = SqlModelParser::parse_frontmatter(raw_sql)
125            .map_err(|e| anyhow::anyhow!("model '{}': {}", name, e))?;
126        let dependencies = SqlModelParser::extract_dependencies(&pure_sql)?;
127        let compiled_sql = SqlModelParser::compile_sql(&pure_sql, catalog_prefix)?;
128        let layer = ModelLayer::from_name(&name);
129        let description = frontmatter.as_ref().and_then(|f| f.description.clone());
130
131        let node = ModelNode {
132            name: name.clone(),
133            description,
134            raw_sql: raw_sql.to_string(),
135            compiled_sql,
136            materialization,
137            output_format,
138            output_path,
139            dependencies,
140            layer,
141            frontmatter,
142        };
143
144        let idx = self.graph.add_node(node);
145        self.node_map.insert(name, idx);
146        Ok(idx)
147    }
148
149    /// Resolves dependency edges between models and validates that there are no circular dependencies or layer boundary violations.
150    pub fn build_graph(&mut self) -> Result<()> {
151        let name_map = self.node_map.clone();
152        let mut edges = Vec::new();
153
154        for (idx, node) in self.node_map.values().map(|&i| (i, &self.graph[i])) {
155            for dep in &node.dependencies {
156                if let DependencyRef::Model(dep_name) = dep {
157                    if let Some(&dep_idx) = name_map.get(dep_name) {
158                        let dep_node = &self.graph[dep_idx];
159
160                        // Enforce Layer Boundary Rule: Transform models cannot depend on Mart models
161                        if node.layer == ModelLayer::Transform && dep_node.layer == ModelLayer::Mart
162                        {
163                            bail!(
164                                "Illegal DAG Layer Boundary: Transform model '{}' (tf_) cannot depend on Mart model '{}' (dim_/fact_/obt_)",
165                                node.name,
166                                dep_node.name
167                            );
168                        }
169
170                        // Enforce Layer Boundary Rule: Staging models cannot depend on downstream models
171                        if node.layer == ModelLayer::Staging {
172                            bail!(
173                                "Illegal DAG Layer Boundary: Staging model '{}' (stg_) cannot depend on downstream model '{}'",
174                                node.name,
175                                dep_node.name
176                            );
177                        }
178
179                        edges.push((dep_idx, idx));
180                    } else {
181                        bail!(
182                            "Missing model dependency '{}' referenced by model '{}'",
183                            dep_name,
184                            node.name
185                        );
186                    }
187                }
188            }
189        }
190
191        for (from, to) in edges {
192            self.graph.add_edge(from, to, ());
193        }
194
195        if is_cyclic_directed(&self.graph) {
196            bail!("Circular dependency detected in model pipeline DAG!");
197        }
198
199        Ok(())
200    }
201
202    /// Computes a flat topologically sorted execution sequence.
203    pub fn topological_sequence(&self) -> Result<Vec<ModelNode>> {
204        let indices = toposort(&self.graph, None)
205            .map_err(|_| anyhow::anyhow!("Circular dependency found during topological sort"))?;
206
207        Ok(indices.into_iter().map(|i| self.graph[i].clone()).collect())
208    }
209
210    /// Computes parallel execution tiers where all models in a tier can be run concurrently.
211    pub fn execution_tiers(&self) -> Result<Vec<Vec<ModelNode>>> {
212        let _sorted = self.topological_sequence()?;
213        let mut in_degrees: HashMap<NodeIndex, usize> = HashMap::new();
214
215        for idx in self.graph.node_indices() {
216            in_degrees.insert(
217                idx,
218                self.graph
219                    .neighbors_directed(idx, petgraph::Direction::Incoming)
220                    .count(),
221            );
222        }
223
224        let mut current_tier: Vec<NodeIndex> = in_degrees
225            .iter()
226            .filter(|(_, &deg)| deg == 0)
227            .map(|(&idx, _)| idx)
228            .collect();
229
230        let mut tiers = Vec::new();
231        let mut visited = HashSet::new();
232
233        while !current_tier.is_empty() {
234            let tier_nodes: Vec<ModelNode> = current_tier
235                .iter()
236                .map(|&i| self.graph[i].clone())
237                .collect();
238            tiers.push(tier_nodes);
239
240            for &node_idx in &current_tier {
241                visited.insert(node_idx);
242            }
243
244            let mut next_tier = Vec::new();
245            for idx in self.graph.node_indices() {
246                if visited.contains(&idx) {
247                    continue;
248                }
249                let incoming_unvisited = self
250                    .graph
251                    .neighbors_directed(idx, petgraph::Direction::Incoming)
252                    .filter(|n| !visited.contains(n))
253                    .count();
254
255                if incoming_unvisited == 0 {
256                    next_tier.push(idx);
257                }
258            }
259
260            current_tier = next_tier;
261        }
262
263        Ok(tiers)
264    }
265
266    /// First `source('ns','table')` dependency, if any.
267    pub fn primary_source(node: &ModelNode) -> Option<(&str, &str)> {
268        node.dependencies.iter().find_map(|d| match d {
269            DependencyRef::Source {
270                source_name,
271                table_name,
272            } => Some((source_name.as_str(), table_name.as_str())),
273            _ => None,
274        })
275    }
276
277    /// Resolve registration identity for a bronze-backed model.
278    pub fn bronze_source_ident(node: &ModelNode) -> Option<(String, String)> {
279        if let Some(fm) = &node.frontmatter {
280            if let (Some(s), Some(t)) = (&fm.source_name, &fm.source_table) {
281                return Some((s.clone(), t.clone()));
282            }
283            if let Some((s, t)) = Self::primary_source(node) {
284                let source = fm.source_name.clone().unwrap_or_else(|| s.to_string());
285                let table = fm.source_table.clone().unwrap_or_else(|| t.to_string());
286                return Some((source, table));
287            }
288            if fm.has_scan_contract() {
289                // Fall back to model name under schema "bronze"
290                let table = fm.source_table.clone().unwrap_or_else(|| node.name.clone());
291                let source = fm
292                    .source_name
293                    .clone()
294                    .unwrap_or_else(|| "bronze".to_string());
295                return Some((source, table));
296            }
297        }
298        Self::primary_source(node).map(|(s, t)| (s.to_string(), t.to_string()))
299    }
300
301    /// Compile-time bronze frontmatter / scan_path validation.
302    ///
303    /// * `Off` — no filesystem checks
304    /// * `Warn` — missing paths become warnings
305    /// * `Fail` — missing paths become errors (`report.has_errors()`)
306    pub fn validate_bronze_sources(
307        &self,
308        project_dir: &Path,
309        mode: BronzeCheckMode,
310    ) -> Result<BronzeValidationReport> {
311        self.validate_bronze_sources_with_roots(
312            project_dir,
313            mode,
314            &std::collections::HashMap::new(),
315        )
316    }
317
318    /// Compile-time bronze checks with project `roots:` for `$name` path templates.
319    pub fn validate_bronze_sources_with_roots(
320        &self,
321        project_dir: &Path,
322        mode: BronzeCheckMode,
323        roots: &std::collections::HashMap<String, String>,
324    ) -> Result<BronzeValidationReport> {
325        if mode == BronzeCheckMode::Off {
326            return Ok(BronzeValidationReport::default());
327        }
328
329        let severity = match mode {
330            BronzeCheckMode::Fail => DiagnosticSeverity::Error,
331            BronzeCheckMode::Warn => DiagnosticSeverity::Warning,
332            BronzeCheckMode::Off => unreachable!(),
333        };
334
335        let mut report = BronzeValidationReport::default();
336
337        for idx in self.graph.node_indices() {
338            let node = &self.graph[idx];
339            let has_source_dep = node
340                .dependencies
341                .iter()
342                .any(|d| matches!(d, DependencyRef::Source { .. }));
343            let fm = node.frontmatter.as_ref();
344
345            // Staging models with source() should declare a scan contract.
346            if has_source_dep && node.layer == ModelLayer::Staging {
347                let missing_scan = fm.map(|f| !f.has_scan_contract()).unwrap_or(true);
348                if missing_scan {
349                    report.diagnostics.push(BronzeDiagnostic {
350                        model: node.name.clone(),
351                        severity,
352                        code: "E_RBT_BRONZE_SCAN_PATH_MISSING",
353                        message:
354                            "staging model references source() but has no frontmatter scan_path; \
355                             add YAML frontmatter with scan_path (and source_format)"
356                                .to_string(),
357                    });
358                    continue;
359                }
360            }
361
362            let Some(fm) = fm else {
363                continue;
364            };
365
366            if !fm.has_scan_contract() {
367                continue;
368            }
369
370            let scan_path = fm.scan_path.as_deref().unwrap();
371
372            // Format resolvable?
373            if let Err(e) = fm.resolve_format() {
374                report.diagnostics.push(BronzeDiagnostic {
375                    model: node.name.clone(),
376                    severity,
377                    code: "E_RBT_BRONZE_FORMAT_UNKNOWN",
378                    message: e.to_string(),
379                });
380            }
381
382            if !super::frontmatter::scan_path_exists_with_roots(project_dir, scan_path, roots) {
383                let resolved = super::paths::resolve_project_path(project_dir, scan_path, roots)
384                    .unwrap_or_else(|_| project_dir.join(scan_path));
385                report.diagnostics.push(BronzeDiagnostic {
386                    model: node.name.clone(),
387                    severity,
388                    code: "E_RBT_BRONZE_SCAN_PATH_NOT_FOUND",
389                    message: format!(
390                        "scan_path '{}' does not exist (resolved: {})",
391                        scan_path,
392                        resolved.display()
393                    ),
394                });
395            }
396
397            // Validate path_glob patterns early (syntax only).
398            if let Some(globs) = fm.path_glob.as_ref() {
399                if let Err(e) = super::paths::validate_glob_patterns(globs) {
400                    report.diagnostics.push(BronzeDiagnostic {
401                        model: node.name.clone(),
402                        severity,
403                        code: "E_RBT_PATH_GLOB_INVALID",
404                        message: e.to_string(),
405                    });
406                }
407            }
408
409            // source() identity vs frontmatter overrides — soft check
410            if let (Some((dep_s, dep_t)), Some(ident)) =
411                (Self::primary_source(node), Self::bronze_source_ident(node))
412            {
413                if let Some(fm_s) = &fm.source_name {
414                    if fm_s != dep_s {
415                        report.diagnostics.push(BronzeDiagnostic {
416                            model: node.name.clone(),
417                            severity: DiagnosticSeverity::Warning,
418                            code: "W_RBT_BRONZE_SOURCE_NAME_MISMATCH",
419                            message: format!(
420                                "frontmatter source_name='{}' differs from source('{}', ...); \
421                                 registration will use '{}'.{} ",
422                                fm_s, dep_s, ident.0, ident.1
423                            ),
424                        });
425                    }
426                }
427                if let Some(fm_t) = &fm.source_table {
428                    if fm_t != dep_t {
429                        report.diagnostics.push(BronzeDiagnostic {
430                            model: node.name.clone(),
431                            severity: DiagnosticSeverity::Warning,
432                            code: "W_RBT_BRONZE_SOURCE_TABLE_MISMATCH",
433                            message: format!(
434                                "frontmatter source_table='{}' differs from source(..., '{}')",
435                                fm_t, dep_t
436                            ),
437                        });
438                    }
439                }
440            }
441        }
442
443        Ok(report)
444    }
445}
446
447#[cfg(test)]
448mod tests {
449    use super::*;
450
451    #[test]
452    fn test_dag_building_and_tiering() -> Result<()> {
453        let mut dag = ModelDag::new();
454
455        // Tier 0: Staging models (no dependencies)
456        dag.add_model_with_format(
457            "stg_users",
458            "SELECT * FROM {{ source('raw', 'users') }}",
459            Materialization::View,
460            OutputFormat::Jsonl,
461            None,
462            "db",
463        )?;
464        dag.add_model_with_format(
465            "stg_orders",
466            "SELECT * FROM {{ source('raw', 'orders') }}",
467            Materialization::View,
468            OutputFormat::ParquetAndIceberg,
469            None,
470            "db",
471        )?;
472
473        // Tier 1: Intermediate model depending on stg_users and stg_orders
474        dag.add_model(
475            "int_user_orders",
476            "SELECT * FROM {{ ref('stg_users') }} u JOIN {{ ref('stg_orders') }} o ON u.id = o.user_id",
477            Materialization::Table,
478            "db",
479        )?;
480
481        // Tier 2: Mart model depending on int_user_orders
482        dag.add_model(
483            "fct_revenue",
484            "SELECT user_id, SUM(amount) FROM {{ ref('int_user_orders') }} GROUP BY user_id",
485            Materialization::IncrementalAppend,
486            "db",
487        )?;
488
489        dag.build_graph()?;
490
491        let tiers = dag.execution_tiers()?;
492        assert_eq!(tiers.len(), 3);
493
494        // Tier 0 has 2 parallel models with formats Jsonl and ParquetAndIceberg
495        assert_eq!(tiers[0].len(), 2);
496        let tier0_formats: Vec<OutputFormat> =
497            tiers[0].iter().map(|m| m.output_format.clone()).collect();
498        assert!(tier0_formats.contains(&OutputFormat::Jsonl));
499        assert!(tier0_formats.contains(&OutputFormat::ParquetAndIceberg));
500
501        Ok(())
502    }
503
504    #[test]
505    fn test_cycle_detection() -> Result<()> {
506        let mut dag = ModelDag::new();
507
508        dag.add_model(
509            "model_a",
510            "SELECT * FROM {{ ref('model_b') }}",
511            Materialization::View,
512            "db",
513        )?;
514        dag.add_model(
515            "model_b",
516            "SELECT * FROM {{ ref('model_a') }}",
517            Materialization::View,
518            "db",
519        )?;
520
521        let res = dag.build_graph();
522        assert!(res.is_err());
523        assert!(res.unwrap_err().to_string().contains("Circular dependency"));
524
525        Ok(())
526    }
527
528    #[test]
529    fn test_layer_boundary_enforcement() -> Result<()> {
530        let mut dag = ModelDag::new();
531
532        // Mart model
533        dag.add_model(
534            "fact_orders",
535            "SELECT 1 AS order_id",
536            Materialization::Table,
537            "db",
538        )?;
539
540        // Illegal transform model attempting to pull from a mart model!
541        dag.add_model(
542            "tf_illegal_transform",
543            "SELECT * FROM {{ ref('fact_orders') }}",
544            Materialization::Table,
545            "db",
546        )?;
547
548        let res = dag.build_graph();
549        assert!(res.is_err());
550        assert!(res
551            .unwrap_err()
552            .to_string()
553            .contains("Illegal DAG Layer Boundary"));
554
555        Ok(())
556    }
557
558    #[test]
559    fn test_bronze_scan_path_validation_warn_and_fail() -> Result<()> {
560        let mut dag = ModelDag::new();
561        dag.add_model_with_format(
562            "stg_events",
563            r#"---
564source_format: jsonl
565scan_path: "lake/bronze/missing.jsonl"
566---
567SELECT * FROM {{ source('bronze', 'events') }}
568"#,
569            Materialization::Table,
570            OutputFormat::Parquet,
571            None,
572            "",
573        )?;
574        dag.build_graph()?;
575
576        let project = Path::new("/tmp/rbt_nonexistent_project_root");
577        let warn = dag.validate_bronze_sources(project, BronzeCheckMode::Warn)?;
578        assert_eq!(warn.error_count(), 0);
579        assert!(warn.warning_count() >= 1);
580        assert!(warn
581            .diagnostics
582            .iter()
583            .any(|d| d.code == "E_RBT_BRONZE_SCAN_PATH_NOT_FOUND"));
584
585        let fail = dag.validate_bronze_sources(project, BronzeCheckMode::Fail)?;
586        assert!(fail.has_errors());
587        Ok(())
588    }
589}
rbt/core/dag.rs

rbt/core/
dag.rs
