refactor: port planner identities, scopes, rules and CLI runtime to Rust

crates/sqlbuild-analysis/src/lib.rs

@@ -20,6 +20,9 @@ pub mod lineage_command;
 pub mod lineage_graph;
 pub mod manifest;
 pub mod planner_changes;
+pub mod planner_identity;
+pub mod planner_scope;
+pub mod planner_selection;
 pub mod planner_strategy;
 pub mod python_plan;
 pub mod python_selection;
@@ -33,4 +36,5 @@ pub mod semantic_usage;
 pub mod semantic_validation;
 pub mod sql_test_attachment;
 pub mod sql_tokens;
+pub mod target_paths;
 pub mod type_system;

crates/sqlbuild-analysis/src/planner_identity/_helpers/compose_graph_identity.rs

@@ -0,0 +1,13 @@
+//! A node identity composed from its local hash and the hashes of its available upstreams.
+
+use crate::planner_identity::main::stable_identity_hash::stable_identity_hash;
+
+/// The hash of the local hash and each upstream hash, one per line.
+pub(crate) fn compose_graph_identity(local_hash: &str, upstream_hashes: &[&str]) -> String {
+    let mut text: String = local_hash.to_owned();
+    for upstream_hash in upstream_hashes {
+        text.push('\n');
+        text.push_str(upstream_hash);
+    }
+    stable_identity_hash(&text)
+}

crates/sqlbuild-analysis/src/planner_identity/_helpers/cycle_message.rs

@@ -0,0 +1,65 @@
+//! The dependency-cycle message naming the keys left unordered and the injected edges between them.
+
+use std::collections::{BTreeSet, HashMap, HashSet};
+
+use crate::planner_identity::models::GraphKey;
+
+/// Upstream edges and the origins of injected edges.
+type CycleEdges<'a> = (
+    &'a [(GraphKey, Vec<GraphKey>)],
+    &'a HashMap<(GraphKey, GraphKey), String>,
+);
+
+/// The message for the keys Kahn's algorithm could not order.
+pub(crate) fn cycle_message(
+    ordered: &[GraphKey],
+    node_keys: HashSet<&GraphKey>,
+    (upstream, injected_edge_origins): CycleEdges<'_>,
+) -> String {
+    let ordered_set: HashSet<&GraphKey> = ordered.iter().collect();
+    let mut cycle_keys: Vec<&GraphKey> = node_keys
+        .into_iter()
+        .filter(|key| !ordered_set.contains(key))
+        .collect();
+    cycle_keys.sort();
+    let names: Vec<String> = cycle_keys
+        .iter()
+        .map(|key| format!("{}:{}", key.kind, key.name))
+        .collect();
+    format!(
+        "Dependency cycle detected involving: {}{}",
+        names.join(", "),
+        cycle_origin_notes(&cycle_keys, upstream, injected_edge_origins)
+    )
+}
+
+fn cycle_origin_notes(
+    cycle_keys: &[&GraphKey],
+    upstream: &[(GraphKey, Vec<GraphKey>)],
+    injected_edge_origins: &HashMap<(GraphKey, GraphKey), String>,
+) -> String {
+    if injected_edge_origins.is_empty() {
+        return String::new();
+    }
+    let cycle_set: HashSet<&GraphKey> = cycle_keys.iter().copied().collect();
+    let deps: HashMap<&GraphKey, &Vec<GraphKey>> =
+        upstream.iter().map(|(key, deps)| (key, deps)).collect();
+    let mut notes: BTreeSet<&str> = BTreeSet::new();
+    for key in cycle_keys {
+        for dep_key in deps.get(key).into_iter().flat_map(|deps| deps.iter()) {
+            if !cycle_set.contains(dep_key) {
+                continue;
+            }
+            if let Some(origin) = injected_edge_origins.get(&((*key).clone(), dep_key.clone())) {
+                notes.insert(origin);
+            }
+        }
+    }
+    if notes.is_empty() {
+        return String::new();
+    }
+    format!(
+        " (via {})",
+        notes.into_iter().collect::<Vec<&str>>().join("; ")
+    )
+}

crates/sqlbuild-analysis/src/planner_identity/_helpers/dependency_counts.rs

@@ -0,0 +1,35 @@
+//! Kahn's algorithm inputs: every key, each key's upstream count and its downstream keys.
+
+use std::collections::{HashMap, HashSet};
+
+use crate::planner_identity::models::GraphKey;
+
+/// Every key named in the edges, its number of upstream entries and its downstream keys, one per edge.
+pub(crate) struct DependencyCounts<'a> {
+    pub(crate) node_keys: HashSet<&'a GraphKey>,
+    pub(crate) indegree: HashMap<&'a GraphKey, usize>,
+    pub(crate) downstream: HashMap<&'a GraphKey, Vec<&'a GraphKey>>,
+}
+
+/// The counts of `upstream`, counting repeated upstream entries once per entry.
+pub(crate) fn dependency_counts(upstream: &[(GraphKey, Vec<GraphKey>)]) -> DependencyCounts<'_> {
+    let mut counts = DependencyCounts {
+        node_keys: HashSet::new(),
+        indegree: HashMap::new(),
+        downstream: HashMap::new(),
+    };
+    for (key, dep_keys) in upstream {
+        counts.node_keys.insert(key);
+        for dep_key in dep_keys {
+            counts.node_keys.insert(dep_key);
+            counts.downstream.entry(dep_key).or_default().push(key);
+        }
+    }
+    for key in &counts.node_keys {
+        counts.indegree.insert(key, 0);
+    }
+    for (key, dep_keys) in upstream {
+        counts.indegree.insert(key, dep_keys.len());
+    }
+    counts
+}

crates/sqlbuild-analysis/src/planner_identity/_helpers/identity_resolver.rs

@@ -0,0 +1,169 @@
+//! Memoized identity resolution over one dependency graph, each node once, with an explicit stack.
+
+use std::collections::{HashMap, HashSet};
+
+use crate::planner_identity::_helpers::compose_graph_identity::compose_graph_identity;
+use crate::planner_identity::models::{GraphKey, IdentityNode};
+use crate::planner_identity::types::IdentityHashes;
+
+/// Which hashes a resolution composes from.
+#[derive(Clone, Copy)]
+enum Mode {
+    Expected,
+    Write,
+}
+
+/// A node whose upstream hashes are being resolved.
+struct Frame<'a> {
+    key: &'a GraphKey,
+    node: &'a IdentityNode,
+    local_hash: &'a str,
+    next_upstream: usize,
+    upstream_hashes: Vec<String>,
+}
+
+/// What entering a node yields: its hash already, or a frame to resolve.
+enum Entered<'a> {
+    Resolved(Option<String>),
+    Pending(Frame<'a>),
+}
+
+/// Expected or selected-write identity resolution state.
+pub(crate) struct IdentityResolver<'a> {
+    nodes: &'a HashMap<GraphKey, IdentityNode>,
+    selected: Option<&'a HashSet<GraphKey>>,
+    hashes: IdentityHashes,
+    positions: HashMap<GraphKey, usize>,
+    resolved_selected: HashMap<GraphKey, String>,
+    visiting: HashSet<GraphKey>,
+}
+
+impl<'a> IdentityResolver<'a> {
+    pub(crate) fn new(
+        nodes: &'a HashMap<GraphKey, IdentityNode>,
+        selected: Option<&'a HashSet<GraphKey>>,
+        base_hashes: Vec<(GraphKey, String)>,
+    ) -> Self {
+        let mut resolver = Self {
+            nodes,
+            selected,
+            hashes: Vec::new(),
+            positions: HashMap::new(),
+            resolved_selected: HashMap::new(),
+            visiting: HashSet::new(),
+        };
+        for (key, hash) in base_hashes {
+            resolver.record(&key, Some(hash));
+        }
+        resolver
+    }
+
+    /// Every hash recorded so far, in first-recorded order.
+    pub(crate) fn into_hashes(self) -> IdentityHashes {
+        self.hashes
+    }
+
+    /// A node's expected hash, composed from its upstream expected hashes.
+    pub(crate) fn resolve_expected(&mut self, key: &'a GraphKey) -> Option<String> {
+        self.resolve(key, Mode::Expected)
+    }
+
+    /// A selected node's hash recomputed from upstream write hashes; other nodes keep theirs.
+    pub(crate) fn resolve_write(&mut self, key: &'a GraphKey) -> Option<String> {
+        self.resolve(key, Mode::Write)
+    }
+
+    fn resolve(&mut self, root: &'a GraphKey, mode: Mode) -> Option<String> {
+        let mut stack: Vec<Frame<'a>> = match self.enter(root, mode) {
+            Entered::Resolved(hash) => return hash,
+            Entered::Pending(frame) => vec![frame],
+        };
+        while let Some(frame) = stack.last_mut() {
+            if let Some(upstream_key) = frame.node.upstream_keys.get(frame.next_upstream) {
+                frame.next_upstream += 1;
+                match self.enter(upstream_key, mode) {
+                    Entered::Resolved(hash) => frame.upstream_hashes.extend(hash),
+                    Entered::Pending(upstream_frame) => stack.push(upstream_frame),
+                }
+                continue;
+            }
+            let Some(finished) = stack.pop() else {
+                break;
+            };
+            let composed: String = self.finish(&finished, mode);
+            match stack.last_mut() {
+                Some(parent) => parent.upstream_hashes.push(composed),
+                None => return Some(composed),
+            }
+        }
+        None
+    }
+
+    fn enter(&mut self, key: &'a GraphKey, mode: Mode) -> Entered<'a> {
+        if let Mode::Write = mode {
+            if !self.selected.is_some_and(|selected| selected.contains(key)) {
+                return Entered::Resolved(self.recorded(key));
+            }
+            if let Some(hash) = self.resolved_selected.get(key) {
+                return Entered::Resolved(Some(hash.clone()));
+            }
+        } else if let Some(position) = self.positions.get(key) {
+            return Entered::Resolved(self.hashes[*position].1.clone());
+        }
+        let nodes: &'a HashMap<GraphKey, IdentityNode> = self.nodes;
+        let Some((node, local_hash)) = nodes
+            .get(key)
+            .and_then(|node| node.local_hash.as_deref().map(|local| (node, local)))
+        else {
+            return Entered::Resolved(match mode {
+                Mode::Expected => {
+                    self.record(key, None);
+                    None
+                }
+                Mode::Write => self.recorded(key),
+            });
+        };
+        if self.visiting.contains(key) {
+            if let Mode::Expected = mode {
+                self.record(key, Some(local_hash.to_owned()));
+            }
+            return Entered::Resolved(Some(local_hash.to_owned()));
+        }
+        self.visiting.insert(key.clone());
+        Entered::Pending(Frame {
+            key,
+            node,
+            local_hash,
+            next_upstream: 0,
+            upstream_hashes: Vec::new(),
+        })
+    }
+
+    fn finish(&mut self, frame: &Frame<'a>, mode: Mode) -> String {
+        self.visiting.remove(frame.key);
+        let upstream: Vec<&str> = frame.upstream_hashes.iter().map(String::as_str).collect();
+        let composed: String = compose_graph_identity(frame.local_hash, &upstream);
+        self.record(frame.key, Some(composed.clone()));
+        if let Mode::Write = mode {
+            self.resolved_selected
+                .insert(frame.key.clone(), composed.clone());
+        }
+        composed
+    }
+
+    fn recorded(&self, key: &GraphKey) -> Option<String> {
+        self.positions
+            .get(key)
+            .and_then(|position| self.hashes[*position].1.clone())
+    }
+
+    fn record(&mut self, key: &GraphKey, hash: Option<String>) {
+        match self.positions.get(key) {
+            Some(position) => self.hashes[*position].1 = hash,
+            None => {
+                self.positions.insert(key.clone(), self.hashes.len());
+                self.hashes.push((key.clone(), hash));
+            }
+        }
+    }
+}

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

@@ -0,0 +1,5 @@
+pub(crate) mod compose_graph_identity;
+pub(crate) mod cycle_message;
+pub(crate) mod dependency_counts;
+pub(crate) mod identity_resolver;
+pub(crate) mod staleness_walk;