knut-bifrost 0.1.3

//! Declarative YAML mapping spec — the fabric's blueprint.

use serde::{Deserialize, Serialize};
use std::sync::Arc;

use crate::{
    error::{BifrostError, Result},
    registry::TransformRegistry,
    transform::{EdgeTransform, Transform, VertexTransform},
};

/// Top-level mapping file.
///
/// ```yaml
/// version: 1
/// tables:
///   - table: lake.social.users
///     mode: incremental
///     pipeline:
///       - vertex: { label: Person, id_column: user_id, properties: [name] }
///       - fk_edges:
///           type: WORKS_AT
///           src: { label: Person,  column: user_id }
///           dst: { label: Company, column: company_id }
///           properties: [since]          # edge property columns
///           rename: { since: joined_at } # optional column → prop renaming
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MappingSpec {
    pub version: u32,
    pub tables: Vec<TableMapping>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TableMapping {
    /// Fully qualified table name as Spark sees it.
    pub table: String,
    #[serde(default)]
    pub mode: SyncMode,
    pub pipeline: Vec<PipelineStep>,
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SyncMode {
    /// Re-read the whole table when its snapshot changes (idempotent MERGE).
    #[default]
    Full,
    /// Read only rows appended since the last processed snapshot (`EXCEPT ALL`
    /// diff). Append-only: it sees inserts, not updates or deletes.
    Incremental,
    /// Changelog (CDC) mode: read Iceberg's `system.create_changelog_view`
    /// between the last and the current snapshot, exposing a `_change_type`
    /// column (`INSERT` / `UPDATE_AFTER` / `UPDATE_BEFORE` / `DELETE`).
    /// `INSERT`/`UPDATE_AFTER` rows become upserts; `DELETE` rows become
    /// `DETACH DELETE`s. `UPDATE_BEFORE` (the update pre-image) is dropped — its
    /// `UPDATE_AFTER` upsert already carries the new state. The first round (no
    /// prior snapshot) reads the whole table as inserts, like `Full`.
    Cdc,
}

impl SyncMode {
    /// The sentinel column the Iceberg changelog view adds in [`Cdc`] mode,
    /// and that the listener partitions on before stripping it from values.
    ///
    /// [`Cdc`]: SyncMode::Cdc
    pub const CHANGE_TYPE_COLUMN: &'static str = "_change_type";
}

/// One `- vertex: …` / `- edge: …` / `- fk_edges: …` / `- custom: name`
/// entry. Untagged so the YAML reads as a single-key map per step.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(untagged)]
pub enum PipelineStep {
    Vertex {
        vertex: VertexTransform,
    },
    Edge {
        edge: EdgeTransform,
    },
    /// Same shape as `edge`; named for FK-derived relationships.
    FkEdges {
        fk_edges: EdgeTransform,
    },
    /// Reference to a transform registered in [`TransformRegistry`].
    Custom {
        custom: String,
    },
}

impl MappingSpec {
    pub fn from_yaml(yaml: &str) -> Result<Self> {
        Ok(serde_yaml::from_str(yaml)?)
    }

    pub fn load(path: impl AsRef<std::path::Path>) -> Result<Self> {
        Self::from_yaml(&std::fs::read_to_string(path)?)
    }
}

impl TableMapping {
    /// Resolve the pipeline into executable transforms.
    pub fn compile(&self, registry: &TransformRegistry) -> Result<Vec<Arc<dyn Transform>>> {
        if self.pipeline.is_empty() {
            return Err(BifrostError::EmptyPipeline(self.table.clone()));
        }
        self.pipeline
            .iter()
            .map(|step| -> Result<Arc<dyn Transform>> {
                Ok(match step {
                    PipelineStep::Vertex { vertex } => Arc::new(vertex.clone()),
                    PipelineStep::Edge { edge: t } | PipelineStep::FkEdges { fk_edges: t } => {
                        Arc::new(t.clone())
                    }
                    PipelineStep::Custom { custom } => registry.get(custom)?,
                })
            })
            .collect()
    }
}