nordisk/knut

crates/knut-bifrost/src/mutation.rs

//! 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. An edge-only table (a `knows(from, to)` join table)
/// deletes its rows' edges through [`GraphMutation::edge_deletes`] instead.
#[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,
    /// Property columns whose values participate in the **relationship** MERGE
    /// key. Empty (the default, and every historical producer) keeps the
    /// original behaviour: `MERGE (a)-[e:type]->(b)` collapses to ONE edge per
    /// `(src, type, dst)`. When non-empty, those properties are folded into the
    /// relationship pattern — `MERGE (a)-[e:type {p: r.props.p}]->(b)` — so
    /// distinct property combinations survive as PARALLEL edges between the
    /// same endpoints (e.g. two `WORKS_AT` edges with different `since`). Each
    /// name must be a property column of `batch` (i.e. not `src`/`dst`).
    #[allow(clippy::doc_markdown)]
    pub edge_key_props: Vec<String>,
    /// **One edge of this type per source node** — the row is the truth: before
    /// the upsert, the source node's other outgoing edges of this type (to any
    /// other target, or all of them when `dst` is null — the foreign key was
    /// cleared) are deleted. This is how a changed foreign key drops its old edge
    /// whatever the changelog carried for the old row. Set by `fk_edges` steps;
    /// ignored when `edge_key_props` is non-empty (parallel edges are the point).
    pub replace_outgoing: bool,
    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>,
    /// Relationships to remove: the changelog `DELETE` rows of an edge step,
    /// in [`EdgeSet`] shape (`src`, `dst`, and the `edge_key_props` that tell
    /// parallel edges apart). Applied in phase 3 with the vertex deletes and by
    /// the same rule — an edge also upserted this round survives — and a row
    /// whose `src` or `dst` is null (an equality delete that carries only the
    /// row's key) cannot name its edge and is skipped, counted in the log.
    pub edge_deletes: Vec<EdgeSet>,
}

impl GraphMutation {
    pub fn is_empty(&self) -> bool {
        self.vertices.is_empty() && self.edges.is_empty() && self.deletes.is_empty() && self.edge_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);
        self.edge_deletes.extend(other.edge_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` and edges removed via `DELETE`
    /// (changelog `DELETE` rows).
    pub deleted: usize,
}