knut_bifrost/
mutation.rs

1//! The common currency between transforms and sinks.
2
3use arrow_array::RecordBatch;
4
5/// Nodes to upsert: one batch per label.
6#[derive(Debug, Clone)]
7pub struct VertexSet {
8    pub label: String,
9    /// Column whose value identifies the node (MERGE key).
10    pub id_column: String,
11    pub batch: RecordBatch,
12}
13
14/// Nodes to remove. The `batch` carries the id column for the label; every
15/// row identifies one node to `MATCH (n:Label {id: x}) DETACH DELETE n`.
16///
17/// Deleting a vertex DETACH-deletes its incident edges too, so a changelog
18/// `DELETE` on a vertex table needs no separate edge-delete set: the edges
19/// vanish with the node. An edge-only table (a `knows(from, to)` join table)
20/// deletes its rows' edges through [`GraphMutation::edge_deletes`] instead.
21#[derive(Debug, Clone)]
22pub struct VertexDelete {
23    pub label: String,
24    /// Column whose value identifies the node to delete (same key MERGE used).
25    pub id_column: String,
26    /// One row per node to delete; only `id_column` is read.
27    pub batch: RecordBatch,
28}
29
30/// Relationships to upsert: `batch` must hold `src` and `dst` columns with
31/// the id values of the endpoint nodes; remaining columns become properties.
32#[derive(Debug, Clone)]
33pub struct EdgeSet {
34    pub src_label: String,
35    pub edge_type: String,
36    pub dst_label: String,
37    /// Id column on the endpoint nodes that `src`/`dst` values refer to.
38    pub src_id_column: String,
39    pub dst_id_column: String,
40    /// Property columns whose values participate in the **relationship** MERGE
41    /// key. Empty (the default, and every historical producer) keeps the
42    /// original behaviour: `MERGE (a)-[e:type]->(b)` collapses to ONE edge per
43    /// `(src, type, dst)`. When non-empty, those properties are folded into the
44    /// relationship pattern — `MERGE (a)-[e:type {p: r.props.p}]->(b)` — so
45    /// distinct property combinations survive as PARALLEL edges between the
46    /// same endpoints (e.g. two `WORKS_AT` edges with different `since`). Each
47    /// name must be a property column of `batch` (i.e. not `src`/`dst`).
48    #[allow(clippy::doc_markdown)]
49    pub edge_key_props: Vec<String>,
50    /// **One edge of this type per source node** — the row is the truth: before
51    /// the upsert, the source node's other outgoing edges of this type (to any
52    /// other target, or all of them when `dst` is null — the foreign key was
53    /// cleared) are deleted. This is how a changed foreign key drops its old edge
54    /// whatever the changelog carried for the old row. Set by `fk_edges` steps;
55    /// ignored when `edge_key_props` is non-empty (parallel edges are the point).
56    pub replace_outgoing: bool,
57    pub batch: RecordBatch,
58}
59
60/// Everything one sync round wants to apply to the graph.
61///
62/// Sinks apply this in three phases, in order:
63/// 1. all `vertices` (upserts), then
64/// 2. all `edges` (upserts) — so edges can reference nodes created above, and
65/// 3. all `deletes` — last, but with a per-key net-effect rule: a delete whose
66///    `(label, id)` was also upserted this round is suppressed (the upsert
67///    wins). This is what makes an in-place UPDATE safe: a changelog UPDATE
68///    emits UPDATE_BEFORE(id) + UPDATE_AFTER(id), and treating the BEFORE row
69///    as a blind delete would destroy the node the AFTER row just rewrote. A
70///    *true* delete (no matching upsert) still removes the node. See
71///    `upserted_vertex_keys` / `surviving_delete_ids` in `sink.rs`.
72#[derive(Debug, Clone, Default)]
73pub struct GraphMutation {
74    pub vertices: Vec<VertexSet>,
75    pub edges: Vec<EdgeSet>,
76    /// Nodes to remove (DETACH DELETE). Empty for append-only rounds; carries
77    /// the `DELETE` rows of a changelog round (UPDATE_BEFORE is dropped, not
78    /// deleted). Defaulting to empty keeps every existing producer of
79    /// `GraphMutation` source-compatible.
80    pub deletes: Vec<VertexDelete>,
81    /// Relationships to remove: the changelog `DELETE` rows of an edge step,
82    /// in [`EdgeSet`] shape (`src`, `dst`, and the `edge_key_props` that tell
83    /// parallel edges apart). Applied in phase 3 with the vertex deletes and by
84    /// the same rule — an edge also upserted this round survives — and a row
85    /// whose `src` or `dst` is null (an equality delete that carries only the
86    /// row's key) cannot name its edge and is skipped, counted in the log.
87    pub edge_deletes: Vec<EdgeSet>,
88}
89
90impl GraphMutation {
91    pub fn is_empty(&self) -> bool {
92        self.vertices.is_empty() && self.edges.is_empty() && self.deletes.is_empty() && self.edge_deletes.is_empty()
93    }
94
95    /// Concatenate another mutation into this one (pipeline steps compose).
96    pub fn merge(&mut self, other: GraphMutation) {
97        self.vertices.extend(other.vertices);
98        self.edges.extend(other.edges);
99        self.deletes.extend(other.deletes);
100        self.edge_deletes.extend(other.edge_deletes);
101    }
102}
103
104/// What a sink did with a mutation.
105#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
106pub struct SinkReport {
107    pub vertices_written: usize,
108    pub edges_written: usize,
109    /// Nodes removed via `DETACH DELETE` and edges removed via `DELETE`
110    /// (changelog `DELETE` rows).
111    pub deleted: usize,
112}
knut_bifrost/mutation.rs

knut_bifrost/
mutation.rs
