refactor: finish moving Python logic to Rust and enforce boundary rules

crates/sqlbuild-analysis/src/assembly/analysis_session/_helpers/cte_facts.rs

@@ -32,6 +32,9 @@ use crate::assembly::analysis_session::constants::{
 use crate::assembly::analysis_session::models::NullabilityCallback;
 use crate::assembly::analysis_session::models::{ColumnFact, LineageRow};
 use crate::assembly::analysis_session::types::{Pairs, Shapes};
+use crate::nullability_rules::main::conditional_result_nullability::conditional_result_nullability;
+use crate::nullability_rules::main::first_arg_nullability::first_arg_nullability;
+use crate::nullability_rules::main::safe_cast_nullability::safe_cast_nullability;
 use crate::type_system::main::normalize_type::normalize_type;
 use crate::type_system::models::{NormalizedType, TypeFamily};
 
@@ -1020,36 +1023,13 @@ fn combined(node_kind: &str, rule: Option<&str>, arguments: &[&'static str]) ->
         return UNKNOWN_NULLABILITY;
     }
     match rule {
-        Some(FIRST_ARG_RULE) => arguments.first().copied().unwrap_or(UNKNOWN_NULLABILITY),
-        Some(CONDITIONAL_RESULT_RULE) => conditional_result(arguments),
-        Some(SAFE_CAST_RULE) => safe_cast(arguments),
+        Some(FIRST_ARG_RULE) => first_arg_nullability(arguments),
+        Some(CONDITIONAL_RESULT_RULE) => conditional_result_nullability(arguments),
+        Some(SAFE_CAST_RULE) => safe_cast_nullability(arguments),
         _ => UNKNOWN_NULLABILITY,
     }
 }
 
-/// Python's `safe_cast_nullability`.
-fn safe_cast(arguments: &[&'static str]) -> &'static str {
-    if arguments.first() == Some(&NULLABLE_NULLABILITY) {
-        NULLABLE_NULLABILITY
-    } else {
-        UNKNOWN_NULLABILITY
-    }
-}
-
-/// Python's `conditional_result_nullability`.
-fn conditional_result(arguments: &[&'static str]) -> &'static str {
-    let [_, when_true, when_false, ..] = arguments else {
-        return UNKNOWN_NULLABILITY;
-    };
-    if *when_true == NON_NULL_NULLABILITY && *when_false == NON_NULL_NULLABILITY {
-        return NON_NULL_NULLABILITY;
-    }
-    if *when_true == NULLABLE_NULLABILITY || *when_false == NULLABLE_NULLABILITY {
-        return NULLABLE_NULLABILITY;
-    }
-    UNKNOWN_NULLABILITY
-}
-
 /// Python's `_infer_polyglot_column_nullability`.
 fn column_nullability(
     column: &Expression,

crates/sqlbuild-analysis/src/assembly/analysis_session/_helpers/dynamic_pivot.rs

@@ -13,15 +13,15 @@ use crate::assembly::analysis_session::_helpers::dict_walk::{
     parse_one_or_error, payload, relation_name, render_type, to_dict, truthy, upper,
 };
 use crate::assembly::analysis_session::constants::{
-    ALIAS_AST_KIND, CAST_AST_KIND, CTE_AST_KIND, DUCKDB_DIALECT, DYNAMIC_VALUE_SOURCE_KINDS,
-    MOTHERDUCK_DIALECT, PIVOT_AST_KIND, QUERY_AST_KINDS, SELECT_AST_KIND,
-    SIMPLIFIED_PIVOT_DIALECTS, SUPPORTED_PIVOT_DIALECTS, TYPE_PRESERVING_AGGREGATES,
-    UNKNOWN_NULLABILITY, UNKNOWN_TYPE,
+    ALIAS_AST_KIND, CAST_AST_KIND, DUCKDB_DIALECT, DYNAMIC_VALUE_SOURCE_KINDS, MOTHERDUCK_DIALECT,
+    PIVOT_AST_KIND, QUERY_AST_KINDS, SELECT_AST_KIND, SIMPLIFIED_PIVOT_DIALECTS,
+    SUPPORTED_PIVOT_DIALECTS, TYPE_PRESERVING_AGGREGATES, UNKNOWN_NULLABILITY, UNKNOWN_TYPE,
 };
 use crate::assembly::analysis_session::models::{
     ColumnFact, ContractProof, DynamicFamily, PivotModel, PivotOutcome, PivotTables,
 };
 use crate::assembly::analysis_session::types::Shapes;
+use sqlbuild_core::constants::CTE_AST_KIND;
 
 /// What an internal native failure of a pivot proof names.
 const PIVOT_CONTEXT: &str = "native dynamic pivot proof";

crates/sqlbuild-analysis/src/assembly/analysis_session/constants.rs

@@ -72,7 +72,6 @@ pub(crate) const COLUMN_AST_KIND: &str = "column";
 pub(crate) const TABLE_AST_KIND: &str = "table";
 pub(crate) const FUNCTION_AST_KIND: &str = "function";
 pub(crate) const ALIAS_AST_KIND: &str = "alias";
-pub(crate) const CTE_AST_KIND: &str = "cte";
 pub(crate) const STAR_AST_KIND: &str = "star";
 pub(crate) const DYNAMIC_VALUE_SOURCE_KINDS: [&str; 3] = ["pivot_any", "select", "query"];
 pub(crate) const DEPENDENCY_FUNCTION_NAMES: [&str; 3] = ["__ref", "__source", "__seed"];

crates/sqlbuild-analysis/src/contracts/_helpers/promotion.rs

@@ -5,10 +5,12 @@ use crate::contracts::_helpers::setting_help::{
 };
 use crate::contracts::constants::{
     CONTRACT_ENFORCED, IMMEDIATE_PROMOTION_MODE, MICROBATCH_INCREMENTAL_MODE,
-    PROMOTION_CONFLICT_CODE, PROMOTION_REMOVE_CONTRACT_HELP, SETTINGS_SECTION,
-    STAGED_LIFECYCLE_MATERIALIZATIONS, STAGED_PROMOTION_MODE, TABLE_PROMOTION_MODE_SETTING_KEY,
+    PROMOTION_CONFLICT_CODE, PROMOTION_REMOVE_CONTRACT_HELP, STAGED_LIFECYCLE_MATERIALIZATIONS,
 };
 use crate::contracts::models::{PromotionConflict, PromotionModel, PromotionRequest};
+use sqlbuild_core::constants::SETTINGS_SECTION;
+use sqlbuild_core::constants::STAGED_PROMOTION_MODE;
+use sqlbuild_core::constants::TABLE_PROMOTION_MODE_SETTING_KEY;
 
 /// One conflict per enforced-contract table model when the effective promotion is immediate.
 pub(crate) fn promotion_conflicts(request: &PromotionRequest) -> Vec<PromotionConflict> {

crates/sqlbuild-analysis/src/contracts/constants.rs

@@ -9,14 +9,11 @@ pub(crate) const MISSING_DECLARATIONS_CODE: &str = "K006";
 pub(crate) const DYNAMIC_OUTPUT_NOT_PROVEN_CODE: &str = "K011";
 pub(crate) const PROMOTION_CONFLICT_CODE: &str = "K011";
 
-pub(crate) const CONTRACT_ENFORCED: &str = "enforced";
+pub const CONTRACT_ENFORCED: &str = "enforced";
 pub(crate) const CONTRACT_NONE: &str = "none";
 pub(crate) const STAGED_LIFECYCLE_MATERIALIZATIONS: [&str; 2] = ["table", "incremental"];
 pub(crate) const MICROBATCH_INCREMENTAL_MODE: &str = "microbatch";
 pub(crate) const IMMEDIATE_PROMOTION_MODE: &str = "immediate";
-pub(crate) const STAGED_PROMOTION_MODE: &str = "staged";
-pub(crate) const SETTINGS_SECTION: &str = "settings";
-pub(crate) const TABLE_PROMOTION_MODE_SETTING_KEY: &str = "table_promotion_mode";
 
 pub(crate) const NO_COMPILER_EVIDENCE: &str = "no compiler evidence";
 pub(crate) const DYNAMIC_OUTPUT_NOT_PROVEN_HELP: &str = "project one supported runtime-dynamic pivot through a wildcard output boundary and declare every generated family";

crates/sqlbuild-analysis/src/contracts/mod.rs

@@ -1,7 +1,7 @@
 //! Model column contracts and promotion conflicts.
 
 pub(crate) mod _helpers;
-pub(crate) mod constants;
+pub mod constants;
 pub mod main;
 pub mod models;
 

crates/sqlbuild-analysis/src/dag_artifact/_helpers/mod.rs

@@ -1,5 +1,6 @@
 pub(crate) mod checks;
 pub(crate) mod edges;
 pub(crate) mod nodes;
+pub(crate) mod projection;
 pub(crate) mod selection;
 pub(crate) mod values;

crates/sqlbuild-analysis/src/dag_artifact/_helpers/projection.rs

@@ -0,0 +1,27 @@
+//! Node lookups the orchestrator projections share.
+
+use std::collections::HashMap;
+
+use crate::dag_artifact::models::DagProjectionGraph;
+
+/// Node index by id; a later node with a repeated id replaces the earlier one.
+pub(crate) fn nodes_by_id(graph: &DagProjectionGraph) -> HashMap<&str, usize> {
+    graph
+        .nodes
+        .iter()
+        .enumerate()
+        .map(|(index, node)| (node.id.as_str(), index))
+        .collect()
+}
+
+/// Upstream node ids by downstream node id, in edge order.
+pub(crate) fn upstream_by_id(graph: &DagProjectionGraph) -> HashMap<&str, Vec<&str>> {
+    let mut upstream: HashMap<&str, Vec<&str>> = HashMap::new();
+    for (from_id, to_id) in &graph.edges {
+        upstream
+            .entry(to_id.as_str())
+            .or_default()
+            .push(from_id.as_str());
+    }
+    upstream
+}

crates/sqlbuild-analysis/src/dag_artifact/constants.rs

@@ -34,3 +34,5 @@ pub(crate) const SELECTOR_KIND_CHECK: &str = "check";
 pub(crate) const SELECTOR_KIND_TEST: &str = "test";
 pub(crate) const SELECTOR_KIND_SEPARATOR: &str = ":";
 pub(crate) const SELECTOR_MODELS_PREFIX: &str = "models/";
+/// The orchestrator kind every SQLBuild asset carries.
+pub(crate) const SQLBUILD_ASSET_KIND: &str = "sqlbuild";

crates/sqlbuild-analysis/src/dag_artifact/main/dag_asset_plans.rs

@@ -0,0 +1,37 @@
+//! The asset nodes an orchestrator definition owns and their dependencies.
+
+use std::collections::HashMap;
+
+use crate::dag_artifact::_helpers::projection::{nodes_by_id, upstream_by_id};
+use crate::dag_artifact::models::{AssetNodePlan, AssetProjectionPolicy, DagProjectionGraph};
+
+/// Each node of an asset kind, in artifact order, with its upstream nodes in edge order.
+#[must_use]
+pub fn dag_asset_plans(
+    graph: &DagProjectionGraph,
+    policy: &AssetProjectionPolicy,
+) -> Vec<AssetNodePlan> {
+    let nodes: HashMap<&str, usize> = nodes_by_id(graph);
+    let upstream: HashMap<&str, Vec<&str>> = upstream_by_id(graph);
+    let is_asset = |index: usize| policy.asset_kinds.contains(&graph.nodes[index].kind);
+    let mut plans: Vec<AssetNodePlan> = Vec::new();
+    for (index, node) in graph.nodes.iter().enumerate() {
+        if !is_asset(index) {
+            continue;
+        }
+        let mut dependencies: Vec<usize> = Vec::new();
+        for upstream_id in upstream.get(node.id.as_str()).into_iter().flatten() {
+            match nodes.get(upstream_id) {
+                Some(dependency) if !policy.upstream_asset_kinds_only || is_asset(*dependency) => {
+                    dependencies.push(*dependency);
+                }
+                _ => {}
+            }
+        }
+        plans.push(AssetNodePlan {
+            node: index,
+            upstream: dependencies,
+        });
+    }
+    plans
+}

crates/sqlbuild-analysis/src/dag_artifact/main/dag_asset_spec_kinds.rs

@@ -0,0 +1,27 @@
+//! The orchestrator kinds of one asset.
+
+use crate::dag_artifact::constants::SQLBUILD_ASSET_KIND;
+use crate::dag_artifact::models::AssetSpecKindPolicy;
+
+/// `sqlbuild` plus `view`/`table` for models, the node kind for direct kinds, else nothing more.
+#[must_use]
+pub fn dag_asset_spec_kinds(
+    (kind, materialization_type): (&str, &str),
+    policy: &AssetSpecKindPolicy,
+) -> Vec<String> {
+    let detail: Option<&str> = if kind == policy.model_kind {
+        Some(if materialization_type == policy.view_materialization {
+            "view"
+        } else {
+            "table"
+        })
+    } else if policy.direct_kinds.contains(kind) {
+        Some(kind)
+    } else {
+        None
+    };
+    std::iter::once(SQLBUILD_ASSET_KIND)
+        .chain(detail)
+        .map(str::to_owned)
+        .collect()
+}

crates/sqlbuild-analysis/src/dag_artifact/main/dag_check_plans.rs

@@ -0,0 +1,42 @@
+//! The checks an orchestrator check definition exposes and the nodes they check.
+
+use std::collections::HashMap;
+
+use crate::dag_artifact::_helpers::projection::nodes_by_id;
+use crate::dag_artifact::models::{
+    CheckKindFilter, CheckPlan, DagProjectionCheck, DagProjectionGraph,
+};
+
+/// Whether `filter` exposes checks of `kind`.
+fn exposed(filter: &CheckKindFilter, kind: &str) -> bool {
+    !filter.excluded.contains(kind)
+        && filter
+            .required
+            .as_ref()
+            .is_none_or(|required| required.contains(kind))
+}
+
+/// The indexes of the known nodes `check` names, in its order.
+fn checked_nodes(nodes: &HashMap<&str, usize>, check: &DagProjectionCheck) -> Vec<usize> {
+    check
+        .checked_asset_ids
+        .iter()
+        .filter_map(|id| nodes.get(id.as_str()).copied())
+        .collect()
+}
+
+/// Every check in artifact order, whether `filter` exposes its kind, and its known checked nodes.
+#[must_use]
+pub fn dag_check_plans(graph: &DagProjectionGraph, filter: &CheckKindFilter) -> Vec<CheckPlan> {
+    let nodes: HashMap<&str, usize> = nodes_by_id(graph);
+    graph
+        .checks
+        .iter()
+        .enumerate()
+        .map(|(index, check)| CheckPlan {
+            check: index,
+            included: exposed(filter, &check.kind),
+            nodes: checked_nodes(&nodes, check),
+        })
+        .collect()
+}

crates/sqlbuild-analysis/src/dag_artifact/main/dag_selected_asset_nodes.rs

@@ -0,0 +1,21 @@
+//! The selected nodes an orchestrator owns as assets.
+
+use std::collections::BTreeSet;
+
+use crate::dag_artifact::models::DagProjectionGraph;
+
+/// Indexes of nodes whose id is selected and whose kind is an asset kind, in artifact order.
+#[must_use]
+pub fn dag_selected_asset_nodes(
+    graph: &DagProjectionGraph,
+    selected_ids: &BTreeSet<String>,
+    asset_kinds: &BTreeSet<String>,
+) -> Vec<usize> {
+    graph
+        .nodes
+        .iter()
+        .enumerate()
+        .filter(|(_, node)| selected_ids.contains(&node.id) && asset_kinds.contains(&node.kind))
+        .map(|(index, _)| index)
+        .collect()
+}

crates/sqlbuild-analysis/src/dag_artifact/main/mod.rs

@@ -1,2 +1,6 @@
 pub mod build_dag_json;
+pub mod dag_asset_plans;
+pub mod dag_asset_spec_kinds;
+pub mod dag_check_plans;
+pub mod dag_selected_asset_nodes;
 pub mod resolve_dag_selection;

crates/sqlbuild-analysis/src/dag_artifact/models.rs

@@ -1,5 +1,7 @@
 //! The compiled project, Python node graph and authored values the DAG artifact is built from.
 
+use std::collections::BTreeSet;
+
 use crate::assembly::project::types::ObjectKey;
 use crate::dag_artifact::types::{DagColumnLineage, DagValue};
 
@@ -243,6 +245,69 @@ pub struct DagSelectionGraph {
     pub checks: Vec<DagSelectionCheck>,
 }
 
+/// One DAG artifact node as orchestrator asset projection reads it.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct DagProjectionNode {
+    pub id: String,
+    pub kind: String,
+}
+
+/// One DAG artifact check as orchestrator check projection reads it.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct DagProjectionCheck {
+    pub kind: String,
+    pub checked_asset_ids: Vec<String>,
+}
+
+/// The DAG artifact's nodes, `(from, to)` edges and checks, in artifact order.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct DagProjectionGraph {
+    pub nodes: Vec<DagProjectionNode>,
+    pub edges: Vec<(String, String)>,
+    pub checks: Vec<DagProjectionCheck>,
+}
+
+/// Which nodes become orchestrator assets and which upstream nodes they depend on.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct AssetProjectionPolicy {
+    /// Node kinds the orchestrator owns as assets.
+    pub asset_kinds: BTreeSet<String>,
+    /// Upstream nodes of other kinds are left out of an asset's dependencies.
+    pub upstream_asset_kinds_only: bool,
+}
+
+/// One asset node and the nodes it depends on, as indexes into the graph's nodes.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct AssetNodePlan {
+    pub node: usize,
+    pub upstream: Vec<usize>,
+}
+
+/// Which check kinds a check definition exposes.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct CheckKindFilter {
+    pub excluded: BTreeSet<String>,
+    /// When set, only these kinds are exposed.
+    pub required: Option<BTreeSet<String>>,
+}
+
+/// One check, whether its kind is exposed, and the nodes it checks as graph indexes.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct CheckPlan {
+    pub check: usize,
+    pub included: bool,
+    pub nodes: Vec<usize>,
+}
+
+/// How an asset's orchestrator kinds follow from its node kind and materialization.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct AssetSpecKindPolicy {
+    pub model_kind: String,
+    pub view_materialization: String,
+    /// Node kinds reported as their own orchestrator kind.
+    pub direct_kinds: BTreeSet<String>,
+}
+
 /// The node lookup and edge indexes every selector token resolves against.
 #[derive(Debug)]
 pub(crate) struct DagSelectionIndex<'a> {

crates/sqlbuild-analysis/src/dag_artifact/tests.rs

@@ -2,5 +2,7 @@
 mod build_dag_json;
 #[path = "tests/helpers.rs"]
 mod helpers;
+#[path = "tests/test_projection.rs"]
+mod projection;
 #[path = "tests/test_types.rs"]
 mod test_types;