knut-bifrost 0.1.3

//! The common currency between transforms and sinks.

use arrow_array::RecordBatch;

/// Nodes to upsert: one batch per label.
#[derive(Debug, Clone)]
pub struct VertexSet {
    pub label: String,
    /// Column whose value identifies the node (MERGE key).
    pub id_column: String,
    pub batch: RecordBatch,
}

/// Nodes to remove. The `batch` carries the id column for the label; every
/// row identifies one node to `MATCH (n:Label {id: x}) DETACH DELETE n`.
///
/// Deleting a vertex DETACH-deletes its incident edges too, so a changelog
/// `DELETE` on a vertex table needs no separate edge-delete set: the edges
/// vanish with the node. (Edge-only changelog deletes are a future need.)
#[derive(Debug, Clone)]
pub struct VertexDelete {
    pub label: String,
    /// Column whose value identifies the node to delete (same key MERGE used).
    pub id_column: String,
    /// One row per node to delete; only `id_column` is read.
    pub batch: RecordBatch,
}

/// Relationships to upsert: `batch` must hold `src` and `dst` columns with
/// the id values of the endpoint nodes; remaining columns become properties.
#[derive(Debug, Clone)]
pub struct EdgeSet {
    pub src_label: String,
    pub edge_type: String,
    pub dst_label: String,
    /// Id column on the endpoint nodes that `src`/`dst` values refer to.
    pub src_id_column: String,
    pub dst_id_column: String,
    pub batch: RecordBatch,
}

/// Everything one sync round wants to apply to the graph.
///
/// Sinks apply this in three phases, in order:
/// 1. all `vertices` (upserts), then
/// 2. all `edges` (upserts) — so edges can reference nodes created above, and
/// 3. all `deletes` — last, but with a per-key net-effect rule: a delete whose
///    `(label, id)` was also upserted this round is suppressed (the upsert
///    wins). This is what makes an in-place UPDATE safe: a changelog UPDATE
///    emits UPDATE_BEFORE(id) + UPDATE_AFTER(id), and treating the BEFORE row
///    as a blind delete would destroy the node the AFTER row just rewrote. A
///    *true* delete (no matching upsert) still removes the node. See
///    `upserted_vertex_keys` / `surviving_delete_ids` in `sink.rs`.
#[derive(Debug, Clone, Default)]
pub struct GraphMutation {
    pub vertices: Vec<VertexSet>,
    pub edges: Vec<EdgeSet>,
    /// Nodes to remove (DETACH DELETE). Empty for append-only rounds; carries
    /// the `DELETE` rows of a changelog round (UPDATE_BEFORE is dropped, not
    /// deleted). Defaulting to empty keeps every existing producer of
    /// `GraphMutation` source-compatible.
    pub deletes: Vec<VertexDelete>,
}

impl GraphMutation {
    pub fn is_empty(&self) -> bool {
        self.vertices.is_empty() && self.edges.is_empty() && self.deletes.is_empty()
    }

    /// Concatenate another mutation into this one (pipeline steps compose).
    pub fn merge(&mut self, other: GraphMutation) {
        self.vertices.extend(other.vertices);
        self.edges.extend(other.edges);
        self.deletes.extend(other.deletes);
    }
}

/// What a sink did with a mutation.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct SinkReport {
    pub vertices_written: usize,
    pub edges_written: usize,
    /// Nodes removed via `DETACH DELETE` (changelog `DELETE` rows).
    pub deleted: usize,
}