refactor: host planner warehouse inspection and CLI helpers in Rust

crates/sqlbuild-analysis/src/lib.rs

@@ -27,6 +27,7 @@ pub mod planner_scenario;
 pub mod planner_scope;
 pub mod planner_selection;
 pub mod planner_strategy;
+pub mod planner_warehouse;
 pub mod python_plan;
 pub mod python_selection;
 pub mod python_values;

crates/sqlbuild-analysis/src/planner_warehouse/_helpers/cache_entries.rs

@@ -0,0 +1,12 @@
+//! Digests binding stored fingerprint cache payloads to their keys.
+
+use sha2::{Digest, Sha256};
+
+/// SHA-256 of the cache key, a NUL byte and the serialized payload.
+pub(crate) fn entry_digest(cache_key: &str, serialized: &str) -> String {
+    let mut digest = Sha256::new();
+    digest.update(cache_key.as_bytes());
+    digest.update(b"\0");
+    digest.update(serialized.as_bytes());
+    format!("{:x}", digest.finalize())
+}

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

@@ -0,0 +1,3 @@
+//! Shared helpers of warehouse-inspection planning.
+
+pub(crate) mod cache_entries;

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

@@ -0,0 +1,29 @@
+//! Fixed texts and limits of warehouse-inspection planning.
+
+pub(crate) const TIMESTAMP_CURSOR_TYPE: &str = "timestamp";
+pub(crate) const RETENTION_INCREASE: &str = "increase";
+pub(crate) const RETENTION_DECREASE: &str = "decrease";
+pub(crate) const RETENTION_MIXED: &str = "mixed";
+pub(crate) const RETENTION_MATCH: &str = "match";
+pub(crate) const RETENTION_PHASE_PRE: &str = "pre";
+pub(crate) const RETENTION_PHASE_POST: &str = "post";
+pub(crate) const CHANGE_PHASE_PREPARE: &str = "prepare";
+pub(crate) const CHANGE_PHASE_FINALIZE: &str = "finalize";
+pub(crate) const TABLE_TYPE_COPY_PREFIX: &str = "__sqb_type_swap__";
+pub(crate) const MESSAGE_ITEM_LIMIT: usize = 10;
+pub(crate) const FRESHNESS_NAME_LIMIT: usize = 10;
+pub(crate) const SOURCE_FRESHNESS_UNKNOWN_WARNING_TITLE: &str = "source freshness unknown";
+pub(crate) const FRESHNESS_UNAVAILABLE_EXPLANATION: &str = "adapter freshness metadata is unavailable, for example for a view, an external table, or a table without a last-modified time";
+pub(crate) const FRESHNESS_ERROR_EXPLANATION: &str = "the freshness observation failed";
+pub(crate) const FRESHNESS_UNAVAILABLE: &str = "unavailable";
+pub(crate) const FRESHNESS_ERROR: &str = "error";
+pub(crate) const SOURCE_CALL_OPEN: &str = "__source(\"";
+pub(crate) const CONTEXT_PLACEHOLDER_OPEN: &str = "${CTX:";
+pub(crate) const VARIABLE_PLACEHOLDER_OPEN: &str = "${";
+pub(crate) const SOURCE_DEFERRAL_CONTEXT_FIELDS: [&str; 2] = ["schema", "database"];
+pub(crate) const FINGERPRINT_CACHE_VERSION: i64 = 1;
+pub(crate) const FINGERPRINT_CACHE_ENTRY_SEPARATOR: char = '\n';
+pub(crate) const MAX_FINGERPRINT_CACHE_ENTRY_CHARACTERS: usize = 4_096;
+pub(crate) const MIGRATION_FINGERPRINT_ALGORITHM: &str = "migration-fingerprint-v1";
+pub(crate) const CHANGE_KIND_FIRST_RUN: &str = "first_run";
+pub(crate) const CHANGE_KIND_RENAMED: &str = "renamed";

crates/sqlbuild-analysis/src/planner_warehouse/main/cursor_bounds.rs

@@ -0,0 +1,13 @@
+//! Which bounds one physical cursor read selects.
+
+use crate::planner_warehouse::models::PhysicalCursorQuery;
+
+/// `min`, `max` or `min,max`, as shown in progress and debug events.
+pub fn cursor_bounds(query: &PhysicalCursorQuery) -> String {
+    match (query.min_tags.is_empty(), query.max_tags.is_empty()) {
+        (false, false) => "min,max".to_owned(),
+        (false, true) => "min".to_owned(),
+        (true, false) => "max".to_owned(),
+        (true, true) => String::new(),
+    }
+}

crates/sqlbuild-analysis/src/planner_warehouse/main/cursor_query_groups.rs

@@ -0,0 +1,41 @@
+//! Group logical cursor bounds by the physical relation column that answers them.
+
+use crate::planner_warehouse::constants::TIMESTAMP_CURSOR_TYPE;
+use crate::planner_warehouse::models::{CursorTagRequest, PhysicalCursorQuery};
+
+/// Merge requests into one query per relation, column and type, in first-request order.
+pub fn cursor_query_groups(requests: &[CursorTagRequest]) -> Vec<PhysicalCursorQuery> {
+    let mut groups: Vec<PhysicalCursorQuery> = Vec::new();
+    for request in requests {
+        let cursor_type = request
+            .cursor_type
+            .clone()
+            .filter(|value| !value.is_empty())
+            .unwrap_or_else(|| TIMESTAMP_CURSOR_TYPE.to_owned());
+        let position = groups.iter().position(|group| {
+            group.relation == request.relation
+                && group.cursor_column == request.cursor_column
+                && group.cursor_type == cursor_type
+        });
+        let index = position.unwrap_or_else(|| {
+            groups.push(PhysicalCursorQuery {
+                relation: request.relation.clone(),
+                cursor_column: request.cursor_column.clone(),
+                cursor_type,
+                min_tags: Vec::new(),
+                max_tags: Vec::new(),
+            });
+            groups.len() - 1
+        });
+        let group = &mut groups[index];
+        if let Some(min_tag) = &request.min_tag
+            && !group.min_tags.contains(min_tag)
+        {
+            group.min_tags.push(min_tag.clone());
+        }
+        if !group.max_tags.contains(&request.max_tag) {
+            group.max_tags.push(request.max_tag.clone());
+        }
+    }
+    groups
+}

crates/sqlbuild-analysis/src/planner_warehouse/main/cursor_query_identity.rs

@@ -0,0 +1,17 @@
+//! The progress label of one physical cursor read.
+
+use crate::planner_warehouse::main::cursor_bounds::cursor_bounds;
+use crate::planner_warehouse::models::PhysicalCursorQuery;
+
+/// `(index/total): relation.column [bounds]`.
+pub fn cursor_query_identity(
+    query: &PhysicalCursorQuery,
+    (index, total): (usize, usize),
+) -> String {
+    format!(
+        "({index}/{total}): {}.{} [{}]",
+        query.relation,
+        query.cursor_column,
+        cursor_bounds(query)
+    )
+}

crates/sqlbuild-analysis/src/planner_warehouse/main/cursor_query_sql.rs

@@ -0,0 +1,21 @@
+//! The unmerged MIN/MAX statement of one physical cursor read.
+
+use crate::planner_warehouse::models::PhysicalCursorQuery;
+
+/// Select the requested bounds of the cursor column as text.
+pub fn cursor_query_sql(query: &PhysicalCursorQuery) -> String {
+    let mut parts: Vec<String> = Vec::new();
+    if !query.min_tags.is_empty() {
+        parts.push(format!(
+            "CAST(MIN({}) AS VARCHAR) AS _min",
+            query.cursor_column
+        ));
+    }
+    if !query.max_tags.is_empty() {
+        parts.push(format!(
+            "CAST(MAX({}) AS VARCHAR) AS _max",
+            query.cursor_column
+        ));
+    }
+    format!("SELECT {} FROM {}", parts.join(", "), query.relation)
+}

crates/sqlbuild-analysis/src/planner_warehouse/main/fingerprint_cache_entry.rs

@@ -0,0 +1,36 @@
+//! The stored payload of one migration fingerprint cache entry.
+
+use crate::planner_warehouse::_helpers::cache_entries::entry_digest;
+use crate::planner_warehouse::constants::{
+    FINGERPRINT_CACHE_ENTRY_SEPARATOR, FINGERPRINT_CACHE_VERSION,
+};
+use sqlbuild_core::json::errors::JsonEmitError;
+use sqlbuild_core::json::main::dumps::dumps;
+use sqlbuild_core::json::models::{JsonDialect, JsonInteger, JsonValue, StdlibJsonOptions};
+
+/// `digest` newline `{"f":...,"k":...,"v":1}`, the digest binding the payload to its key.
+pub fn fingerprint_cache_entry(
+    cache_key: &str,
+    fingerprint: Option<&str>,
+) -> Result<String, JsonEmitError> {
+    let value = JsonValue::Object(vec![
+        (
+            "v".to_owned(),
+            JsonValue::Integer(JsonInteger::from(FINGERPRINT_CACHE_VERSION)),
+        ),
+        ("k".to_owned(), JsonValue::String(cache_key.to_owned())),
+        (
+            "f".to_owned(),
+            fingerprint.map_or(JsonValue::Null, |text| JsonValue::String(text.to_owned())),
+        ),
+    ]);
+    let options = StdlibJsonOptions {
+        sort_keys: true,
+        ..StdlibJsonOptions::new(None).with_separators(",", ":")
+    };
+    let serialized = dumps(&value, &JsonDialect::Stdlib(options))?;
+    Ok(format!(
+        "{}{FINGERPRINT_CACHE_ENTRY_SEPARATOR}{serialized}",
+        entry_digest(cache_key, &serialized)
+    ))
+}

crates/sqlbuild-analysis/src/planner_warehouse/main/fingerprint_cache_key.rs

@@ -0,0 +1,36 @@
+//! The persistent key of one migration fingerprint cache entry.
+
+use crate::planner_warehouse::models::FingerprintCacheKey;
+use sha2::{Digest, Sha256};
+use sqlbuild_core::json::errors::JsonEmitError;
+use sqlbuild_core::json::main::dumps::dumps;
+use sqlbuild_core::json::models::{JsonDialect, JsonValue, StdlibJsonOptions};
+
+/// SHA-256 of the compact, non-ASCII-escaped JSON array of the key parts.
+pub fn fingerprint_cache_key(key: &FingerprintCacheKey) -> Result<String, JsonEmitError> {
+    let renames: Vec<JsonValue> = key
+        .visible_renames
+        .iter()
+        .map(|(name, identity)| {
+            JsonValue::Array(vec![
+                JsonValue::String(name.clone()),
+                JsonValue::String(identity.clone()),
+            ])
+        })
+        .collect();
+    let value = JsonValue::Array(vec![
+        JsonValue::String(key.shared_digest.clone()),
+        JsonValue::String(key.query_sql.clone()),
+        JsonValue::String(key.metadata_json.clone()),
+        key.dialect
+            .clone()
+            .map_or(JsonValue::Null, JsonValue::String),
+        JsonValue::Array(renames),
+    ]);
+    let options = StdlibJsonOptions {
+        ensure_ascii: false,
+        ..StdlibJsonOptions::new(None).with_separators(",", ":")
+    };
+    let text = dumps(&value, &JsonDialect::Stdlib(options))?;
+    Ok(format!("{:x}", Sha256::digest(text.as_bytes())))
+}

crates/sqlbuild-analysis/src/planner_warehouse/main/fingerprint_cache_lookup.rs

@@ -0,0 +1,39 @@
+//! Validate one stored migration fingerprint cache entry.
+
+use crate::planner_warehouse::_helpers::cache_entries::entry_digest;
+use crate::planner_warehouse::constants::{
+    FINGERPRINT_CACHE_ENTRY_SEPARATOR, FINGERPRINT_CACHE_VERSION,
+    MAX_FINGERPRINT_CACHE_ENTRY_CHARACTERS,
+};
+use serde_json::Value;
+
+/// The stored fingerprint (possibly `None`), or `None` when the entry is not valid for the key.
+pub fn fingerprint_cache_lookup(
+    contents: &str,
+    expected_cache_key: &str,
+) -> Option<Option<String>> {
+    if contents.chars().count() > MAX_FINGERPRINT_CACHE_ENTRY_CHARACTERS {
+        return None;
+    }
+    let (stored_digest, serialized) = contents.split_once(FINGERPRINT_CACHE_ENTRY_SEPARATOR)?;
+    if !stored_digest.is_ascii() || stored_digest != entry_digest(expected_cache_key, serialized) {
+        return None;
+    }
+    let Ok(Value::Object(values)) = serde_json::from_str::<Value>(serialized) else {
+        return None;
+    };
+    let version_matches = match values.get("v") {
+        Some(Value::Number(number)) => number.as_f64() == Some(FINGERPRINT_CACHE_VERSION as f64),
+        Some(Value::Bool(flag)) => *flag,
+        _ => false,
+    };
+    let key_matches = values.get("k").and_then(Value::as_str) == Some(expected_cache_key);
+    if !version_matches || !key_matches {
+        return None;
+    }
+    match values.get("f") {
+        None | Some(Value::Null) => Some(None),
+        Some(Value::String(fingerprint)) => Some(Some(fingerprint.clone())),
+        Some(_) => None,
+    }
+}

crates/sqlbuild-analysis/src/planner_warehouse/main/fingerprint_cache_shared_digest.rs

@@ -0,0 +1,40 @@
+//! The digest shared by every migration fingerprint cache key of one build.
+
+use crate::planner_warehouse::constants::{
+    FINGERPRINT_CACHE_VERSION, MIGRATION_FINGERPRINT_ALGORITHM,
+};
+use sha2::{Digest, Sha256};
+use sqlbuild_core::json::errors::JsonEmitError;
+use sqlbuild_core::json::main::dumps::dumps;
+use sqlbuild_core::json::models::{JsonDialect, JsonInteger, JsonValue, StdlibJsonOptions};
+
+/// SHA-256 of the sorted JSON of the algorithm, cache version and package versions.
+pub fn fingerprint_cache_shared_digest(
+    sqlbuild_version: &str,
+    polyglot_version: &str,
+) -> Result<String, JsonEmitError> {
+    let value = JsonValue::Object(vec![
+        (
+            "algorithm".to_owned(),
+            JsonValue::String(MIGRATION_FINGERPRINT_ALGORITHM.to_owned()),
+        ),
+        (
+            "cache_version".to_owned(),
+            JsonValue::Integer(JsonInteger::from(FINGERPRINT_CACHE_VERSION)),
+        ),
+        (
+            "sqlbuild_version".to_owned(),
+            JsonValue::String(sqlbuild_version.to_owned()),
+        ),
+        (
+            "polyglot_version".to_owned(),
+            JsonValue::String(polyglot_version.to_owned()),
+        ),
+    ]);
+    let options = StdlibJsonOptions {
+        sort_keys: true,
+        ..StdlibJsonOptions::new(None)
+    };
+    let text = dumps(&value, &JsonDialect::Stdlib(options))?;
+    Ok(format!("{:x}", Sha256::digest(text.as_bytes())))
+}

crates/sqlbuild-analysis/src/planner_warehouse/main/freshness_unknown_message.rs

@@ -0,0 +1,31 @@
+//! The plan warning for sources whose freshness could not be observed.
+
+use crate::planner_warehouse::constants::{
+    FRESHNESS_ERROR, FRESHNESS_ERROR_EXPLANATION, FRESHNESS_NAME_LIMIT, FRESHNESS_UNAVAILABLE,
+    FRESHNESS_UNAVAILABLE_EXPLANATION, SOURCE_FRESHNESS_UNKNOWN_WARNING_TITLE,
+};
+
+/// The warning for one explained reason, or `None` for reasons that need no warning.
+pub fn freshness_unknown_message(reason: &str, names: &[String]) -> Option<String> {
+    let explanation = if reason == FRESHNESS_UNAVAILABLE {
+        FRESHNESS_UNAVAILABLE_EXPLANATION
+    } else if reason == FRESHNESS_ERROR {
+        FRESHNESS_ERROR_EXPLANATION
+    } else {
+        return None;
+    };
+    let mut ordered: Vec<&String> = names.iter().collect();
+    ordered.sort();
+    let mut shown = ordered
+        .iter()
+        .take(FRESHNESS_NAME_LIMIT)
+        .map(|name| name.as_str())
+        .collect::<Vec<_>>()
+        .join(", ");
+    if ordered.len() > FRESHNESS_NAME_LIMIT {
+        shown = format!("{shown} and {} more", ordered.len() - FRESHNESS_NAME_LIMIT);
+    }
+    Some(format!(
+        "{SOURCE_FRESHNESS_UNKNOWN_WARNING_TITLE} ({reason}) for {shown}: {explanation}; run sqb freshness for each source's details"
+    ))
+}

crates/sqlbuild-analysis/src/planner_warehouse/main/full_rebuild_cause.rs

@@ -0,0 +1,26 @@
+//! Why an incremental model is planned as a full rebuild.
+
+use crate::planner_warehouse::constants::{CHANGE_KIND_FIRST_RUN, CHANGE_KIND_RENAMED};
+use crate::planner_warehouse::models::FullRebuildFacts;
+
+/// The cause text shown in plans and full-refresh refusals.
+pub fn full_rebuild_cause(facts: &FullRebuildFacts) -> String {
+    if facts.full_refresh {
+        "--full-refresh".to_owned()
+    } else if facts.change_kind == CHANGE_KIND_FIRST_RUN {
+        "first run".to_owned()
+    } else if facts.change_kind == CHANGE_KIND_RENAMED {
+        "rename".to_owned()
+    } else if facts.query_changed {
+        "query changed with replay_on_change full".to_owned()
+    } else if !facts.changed_functions.is_empty() {
+        format!(
+            "function {} changed with replay_on_change full",
+            facts.changed_functions.join(", ")
+        )
+    } else if facts.has_schema_findings {
+        "schema changed with replay_on_change full".to_owned()
+    } else {
+        "full backfill".to_owned()
+    }
+}

crates/sqlbuild-analysis/src/planner_warehouse/main/missing_source_tables_help.rs

@@ -0,0 +1,32 @@
+//! The remedy for selected source tables the warehouse cannot read.
+
+use crate::planner_warehouse::constants::MESSAGE_ITEM_LIMIT;
+use crate::planner_warehouse::models::MissingSourceTable;
+use std::collections::BTreeSet;
+
+/// Suggest creating, granting or correcting the sources, with their declarations.
+pub fn missing_source_tables_help(missing: &[MissingSourceTable]) -> String {
+    let locations: Vec<String> = missing
+        .iter()
+        .take(MESSAGE_ITEM_LIMIT)
+        .map(|item| format!("'{}' at {}", item.source_name, item.location))
+        .collect();
+    let roles: BTreeSet<&str> = missing
+        .iter()
+        .filter_map(|item| item.role.as_deref())
+        .collect();
+    let role_text = if roles.is_empty() {
+        String::new()
+    } else {
+        format!(" ({})", roles.into_iter().collect::<Vec<_>>().join(", "))
+    };
+    let grant = if missing.iter().any(|item| item.unreadable) {
+        format!("grant SELECT on it to the role in use{role_text}, ")
+    } else {
+        String::new()
+    };
+    format!(
+        "create or load the table, {grant}correct its database, schema, or table in the source declaration, or stop reading it from the selected models; declared {}",
+        locations.join(", ")
+    )
+}