kglite 0.17.10

//! Blueprint compute pipeline executor.
//!
//! Runs the `compute:` block ops as a CSV-shaping pre-phase: each
//! op reads its source CSV (already on disk in `processed/`), runs
//! the primitive, and writes its output into `computed/`. The
//! existing 5-phase loader then consumes those CSVs as if they
//! were regular blueprint inputs — no in-memory graph mutation,
//! no special phase, no read-back of column stores.
//!
//! Each primitive mutates the `Blueprint` in place to:
//!   - Repoint a NodeSpec's `csv` field to the augmented file
//!     (`derive` adds property columns to the source)
//!   - Add a new NodeSpec for output types (`filter.into`,
//!     `aggregate.into`, `calendar.type`)
//!   - Add junction-edge entries for chain/calendar/aggregate edges
//!
//! After this pre-phase, `build()`'s collect_specs sees the
//! augmented blueprint and the load proceeds normally.

use std::path::Path;

/// Compute opens and rewrites its source files itself, so it resolves paths
/// the way the input registry does — one rule for where an input lives.
pub(crate) use super::input::resolve_input_path;
use super::schema::{Blueprint, ComputeOp};

pub mod aggregate;
pub mod calendar;
pub mod chain;
pub mod derive;
pub mod filter;
mod output;
mod paths;
mod values;

/// Preferred filename spelling only; ComputePaths resolves collisions and reserves inputs.
pub(super) fn sanitize_filename(s: &str) -> String {
    s.chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '_' {
                c
            } else {
                '_'
            }
        })
        .collect()
}

/// Run every compute op in declared order. Each op reads from
/// `input_root` (typically the blueprint's resolved root) and
/// writes its output CSVs under `input_root/computed/`. The
/// blueprint is mutated to point at the new CSVs.
pub fn apply_compute(blueprint: &mut Blueprint, input_root: &Path) -> Result<(), String> {
    if blueprint.compute.is_empty() {
        return Ok(());
    }

    // Ensure the output directory exists.
    let computed_dir = input_root.join("computed");
    std::fs::create_dir_all(&computed_dir).map_err(|e| {
        format!(
            "compute: failed to create {}: {}",
            computed_dir.display(),
            e
        )
    })?;

    let paths = paths::ComputePaths::new(blueprint, input_root, &blueprint.compute)?;

    // Move ops out of the blueprint so we can iterate them while
    // mutating the rest of the blueprint structure.
    let ops = std::mem::take(&mut blueprint.compute);
    for (i, op) in ops.iter().enumerate() {
        dispatch(op, blueprint, input_root, &paths)
            .map_err(|e| format!("compute[{}] ({}): {}", i, op_name(op), e))?;
    }
    Ok(())
}

fn dispatch(
    op: &ComputeOp,
    blueprint: &mut Blueprint,
    input_root: &Path,
    paths: &paths::ComputePaths,
) -> Result<(), String> {
    match op {
        ComputeOp::Derive { from, set } => {
            derive::run_derive_allocated(blueprint, input_root, from, set, Some(paths))
        }
        ComputeOp::Filter {
            from,
            where_expr,
            into,
        } => filter::run_filter_allocated(
            blueprint,
            input_root,
            from,
            where_expr,
            into.as_deref(),
            Some(paths),
        ),
        ComputeOp::Chain {
            from,
            group_by,
            order_by,
            edge,
        } => chain::run_chain_allocated(
            blueprint,
            input_root,
            from,
            group_by,
            order_by,
            edge,
            Some(paths),
        ),
        ComputeOp::Calendar {
            node_type,
            start,
            end,
            next_edge,
            in_month_edge,
            in_quarter_edge,
            in_year_edge,
            links,
        } => calendar::run_calendar_allocated(
            blueprint,
            input_root,
            node_type,
            start,
            end,
            next_edge,
            in_month_edge.as_deref(),
            in_quarter_edge.as_deref(),
            in_year_edge.as_deref(),
            links,
            Some(paths),
        ),
        ComputeOp::Aggregate {
            from,
            group_by,
            into,
            agg,
            edges,
        } => aggregate::run_aggregate_allocated(
            blueprint,
            input_root,
            from,
            group_by,
            into,
            agg,
            edges,
            Some(paths),
        ),
    }
}

fn op_name(op: &ComputeOp) -> &'static str {
    match op {
        ComputeOp::Derive { .. } => "derive",
        ComputeOp::Filter { .. } => "filter",
        ComputeOp::Chain { .. } => "chain",
        ComputeOp::Calendar { .. } => "calendar",
        ComputeOp::Aggregate { .. } => "aggregate",
    }
}

/// Resolve a source-type name to its NodeSpec. Looks in
/// `blueprint.nodes` first, then walks each parent's `sub_nodes` —
/// so compute primitives can target sub-node types (e.g. SEC's
/// `Transaction`, which lives at
/// `nodes.Person.sub_nodes.Transaction`).
pub(crate) fn resolve_source_spec<'a>(
    blueprint: &'a super::schema::Blueprint,
    name: &str,
) -> Option<&'a super::schema::NodeSpec> {
    if let Some(s) = blueprint.nodes.get(name) {
        return Some(s);
    }
    for parent in blueprint.nodes.values() {
        if let Some(s) = parent.sub_nodes.get(name) {
            return Some(s);
        }
    }
    None
}

/// Mutable counterpart of `resolve_source_spec`. Returns the
/// NodeSpec at its declared location so callers can rewire `csv:`
/// or extend `properties:` / `connections:` in place — sub-node
/// mutations stay on the parent's `sub_nodes` map, top-level
/// mutations on `blueprint.nodes`.
pub(crate) fn resolve_source_spec_mut<'a>(
    blueprint: &'a mut super::schema::Blueprint,
    name: &str,
) -> Option<&'a mut super::schema::NodeSpec> {
    if blueprint.nodes.contains_key(name) {
        return blueprint.nodes.get_mut(name);
    }
    for parent in blueprint.nodes.values_mut() {
        if parent.sub_nodes.contains_key(name) {
            return parent.sub_nodes.get_mut(name);
        }
    }
    None
}

/// Convert a Value (from expression evaluation) to a CSV cell.
/// Handles null → empty, booleans as "true"/"false", floats with
/// reasonable precision (no scientific notation for typical money
/// values).
pub(crate) fn value_to_csv_cell(v: &super::expr::Value) -> String {
    use super::expr::Value;
    match v {
        Value::Null => String::new(),
        Value::Bool(b) => {
            if *b {
                "true".to_string()
            } else {
                "false".to_string()
            }
        }
        Value::Int(i) => i.to_string(),
        Value::Float(f) => {
            if f.is_nan() {
                String::new()
            } else if f.fract() == 0.0 && f.abs() < 1e15 {
                format!("{}.0", *f as i64)
            } else {
                format!("{}", f)
            }
        }
        Value::String(s) => s.clone(),
        Value::List(_) => v.to_string(),
    }
}

/// Map a CSV cell string + declared type to a Value the expression
/// engine can consume. Empty → Null. The declared type comes from
/// the NodeSpec's properties map (when set) or falls back to
/// inference (number? bool? string?).
pub(crate) fn csv_cell_to_value(cell: &str, declared_type: Option<&str>) -> super::expr::Value {
    use super::expr::Value;
    let trimmed = cell.trim();
    if trimmed.is_empty() {
        return Value::Null;
    }
    match declared_type {
        Some("int") | Some("integer") => trimmed
            .parse::<i64>()
            .map(Value::Int)
            .unwrap_or_else(|_| Value::String(cell.to_string())),
        Some("float") => trimmed
            .parse::<f64>()
            .map(Value::Float)
            .unwrap_or_else(|_| Value::String(cell.to_string())),
        Some("bool") | Some("boolean") => match trimmed.to_lowercase().as_str() {
            "true" | "1" | "yes" | "y" => Value::Bool(true),
            "false" | "0" | "no" | "n" => Value::Bool(false),
            _ => Value::String(cell.to_string()),
        },
        Some("string") | Some("str") => Value::String(cell.to_string()),
        Some(_) | None => {
            // Infer: int → float → string.
            if let Ok(i) = trimmed.parse::<i64>() {
                Value::Int(i)
            } else if let Ok(f) = trimmed.parse::<f64>() {
                Value::Float(f)
            } else {
                Value::String(cell.to_string())
            }
        }
    }
}