crates/knut-bifrost/src/sink/graph.rs
//! **Graph** sinks — the `GraphSink` family. The FalkorDB sink upserts with
//! `MERGE`, keyed on the mutation's id columns, so replaying a round after a
//! crash is idempotent.
//!
//! The **table** family ([`crate::sink::table`]) is a deliberate sibling, not a
//! generalisation of this one: every field of [`GraphMutation`] is a label, an
//! edge type or an id column, and a row-and-column load has none of those. See
//! `.nornir/doris-sink-design.md` f2.
use std::collections::{HashMap, HashSet};
use arrow_array::{Array, RecordBatch};
use async_trait::async_trait;
use graphar_falkordb::{
BulkEdgeGroup, BulkNodeGroup, FalkorDbLoader,
cypher::{arrow_value_to_cypher, row_to_cypher_map},
loader::choose_bulk,
};
use crate::{
error::{BifrostError, Result},
mutation::{EdgeSet, GraphMutation, SinkReport, VertexDelete, VertexSet},
};
#[async_trait]
pub trait GraphSink: Send {
async fn apply(&mut self, mutation: &GraphMutation) -> Result<SinkReport>;
}
/// Upserts a [`GraphMutation`] into FalkorDB.
///
/// **First load in bulk, then MERGE.** A label that is empty in the graph the
/// first time a round writes to it — and whose round deletes nothing — is loaded
/// with a verbatim `GRAPH.BULK` (no Cypher planner, no per-row MERGE): its rows
/// deduplicated on the id (last row wins, as sequential MERGEs would), nulls ids
/// dropped. Every later write to it is the idempotent `UNWIND … MERGE`, so a
/// replay after a crash, an update or a delete is always a MERGE. A bulk that
/// fails for any reason (an Arrow type the bulk wire has no encoding for, a
/// dropped link mid-stream) falls back to MERGE for that label: MERGE on the id
/// finds whatever the bulk did create, so the result is the same graph.
///
/// **Edges too.** The first write of a relationship type that has no edges yet,
/// in a round that deletes nothing, is a `GRAPH.BULK` append: the end nodes'
/// internal ids are looked up once per label (`MATCH (n:L) RETURN n.id, ID(n)`),
/// rows deduplicated on `(src, dst, key)` (last wins). A row whose end node does
/// not exist yet — MERGE would create it as a placeholder — stays on MERGE, as
/// does every row of a type written before. A refused bulk falls back to MERGE,
/// which finds whatever the bulk created.
///
/// **Every MERGE key is indexed**: `(label, id)` of each node set and of each
/// edge endpoint, created once per label (after a bulk load, so the load does
/// not maintain it row by row). Without it every `MERGE (n:L {id: x})` scans
/// the label and a large graph slows down round by round.
pub struct FalkorDbSink {
loader: FalkorDbLoader,
batch_size: usize,
/// `(label, property)` pairs this sink has indexed.
indexed: HashSet<(String, String)>,
/// Labels whose first-load decision is made (bulk-loaded, or found non-empty).
settled: HashSet<String>,
/// Relationship types whose first-load decision is made.
settled_edges: HashSet<String>,
}
impl FalkorDbSink {
pub async fn connect(url: &str, graph: impl Into<String>) -> Result<Self> {
Self::connect_with_identity(url, graph, None).await
}
/// [`connect`](Self::connect) with an optional mutual-TLS client identity, for a
/// `rediss://` FalkorDB that refuses anything else. `None` is the plain path.
pub async fn connect_with_identity(
url: &str,
graph: impl Into<String>,
tls: Option<graphar_falkordb::FalkorTls>,
) -> Result<Self> {
Ok(Self {
loader: FalkorDbLoader::connect_with_identity(url, graph, tls).await?,
batch_size: graphar_falkordb::DEFAULT_BATCH_SIZE,
indexed: HashSet::new(),
settled: HashSet::new(),
settled_edges: HashSet::new(),
})
}
pub fn with_batch_size(mut self, n: usize) -> Self {
self.batch_size = n.max(1);
self
}
/// Index `(label, property)` once per sink — the MERGE key.
async fn ensure_index(&mut self, label: &str, property: &str) -> Result<()> {
let key = (label.to_string(), property.to_string());
if self.indexed.contains(&key) {
return Ok(());
}
self.loader.create_index(label, property).await?;
self.indexed.insert(key);
Ok(())
}
/// The first-load bulk lane: every label of `mutation` seen for the first
/// time, empty in the graph, is bulk-loaded. Returns the labels loaded (their
/// sets are skipped by the MERGE phase) and the nodes created.
async fn first_load_in_bulk(&mut self, mutation: &GraphMutation) -> Result<(HashSet<String>, usize)> {
let mut loaded = HashSet::new();
let mut created = 0;
// A round that deletes is never a first load.
if !mutation.deletes.is_empty() || !mutation.edge_deletes.is_empty() {
return Ok((loaded, created));
}
let mut by_label: Vec<(&str, Vec<&VertexSet>)> = Vec::new();
for set in &mutation.vertices {
match by_label.iter_mut().find(|(l, _)| *l == set.label) {
Some((_, sets)) => sets.push(set),
None => by_label.push((set.label.as_str(), vec![set])),
}
}
for (label, sets) in by_label {
if self.settled.contains(label) {
continue;
}
self.settled.insert(label.to_string());
if !self.loader.label_is_empty(label).await {
continue;
}
let Some(batch) = first_load_rows(&sets)? else { continue };
let n_rows = batch.num_rows();
match self
.loader
.bulk_insert_nodes_verbatim(&[BulkNodeGroup { label, batches: std::slice::from_ref(&batch) }])
.await
{
Ok(n) => {
tracing::info!(
target: "bifrost::sink",
event = "vertices_bulk_loaded",
label = %label,
rows = n_rows,
count = n,
"first load of an empty label: GRAPH.BULK",
);
self.ensure_index(label, &sets[0].id_column).await?;
loaded.insert(label.to_string());
created += n;
}
Err(e) => tracing::warn!(
target: "bifrost::sink",
event = "bulk_fallback",
label = %label,
error = %e,
"GRAPH.BULK refused the first load; MERGE instead (same graph)",
),
}
}
Ok((loaded, created))
}
async fn upsert_vertices(&mut self, set: &VertexSet) -> Result<usize> {
self.ensure_index(&set.label, &set.id_column).await?;
let rows = rows_as_cypher_maps(&set.batch)?;
let mut written = 0;
for chunk in rows.chunks(self.batch_size) {
// MERGE on the id property, then overwrite all properties.
let query = format!(
"UNWIND [{rows}] AS r MERGE (n:`{label}` {{`{id}`: r.`{id}`}}) SET n += r",
rows = chunk.join(", "),
label = set.label,
id = set.id_column,
);
self.loader.run_query(&query).await?;
written += chunk.len();
}
Ok(written)
}
async fn upsert_edges(&mut self, set: &EdgeSet) -> Result<usize> {
self.ensure_index(&set.src_label, &set.src_id_column).await?;
self.ensure_index(&set.dst_label, &set.dst_id_column).await?;
let rows = rows_as_cypher_maps(&set.batch)?;
let mut written = 0;
for chunk in rows.chunks(self.batch_size) {
self.loader
.run_query(&edge_upsert_query(set, chunk))
.await?;
written += chunk.len();
}
Ok(written)
}
/// `DETACH DELETE` every node named by a [`VertexDelete`]. Deleting the
/// node drops its incident edges too, so a changelog `DELETE` on a vertex
/// table cleans up its relationships for free. Idempotent: a `MATCH` that
/// finds nothing (replay after a crash, or a never-ingested id) is a no-op.
async fn delete_vertices(
&mut self,
set: &VertexDelete,
upserted: &HashSet<(String, String)>,
) -> Result<usize> {
let ids = surviving_delete_ids(set, upserted)?;
if !ids.is_empty() {
self.ensure_index(&set.label, &set.id_column).await?;
}
let mut deleted = 0;
for chunk in ids.chunks(self.batch_size) {
self.loader
.run_query(&detach_delete_query(&set.label, &set.id_column, chunk))
.await?;
deleted += chunk.len();
}
Ok(deleted)
}
/// `DELETE` every relationship named by an edge-delete [`EdgeSet`] — the
/// changelog `DELETE` rows of an edge step. Idempotent like the vertex half:
/// a `MATCH` that finds nothing is a no-op. Rows the round also upserted
/// survive, and rows with a null endpoint cannot name an edge and are skipped.
async fn delete_edges(&mut self, set: &EdgeSet, upserted: &HashSet<EdgeKey>, by_key: usize) -> Result<usize> {
let (rows, null_endpoint) = surviving_edge_deletes(set, upserted)?;
let skipped = null_endpoint.saturating_sub(by_key);
if skipped > 0 {
tracing::warn!(
target: "bifrost::sink",
event = "edge_deletes_skipped",
edge_type = %set.edge_type,
skipped,
"edge DELETE rows with a null src/dst and no edge key cannot name their edge",
);
}
if !rows.is_empty() {
self.ensure_index(&set.src_label, &set.src_id_column).await?;
self.ensure_index(&set.dst_label, &set.dst_id_column).await?;
}
let mut deleted = 0;
for chunk in rows.chunks(self.batch_size) {
self.loader.run_query(&edge_delete_query(set, chunk)).await?;
deleted += chunk.len();
}
Ok(deleted)
}
/// The first-load bulk lane for one edge set: `None` when the type was written
/// before (or has edges), else the relationships bulk-created and the rows
/// left for MERGE (end node not there yet, or the whole set if the bulk was
/// refused).
async fn first_load_edges_in_bulk(&mut self, set: &EdgeSet) -> Result<Option<(usize, EdgeSet)>> {
if !self.settled_edges.insert(set.edge_type.clone()) {
return Ok(None);
}
let n_existing = self
.loader
.run_query(&format!("MATCH ()-[e:`{}`]->() RETURN count(e)", set.edge_type))
.await
.ok()
.and_then(|v| reply_rows(&v).first().and_then(|r| r.first().cloned()))
.and_then(|v| if let NodeKey::Int(i) = v { Some(i) } else { None });
if n_existing != Some(0) {
return Ok(None);
}
let set = dedup_edges(set)?;
let (Some(src_keys), Some(dst_keys)) = (node_keys(&set.batch, "src"), node_keys(&set.batch, "dst")) else {
return Ok(None); // an id type the lookup does not key on: MERGE
};
let src_ids = self.node_ids(&set.src_label, &set.src_id_column).await?;
let dst_ids = self.node_ids(&set.dst_label, &set.dst_id_column).await?;
let (mut bulk_rows, mut s_ids, mut d_ids, mut merge_rows) = (Vec::new(), Vec::new(), Vec::new(), Vec::new());
for (row, (s, d)) in src_keys.iter().zip(&dst_keys).enumerate() {
match (s.as_ref().and_then(|k| src_ids.get(k)), d.as_ref().and_then(|k| dst_ids.get(k))) {
(Some(si), Some(di)) => {
bulk_rows.push(row as u32);
s_ids.push(*si);
d_ids.push(*di);
}
// A null endpoint is no edge: dropped by the MERGE renderer too.
_ => merge_rows.push(row as u32),
}
}
if bulk_rows.is_empty() {
return Ok(None);
}
let bulk = with_endpoints(&take_rows(&set.batch, &bulk_rows)?, s_ids, d_ids)?;
let rest = EdgeSet { batch: take_rows(&set.batch, &merge_rows)?, ..set.clone() };
let group = BulkEdgeGroup {
src_label: &set.src_label,
edge_label: &set.edge_type,
dst_label: &set.dst_label,
batches: std::slice::from_ref(&bulk),
};
match self.loader.bulk_insert_edges_absolute(&[group]).await {
Ok(n) => {
tracing::info!(
target: "bifrost::sink",
event = "edges_bulk_loaded",
edge_type = %set.edge_type,
count = n,
left_for_merge = rest.batch.num_rows(),
"first load of an empty relationship type: GRAPH.BULK",
);
for k in &set.edge_key_props {
self.ensure_edge_index(&set.edge_type, k).await?;
}
Ok(Some((n, rest)))
}
Err(e) => {
tracing::warn!(
target: "bifrost::sink",
event = "bulk_fallback",
edge_type = %set.edge_type,
error = %e,
"GRAPH.BULK refused the first edge load; MERGE instead (same graph)",
);
Ok(Some((0, set)))
}
}
}
/// `key -> ID(n)` for every node of `label` that has `property`.
async fn node_ids(&mut self, label: &str, property: &str) -> Result<HashMap<NodeKey, i64>> {
let reply = self
.loader
.run_query(&format!("MATCH (n:`{label}`) WHERE n.`{property}` IS NOT NULL RETURN n.`{property}`, ID(n)"))
.await?;
Ok(reply_rows(&reply)
.into_iter()
.filter_map(|r| match (r.first(), r.get(1)) {
(Some(k), Some(NodeKey::Int(id))) => Some((k.clone(), *id)),
_ => None,
})
.collect())
}
/// Delete the edges named only by their key: the rows of an edge-delete set
/// whose `src` or `dst` is null but whose `edge_key_props` are all set.
/// Indexes the edge key first (a relationship index), once per sink.
async fn delete_edges_by_key(&mut self, set: &EdgeSet) -> Result<usize> {
let rows = key_only_edge_rows(set)?;
if rows.is_empty() {
return Ok(0);
}
for k in &set.edge_key_props {
self.ensure_edge_index(&set.edge_type, k).await?;
}
let mut deleted = 0;
for chunk in rows.chunks(self.batch_size) {
self.loader.run_query(&edge_delete_by_key_query(set, chunk)).await?;
deleted += chunk.len();
}
Ok(deleted)
}
/// Before a replace set's upsert: each source node's outgoing edges of the
/// type to any OTHER target — or all of them, when the row's `dst` is null —
/// are deleted. Returns the rows sent (not the edges removed).
async fn replace_outgoing(&mut self, set: &EdgeSet) -> Result<usize> {
let rows = replace_rows(set)?;
if rows.is_empty() {
return Ok(0);
}
self.ensure_index(&set.src_label, &set.src_id_column).await?;
for chunk in rows.chunks(self.batch_size) {
self.loader.run_query(&replace_outgoing_query(set, chunk)).await?;
}
Ok(0)
}
/// A relationship-property index, once per sink; "already indexed" is success.
async fn ensure_edge_index(&mut self, edge_type: &str, property: &str) -> Result<()> {
let key = (format!("()-[{edge_type}]-()"), property.to_string());
if self.indexed.contains(&key) {
return Ok(());
}
let q = format!("CREATE INDEX FOR ()-[e:`{edge_type}`]-() ON (e.`{property}`)");
match self.loader.run_query(&q).await {
Ok(_) => {}
Err(e) if e.to_string().to_ascii_lowercase().contains("already indexed") => {}
Err(e) => return Err(e.into()),
}
self.indexed.insert(key);
Ok(())
}
/// Whether **every** label in `sets` currently holds zero nodes — the live
/// half of the empty-graph fast-lane gate. If any one label already has
/// nodes, the whole vertex load must take the idempotent MERGE path (a
/// partial bulk `CREATE` onto a populated label would duplicate), so this
/// short-circuits to `false` on the first non-empty label.
async fn labels_all_empty(&mut self, sets: &[VertexSet]) -> bool {
for set in sets {
if !self.loader.label_is_empty(&set.label).await {
return false;
}
}
true
}
/// Land a vertex-only initial load through `graphar-falkordb`'s native
/// `GRAPH.BULK` encoder instead of per-batch `UNWIND … MERGE`. One
/// [`BulkNodeGroup`] per [`VertexSet`] (label + its single batch), no edge
/// groups — bifrost never bulk-loads edges (their `src`/`dst` carry user id
/// values, not the 0-based `_gar_id` positions the bulk edge encoder wants).
/// Delegates entirely to [`FalkorDbLoader::bulk_insert_streamed`] — bifrost
/// re-implements no encoder. Returns nodes created.
///
/// Only reached via [`bulk_lane_eligible`] + [`FalkorDbSink::labels_all_empty`]:
/// the batches all carry `_gar_id`, so the bulk encoder stores it verbatim
/// (no synthesized/duplicate property) and the produced nodes are
/// byte-identical to what the MERGE path writes on an empty label.
async fn bulk_load_vertices(&mut self, sets: &[VertexSet]) -> Result<usize> {
let node_groups: Vec<BulkNodeGroup> = sets
.iter()
.map(|set| BulkNodeGroup {
label: &set.label,
batches: std::slice::from_ref(&set.batch),
})
.collect();
let edge_groups: [BulkEdgeGroup; 0] = [];
let (nodes, _relations) = self
.loader
.bulk_insert_streamed(&node_groups, &edge_groups)
.await?;
Ok(nodes)
}
}
#[async_trait]
impl GraphSink for FalkorDbSink {
async fn apply(&mut self, mutation: &GraphMutation) -> Result<SinkReport> {
let mut report = SinkReport::default();
// Empty-graph GRAPH.BULK fast lane. Mirrors knut-popviz's
// `choose_bulk(graph_is_empty)` auto-select, adapted to bifrost's
// one-label-per-`apply` granularity: a vertex-only initial load into
// fresh (empty) labels ships the native, planner-free `GRAPH.BULK`
// payload instead of per-batch `UNWIND … MERGE`. Checked *before*
// rendering any Cypher maps so a large initial load skips the per-row
// MERGE-map + upserted-key work entirely. Any incremental/edge/delete
// round, or a non-empty target label, keeps the idempotent MERGE path
// below — so replay-after-crash (the label is now populated) falls back
// and stays idempotent.
if bulk_lane_eligible(mutation)
&& choose_bulk(self.labels_all_empty(&mutation.vertices).await)
{
let n = self.bulk_load_vertices(&mutation.vertices).await?;
report.vertices_written += n;
tracing::info!(
target: "bifrost::sink",
event = "vertices_bulk_loaded",
count = n,
"GRAPH.BULK vertices (empty-graph fast lane)",
);
tracing::info!(
target: "bifrost::sink",
event = "sink_applied",
vertices = report.vertices_written,
edges = 0,
deleted = 0,
"graph mutation applied",
);
return Ok(report);
}
// Keys upserted this round; a delete of the same (label, id) must be
// suppressed so an in-place UPDATE (UPDATE_BEFORE + UPDATE_AFTER on a
// stable key) doesn't destroy the node it just rewrote.
let upserted = upserted_vertex_keys(mutation)?;
let (bulk_loaded, created) = self.first_load_in_bulk(mutation).await?;
report.vertices_written += created;
// Vertices strictly before edges: edges MATCH their endpoints.
for set in mutation.vertices.iter().filter(|s| !bulk_loaded.contains(&s.label)) {
let n = self.upsert_vertices(set).await?;
tracing::debug!(
target: "bifrost::sink",
event = "vertices_upserted",
label = %set.label,
count = n,
"MERGE vertices",
);
report.vertices_written += n;
}
// Edge deletes that carry only the edge's key (an equality delete on an
// edge table): no endpoints to MATCH, so by key — and BEFORE the upserts,
// so an edge whose endpoints changed is replaced, not lost.
let mut by_key = vec![0usize; mutation.edge_deletes.len()];
for (i, set) in mutation.edge_deletes.iter().enumerate() {
by_key[i] = self.delete_edges_by_key(set).await?;
report.deleted += by_key[i];
}
let deletes_nothing = mutation.deletes.is_empty() && mutation.edge_deletes.is_empty();
for set in &mutation.edges {
// One edge per source node: the row is the truth. Last row per
// source, and the source's other outgoing edges of this type go.
let replaced;
let set = if set.replace_outgoing && set.edge_key_props.is_empty() {
replaced = last_row_per_src(set)?;
report.deleted += self.replace_outgoing(&replaced).await?;
&replaced
} else {
set
};
// The first load of an empty relationship type: bulk what resolves,
// MERGE the rest.
let rest;
let set = if deletes_nothing {
match self.first_load_edges_in_bulk(set).await? {
Some((n, remaining)) => {
report.edges_written += n;
rest = remaining;
&rest
}
None => set,
}
} else {
set
};
if set.batch.num_rows() == 0 {
continue;
}
let n = self.upsert_edges(set).await?;
tracing::debug!(
target: "bifrost::sink",
event = "edges_upserted",
edge_type = %set.edge_type,
count = n,
"MERGE edges",
);
report.edges_written += n;
}
// Deletes last: an upsert + delete of the same key in one changelog
// round must resolve to "gone" (see GraphMutation phase ordering).
// Edges first, while their endpoints still exist to MATCH on.
if !mutation.edge_deletes.is_empty() {
let upserted_edges = upserted_edge_keys(mutation)?;
for (i, set) in mutation.edge_deletes.iter().enumerate() {
let n = self.delete_edges(set, &upserted_edges, by_key[i]).await?;
tracing::debug!(
target: "bifrost::sink",
event = "edges_deleted",
edge_type = %set.edge_type,
count = n,
"DELETE edges",
);
report.deleted += n;
}
}
for set in &mutation.deletes {
let n = self.delete_vertices(set, &upserted).await?;
tracing::debug!(
target: "bifrost::sink",
event = "vertices_deleted",
label = %set.label,
count = n,
"DETACH DELETE vertices",
);
report.deleted += n;
}
tracing::info!(
target: "bifrost::sink",
event = "sink_applied",
vertices = report.vertices_written,
edges = report.edges_written,
deleted = report.deleted,
"graph mutation applied",
);
Ok(report)
}
}
/// Build the idempotent edge `UNWIND … MERGE` query for one chunk of
/// pre-rendered `{src, dst, props}` row maps. Pure (no IO) so the query shape —
/// crucially the parallel-edge merge key — is unit-testable without a database.
///
/// MERGEs (not MATCHes) the endpoints: tables sync in arbitrary order, so an
/// edge may arrive before its endpoint's vertex table; the placeholder node is
/// enriched later by the vertex MERGE on the same id key.
///
/// When `set.edge_key_props` is empty the relationship pattern is the historical
/// `-[e:`type`]->` (one edge per `(src, type, dst)`). When non-empty those
/// property values are folded INTO the relationship MERGE key —
/// `-[e:`type` {`p`: r.props.`p`}]->` — so distinct property combinations
/// create PARALLEL edges instead of collapsing. `SET e += r.props` still writes
/// every property (the merge-key props included) so the edge carries them.
fn edge_upsert_query(set: &EdgeSet, rows: &[String]) -> String {
let rel_key = edge_rel_key(set);
format!(
"UNWIND [{rows}] AS r \
MERGE (a:`{src_label}` {{`{src_id}`: r.src}}) \
MERGE (b:`{dst_label}` {{`{dst_id}`: r.dst}}) \
MERGE (a)-[e:`{etype}`{rel_key}]->(b) \
SET e += r.props",
rows = rows.join(", "),
src_label = set.src_label,
src_id = set.src_id_column,
dst_label = set.dst_label,
dst_id = set.dst_id_column,
etype = set.edge_type,
)
}
/// One label's rows for a first bulk load: its sets concatenated, rows with a
/// null id dropped, one row per id (the last, as sequential MERGEs keep the last
/// write). `None` when the sets differ in shape (two steps writing one label with
/// different columns) or nothing is left — MERGE handles those.
fn first_load_rows(sets: &[&VertexSet]) -> Result<Option<RecordBatch>> {
let schema = sets[0].batch.schema();
if sets.iter().any(|s| s.batch.schema() != schema || s.id_column != sets[0].id_column) {
return Ok(None);
}
let batch = arrow::compute::concat_batches(&schema, sets.iter().map(|s| &s.batch))?;
let idx = schema
.index_of(&sets[0].id_column)
.map_err(|_| BifrostError::ColumnNotFound(sets[0].id_column.clone()))?;
let ids = batch.column(idx).as_ref();
let mut last: HashMap<String, u32> = HashMap::with_capacity(batch.num_rows());
let mut order: Vec<String> = Vec::with_capacity(batch.num_rows());
for row in 0..batch.num_rows() {
if ids.is_null(row) {
continue;
}
let key = arrow_value_to_cypher(ids, row).map_err(BifrostError::Falkor)?;
if last.insert(key.clone(), row as u32).is_none() {
order.push(key);
}
}
if order.is_empty() {
return Ok(None);
}
let keep = arrow_array::UInt32Array::from(order.iter().map(|k| last[k]).collect::<Vec<_>>());
let cols = batch
.columns()
.iter()
.map(|c| arrow::compute::take(c.as_ref(), &keep, None))
.collect::<std::result::Result<Vec<_>, _>>()?;
Ok(Some(RecordBatch::try_new(schema, cols)?))
}
/// The relationship pattern's property map: ` {`k`: r.props.`k`, …}` for the
/// `edge_key_props`, empty for one edge per `(src, type, dst)`.
fn edge_rel_key(set: &EdgeSet) -> String {
if set.edge_key_props.is_empty() {
return String::new();
}
let parts: Vec<String> = set
.edge_key_props
.iter()
.map(|p| format!("`{p}`: r.props.`{p}`"))
.collect();
format!(" {{{}}}", parts.join(", "))
}
/// The `DELETE` query for one chunk of `{src, dst, props}` edge rows: the
/// relationship between those endpoints, of that type, with those key props.
/// Pure (no IO) so the query shape is unit-testable without a database.
fn edge_delete_query(set: &EdgeSet, rows: &[String]) -> String {
format!(
"UNWIND [{rows}] AS r \
MATCH (a:`{src_label}` {{`{src_id}`: r.src}})-[e:`{etype}`{rel_key}]->(b:`{dst_label}` {{`{dst_id}`: r.dst}}) \
DELETE e",
rows = rows.join(", "),
src_label = set.src_label,
src_id = set.src_id_column,
dst_label = set.dst_label,
dst_id = set.dst_id_column,
etype = set.edge_type,
rel_key = edge_rel_key(set),
)
}
/// A node key as both sides see it — an Arrow id value and a FalkorDB reply value.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
enum NodeKey {
Int(i64),
Str(String),
}
/// A `GRAPH.QUERY` reply's rows (`[header, rows, stats]`), each value as a key.
fn reply_rows(v: &redis::Value) -> Vec<Vec<NodeKey>> {
let redis::Value::Array(parts) = v else { return Vec::new() };
let Some(redis::Value::Array(rows)) = parts.get(1) else { return Vec::new() };
rows.iter()
.filter_map(|r| match r {
redis::Value::Array(cells) => Some(
cells
.iter()
.filter_map(|c| match c {
redis::Value::Int(i) => Some(NodeKey::Int(*i)),
redis::Value::BulkString(b) => Some(NodeKey::Str(String::from_utf8_lossy(b).into_owned())),
redis::Value::SimpleString(s) => Some(NodeKey::Str(s.clone())),
_ => None,
})
.collect(),
),
_ => None,
})
.collect()
}
/// A `src`/`dst` column's values as node keys (`None` for a null); `None` for an
/// id type the lookup does not key on (it is MERGEd instead).
fn node_keys(batch: &RecordBatch, column: &str) -> Option<Vec<Option<NodeKey>>> {
use arrow_array::cast::AsArray;
let col = batch.column_by_name(column)?;
match col.data_type() {
arrow_schema::DataType::Int64 => {
let a = col.as_primitive::<arrow_array::types::Int64Type>();
Some((0..a.len()).map(|i| (!a.is_null(i)).then(|| NodeKey::Int(a.value(i)))).collect())
}
arrow_schema::DataType::Int32 => {
let a = col.as_primitive::<arrow_array::types::Int32Type>();
Some((0..a.len()).map(|i| (!a.is_null(i)).then(|| NodeKey::Int(a.value(i) as i64))).collect())
}
arrow_schema::DataType::Utf8 => {
let a = col.as_string::<i32>();
Some((0..a.len()).map(|i| (!a.is_null(i)).then(|| NodeKey::Str(a.value(i).to_string()))).collect())
}
_ => None,
}
}
/// One row per `(src, dst, key props)`, the last — as sequential MERGEs keep.
fn dedup_edges(set: &EdgeSet) -> Result<EdgeSet> {
let (rows, _) = edge_rows_with_keys(set)?;
if rows.len() == set.batch.num_rows() && {
let mut seen = HashSet::new();
rows.iter().all(|(_, k)| seen.insert(k.clone()))
} {
return Ok(set.clone());
}
// Recompute over the rows WITH endpoints, remembering row indices.
let batch = &set.batch;
let (src, dst) = (batch.column(batch.schema().index_of("src")?), batch.column(batch.schema().index_of("dst")?));
let key_cols: Vec<usize> = set.edge_key_props.iter().map(|k| batch.schema().index_of(k)).collect::<std::result::Result<_, _>>()?;
let lit = |c: &dyn Array, r: usize| arrow_value_to_cypher(c, r).map_err(BifrostError::Falkor);
let mut last: HashMap<String, u32> = HashMap::new();
let mut order: Vec<String> = Vec::new();
for r in 0..batch.num_rows() {
let mut k = format!("{}|{}", lit(src.as_ref(), r)?, lit(dst.as_ref(), r)?);
for c in &key_cols {
k.push('|');
k.push_str(&lit(batch.column(*c).as_ref(), r)?);
}
if last.insert(k.clone(), r as u32).is_none() {
order.push(k);
}
}
let keep: Vec<u32> = order.iter().map(|k| last[k]).collect();
Ok(EdgeSet { batch: take_rows(batch, &keep)?, ..set.clone() })
}
fn take_rows(batch: &RecordBatch, rows: &[u32]) -> Result<RecordBatch> {
let idx = arrow_array::UInt32Array::from(rows.to_vec());
let cols = batch
.columns()
.iter()
.map(|c| arrow::compute::take(c.as_ref(), &idx, None))
.collect::<std::result::Result<Vec<_>, _>>()?;
Ok(RecordBatch::try_new(batch.schema(), cols)?)
}
/// `batch` with its `src`/`dst` replaced by internal node ids (Int64).
fn with_endpoints(batch: &RecordBatch, src: Vec<i64>, dst: Vec<i64>) -> Result<RecordBatch> {
let schema = batch.schema();
let mut fields = Vec::with_capacity(schema.fields().len());
let mut cols: Vec<arrow_array::ArrayRef> = Vec::with_capacity(schema.fields().len());
let (mut src, mut dst) = (Some(src), Some(dst));
for (i, f) in schema.fields().iter().enumerate() {
match f.name().as_str() {
"src" => {
fields.push(arrow_schema::Field::new("src", arrow_schema::DataType::Int64, false));
cols.push(std::sync::Arc::new(arrow_array::Int64Array::from(src.take().unwrap_or_default())));
}
"dst" => {
fields.push(arrow_schema::Field::new("dst", arrow_schema::DataType::Int64, false));
cols.push(std::sync::Arc::new(arrow_array::Int64Array::from(dst.take().unwrap_or_default())));
}
_ => {
fields.push(f.as_ref().clone());
cols.push(batch.column(i).clone());
}
}
}
Ok(RecordBatch::try_new(std::sync::Arc::new(arrow_schema::Schema::new(fields)), cols)?)
}
/// The rows of an edge-delete set that only their key can name — `src` or
/// `dst` null, every `edge_key_props` value set — as `{props: {…}}` maps.
fn key_only_edge_rows(set: &EdgeSet) -> Result<Vec<String>> {
if set.edge_key_props.is_empty() {
return Ok(Vec::new());
}
let batch = &set.batch;
let schema = batch.schema();
let col = |name: &str| schema.index_of(name).map_err(|_| BifrostError::ColumnNotFound(name.to_string()));
let (src, dst) = (col("src")?, col("dst")?);
let keys: Vec<(&str, usize)> = set.edge_key_props.iter().map(|k| Ok((k.as_str(), col(k)?))).collect::<Result<_>>()?;
let mut out = Vec::new();
for row in 0..batch.num_rows() {
let endpoints = !batch.column(src).is_null(row) && !batch.column(dst).is_null(row);
if endpoints || keys.iter().any(|(_, i)| batch.column(*i).is_null(row)) {
continue;
}
let props: Vec<String> = keys
.iter()
.map(|(k, i)| Ok(format!("`{k}`: {}", arrow_value_to_cypher(batch.column(*i).as_ref(), row).map_err(BifrostError::Falkor)?)))
.collect::<Result<_>>()?;
out.push(format!("{{props: {{{}}}}}", props.join(", ")));
}
Ok(out)
}
/// `DELETE` by edge key alone: the edge of that type, between those labels,
/// with those key props — wherever its endpoints are.
fn edge_delete_by_key_query(set: &EdgeSet, rows: &[String]) -> String {
format!(
"UNWIND [{rows}] AS r MATCH (:`{src_label}`)-[e:`{etype}`{rel_key}]->(:`{dst_label}`) DELETE e",
rows = rows.join(", "),
src_label = set.src_label,
dst_label = set.dst_label,
etype = set.edge_type,
rel_key = edge_rel_key(set),
)
}
/// A replace set with one row per source: the last (as sequential upserts keep
/// the last write). Rows with a null `src` are dropped; a null `dst` is kept — it
/// clears the source's edges.
fn last_row_per_src(set: &EdgeSet) -> Result<EdgeSet> {
let batch = &set.batch;
let src = batch.column(batch.schema().index_of("src").map_err(|_| BifrostError::ColumnNotFound("src".into()))?);
let mut last: HashMap<String, u32> = HashMap::with_capacity(batch.num_rows());
let mut order: Vec<String> = Vec::with_capacity(batch.num_rows());
for row in 0..batch.num_rows() {
if src.is_null(row) {
continue;
}
let k = arrow_value_to_cypher(src.as_ref(), row).map_err(BifrostError::Falkor)?;
if last.insert(k.clone(), row as u32).is_none() {
order.push(k);
}
}
let keep = arrow_array::UInt32Array::from(order.iter().map(|k| last[k]).collect::<Vec<_>>());
let cols = batch
.columns()
.iter()
.map(|c| arrow::compute::take(c.as_ref(), &keep, None))
.collect::<std::result::Result<Vec<_>, _>>()?;
Ok(EdgeSet { batch: RecordBatch::try_new(batch.schema(), cols)?, ..set.clone() })
}
/// `{src, dst}` per row of a replace set; `dst` may be `null`.
fn replace_rows(set: &EdgeSet) -> Result<Vec<String>> {
let batch = &set.batch;
let schema = batch.schema();
let col = |name: &str| schema.index_of(name).map_err(|_| BifrostError::ColumnNotFound(name.to_string()));
let (src, dst) = (batch.column(col("src")?).as_ref(), batch.column(col("dst")?).as_ref());
(0..batch.num_rows())
.filter(|r| !src.is_null(*r))
.map(|r| {
Ok(format!(
"{{src: {}, dst: {}}}",
arrow_value_to_cypher(src, r).map_err(BifrostError::Falkor)?,
arrow_value_to_cypher(dst, r).map_err(BifrostError::Falkor)?
))
})
.collect()
}
/// Delete each source's outgoing edges of the type to any other target (all of
/// them for a null `dst`) — the MERGE that follows writes the one that stays.
fn replace_outgoing_query(set: &EdgeSet, rows: &[String]) -> String {
format!(
"UNWIND [{rows}] AS r \
MATCH (a:`{src_label}` {{`{src_id}`: r.src}})-[e:`{etype}`]->(b:`{dst_label}`) \
WHERE r.dst IS NULL OR b.`{dst_id}` <> r.dst DELETE e",
rows = rows.join(", "),
src_label = set.src_label,
src_id = set.src_id_column,
dst_label = set.dst_label,
dst_id = set.dst_id_column,
etype = set.edge_type,
)
}
/// One relationship's identity: `(src label, type, dst label, src, dst, key props)`
/// as Cypher literals.
type EdgeKey = (String, String, String, String, String, String);
/// Every row of an edge set as `(row map, key)`, the row map holding only what
/// names the edge — `src`, `dst` and the key props. Rows with a null endpoint
/// are counted, not returned.
fn edge_rows_with_keys(set: &EdgeSet) -> Result<(Vec<(String, EdgeKey)>, usize)> {
let batch = &set.batch;
let schema = batch.schema();
let col = |name: &str| schema.index_of(name).map_err(|_| BifrostError::ColumnNotFound(name.to_string()));
let (src_idx, dst_idx) = (col("src")?, col("dst")?);
let key_idx: Vec<(&str, usize)> =
set.edge_key_props.iter().map(|k| Ok((k.as_str(), col(k)?))).collect::<Result<_>>()?;
let lit = |idx: usize, row: usize| arrow_value_to_cypher(batch.column(idx).as_ref(), row).map_err(BifrostError::Falkor);
let mut out = Vec::with_capacity(batch.num_rows());
let mut skipped = 0;
for row in 0..batch.num_rows() {
if batch.column(src_idx).is_null(row) || batch.column(dst_idx).is_null(row) {
skipped += 1;
continue;
}
let (src, dst) = (lit(src_idx, row)?, lit(dst_idx, row)?);
let props: Vec<String> = key_idx.iter().map(|(k, i)| Ok(format!("`{k}`: {}", lit(*i, row)?))).collect::<Result<_>>()?;
let props = props.join(", ");
let key = (
set.src_label.clone(),
set.edge_type.clone(),
set.dst_label.clone(),
src.clone(),
dst.clone(),
props.clone(),
);
out.push((format!("{{src: {src}, dst: {dst}, props: {{{props}}}}}"), key));
}
Ok((out, skipped))
}
/// The identity of every edge this round upserts — an edge delete of the same
/// identity is suppressed, as a vertex delete is by [`upserted_vertex_keys`].
fn upserted_edge_keys(mutation: &GraphMutation) -> Result<HashSet<EdgeKey>> {
let mut keys = HashSet::new();
for set in &mutation.edges {
keys.extend(edge_rows_with_keys(set)?.0.into_iter().map(|(_, k)| k));
}
Ok(keys)
}
/// The rows of an edge-delete set that are actually deleted, and how many were
/// skipped for a null endpoint. Pure so the rule is unit-testable.
fn surviving_edge_deletes(set: &EdgeSet, upserted: &HashSet<EdgeKey>) -> Result<(Vec<String>, usize)> {
let (rows, skipped) = edge_rows_with_keys(set)?;
Ok((rows.into_iter().filter(|(_, k)| !upserted.contains(k)).map(|(r, _)| r).collect(), skipped))
}
/// Build the `DETACH DELETE` query for one chunk of pre-rendered id literals.
/// Pure (no IO) so the query shape is unit-testable without a database.
fn detach_delete_query(label: &str, id_column: &str, ids: &[String]) -> String {
format!(
"UNWIND [{ids}] AS r \
MATCH (n:`{label}` {{`{id}`: r}}) DETACH DELETE n",
ids = ids.join(", "),
id = id_column,
)
}
/// Render the id column of a [`VertexDelete`] batch as a list of Cypher scalar
/// literals — one per node to delete. Only `id_column` is read; other columns
/// (e.g. a `DELETE` row's stale property values) are ignored.
fn delete_ids_as_cypher(set: &VertexDelete) -> Result<Vec<String>> {
let schema = set.batch.schema();
let idx = schema
.index_of(&set.id_column)
.map_err(|_| BifrostError::ColumnNotFound(set.id_column.clone()))?;
let col = set.batch.column(idx).as_ref();
let mut ids = Vec::with_capacity(set.batch.num_rows());
for row in 0..set.batch.num_rows() {
ids.push(arrow_value_to_cypher(col, row).map_err(BifrostError::Falkor)?);
}
Ok(ids)
}
/// Collect the `(label, id-literal)` of every vertex this round upserts.
///
/// An in-place UPDATE on a stable primary key emits `UPDATE_BEFORE(id=X)` +
/// `UPDATE_AFTER(id=X)`; CDC routing sends the BEFORE row to `deletes` and the
/// AFTER row to `vertices`. Applying all deletes last would then DETACH-DELETE
/// the node the upsert just wrote. By recording upserted keys we can let the
/// upsert win over a same-key delete in the same round. (The same window can
/// arise in the pure-Rust skade path: an equality-delete of the old row + an
/// append of the new row for one id.)
fn upserted_vertex_keys(mutation: &GraphMutation) -> Result<HashSet<(String, String)>> {
let mut keys = HashSet::new();
for set in &mutation.vertices {
let schema = set.batch.schema();
let Ok(idx) = schema.index_of(&set.id_column) else {
continue;
};
let col = set.batch.column(idx).as_ref();
for row in 0..set.batch.num_rows() {
let lit = arrow_value_to_cypher(col, row).map_err(BifrostError::Falkor)?;
keys.insert((set.label.clone(), lit));
}
}
Ok(keys)
}
/// Ids of a [`VertexDelete`] that should actually be deleted: every id *except*
/// those whose `(label, id)` was upserted this round (the upsert wins). Pure
/// (no IO) so the net-effect rule is unit-testable without a database.
fn surviving_delete_ids(
set: &VertexDelete,
upserted: &HashSet<(String, String)>,
) -> Result<Vec<String>> {
Ok(delete_ids_as_cypher(set)?
.into_iter()
.filter(|id| !upserted.contains(&(set.label.clone(), id.clone())))
.collect())
}
/// Whether this mutation is **structurally** eligible for the empty-graph
/// `GRAPH.BULK` vertex fast lane (the DB-free half of the gate; the live half is
/// [`FalkorDbSink::labels_all_empty`]). Pure so the selection predicate is
/// unit-testable without a database.
///
/// Eligible iff the mutation is a **vertex-only** load (no edges, no deletes —
/// bulk `CREATE`s a fresh graph and never MERGEs, matches endpoints, or deletes)
/// with at least one vertex set, AND:
///
/// * **every** vertex batch already carries a `_gar_id` column. The bulk encoder
/// always writes a `_gar_id` property (synthesizing one when absent), while the
/// MERGE path writes only the batch's columns — so byte-identity holds only
/// when the batch supplies `_gar_id` (bulk then stores it verbatim, no extra
/// property). Any batch lacking it falls back to MERGE.
/// * no label repeats across vertex sets — `build_bulk_payload` rejects a label
/// in two node groups, so a duplicate must take the MERGE path instead.
///
/// RED-when-broken: relax the `_gar_id` requirement and a bulk-loaded node would
/// carry a property MERGE never wrote; drop the edge/delete guard and bulk would
/// silently ignore relationships/removals.
fn bulk_lane_eligible(mutation: &GraphMutation) -> bool {
if mutation.vertices.is_empty()
|| !mutation.edges.is_empty()
|| !mutation.deletes.is_empty()
|| !mutation.edge_deletes.is_empty()
{
return false;
}
let mut labels = HashSet::new();
for set in &mutation.vertices {
if set.batch.schema_ref().column_with_name("_gar_id").is_none() {
return false;
}
if !labels.insert(set.label.as_str()) {
return false;
}
}
true
}
/// Render each row as a Cypher map literal.
///
/// Vertex batches become `{col: val, …}`. Edge batches become
/// `{src: …, dst: …, props: {…}}` so the property overwrite can't clobber
/// the endpoint keys.
///
/// **Null endpoints are "no edge".** An edge row whose `src` or `dst` value is
/// NULL is a nullable FK with no target — standard SQL semantics — so it is
/// SKIPPED, never rendered. Emitting it would `MERGE (b {id: null})`, minting a
/// spurious node keyed on `null` and a dangling edge into it (the org-root of a
/// self-referential table, or any optional FK, would otherwise leak such a
/// node). This is exactly what the offline correctness oracle counts as
/// `null_endpoint_rows` and excludes from the graph, so dropping the row here
/// keeps the LIVE sink byte-consistent with that oracle. The returned `Vec` is
/// therefore the *rendered* (non-null) edge rows — the caller counts it, so
/// `edges_written` reflects real relationships only.
pub(crate) fn rows_as_cypher_maps(batch: &RecordBatch) -> Result<Vec<String>> {
let schema = batch.schema();
let edge_endpoints = match (schema.index_of("src"), schema.index_of("dst")) {
(Ok(s), Ok(d)) => Some((s, d)),
_ => None,
};
let cols: Vec<(&str, &dyn Array)> = schema
.fields()
.iter()
.zip(batch.columns())
.map(|(f, c)| (f.name().as_str(), c.as_ref()))
.collect();
let mut rows = Vec::with_capacity(batch.num_rows());
for row in 0..batch.num_rows() {
if let Some((src_idx, dst_idx)) = edge_endpoints {
// Null FK = no edge: skip the row rather than MERGE a null-keyed node.
if batch.column(src_idx).is_null(row) || batch.column(dst_idx).is_null(row) {
continue;
}
let mut src = None;
let mut dst = None;
let mut props = Vec::new();
for (name, array) in &cols {
let val = arrow_value_to_cypher(*array, row).map_err(BifrostError::Falkor)?;
match *name {
"src" => src = Some(val),
"dst" => dst = Some(val),
_ => props.push(format!("`{name}`: {val}")),
}
}
rows.push(format!(
"{{src: {}, dst: {}, props: {{{}}}}}",
src.expect("src checked above"),
dst.expect("dst checked above"),
props.join(", "),
));
} else {
rows.push(row_to_cypher_map(&cols, row).map_err(BifrostError::Falkor)?);
}
}
Ok(rows)
}
#[cfg(test)]
mod tests {
use super::*;
use arrow_array::{Int64Array, StringArray};
use arrow_schema::{DataType, Field, Schema};
use std::sync::Arc;
#[test]
fn vertex_rows_render_as_flat_maps() {
let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("user_id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(vec![7])),
Arc::new(StringArray::from(vec!["alice"])),
],
)
.unwrap();
let rows = rows_as_cypher_maps(&batch).unwrap();
assert_eq!(rows, vec!["{`user_id`: 7, `name`: 'alice'}"]);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"sink::vertex_rows_render_as_flat_maps",
rows == vec!["{`user_id`: 7, `name`: 'alice'}"],
&format!("cypher={}", rows[0]),
);
}
#[test]
fn edge_rows_separate_endpoints_from_props() {
let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, false),
Field::new("since", DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(vec![1])),
Arc::new(Int64Array::from(vec![2])),
Arc::new(StringArray::from(vec!["2024"])),
],
)
.unwrap();
let rows = rows_as_cypher_maps(&batch).unwrap();
assert_eq!(rows, vec!["{src: 1, dst: 2, props: {`since`: '2024'}}"]);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"sink::edge_rows_separate_endpoints_from_props",
rows == vec!["{src: 1, dst: 2, props: {`since`: '2024'}}"],
&format!("cypher={}", rows[0]),
);
}
/// **Null FK = no edge (the live-path invariant).** An edge batch whose
/// `dst` (or `src`) endpoint is NULL — a nullable FK with no target, e.g. a
/// self-referential table's org root — must be DROPPED, not rendered. If it
/// were rendered the sink would `MERGE (b {id: null})`, minting a spurious
/// node keyed on `null` and a dangling edge into it. This mirrors the
/// offline correctness oracle, which counts such rows as `null_endpoint_rows`
/// and excludes them from the graph. RED-when-broken: stop skipping and the
/// live graph gains a null-keyed node the oracle never had.
#[test]
fn edge_rows_with_null_endpoint_are_dropped() {
let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, true), // nullable FK
Field::new("since", DataType::Utf8, true),
])),
vec![
Arc::new(Int64Array::from(vec![1, 2, 3])),
Arc::new(Int64Array::from(vec![Some(10), None, Some(30)])),
Arc::new(StringArray::from(vec![Some("a"), None, Some("c")])),
],
)
.unwrap();
let rows = rows_as_cypher_maps(&batch).unwrap();
// Only rows 0 and 2 survive; the null-dst row 1 is "no edge".
assert_eq!(
rows,
vec![
"{src: 1, dst: 10, props: {`since`: 'a'}}",
"{src: 3, dst: 30, props: {`since`: 'c'}}",
],
"null-endpoint edge row dropped (null FK = no edge)"
);
// A null SRC is dropped symmetrically.
let src_null = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, true),
Field::new("dst", DataType::Int64, false),
])),
vec![
Arc::new(Int64Array::from(vec![None, Some(5)])),
Arc::new(Int64Array::from(vec![9, 6])),
],
)
.unwrap();
assert_eq!(
rows_as_cypher_maps(&src_null).unwrap(),
vec!["{src: 5, dst: 6, props: {}}"],
"null-src edge row dropped too"
);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"sink::edge_rows_with_null_endpoint_are_dropped",
rows.len() == 2,
&format!("kept={} of 3 edge rows (null FK = no edge)", rows.len()),
);
}
#[test]
fn detach_delete_query_matches_on_id_and_detaches() {
let q = detach_delete_query("Person", "user_id", &["7".into(), "8".into()]);
assert_eq!(
q,
"UNWIND [7, 8] AS r MATCH (n:`Person` {`user_id`: r}) DETACH DELETE n"
);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"sink::detach_delete_query_matches_on_id_and_detaches",
q == "UNWIND [7, 8] AS r MATCH (n:`Person` {`user_id`: r}) DETACH DELETE n",
&format!("query={q}"),
);
}
/// A helper edge set with the given merge-key props (batch shape is
/// irrelevant to the query builder — it renders `r.src`/`r.dst`/`r.props`).
fn eset(edge_key_props: Vec<String>) -> EdgeSet {
EdgeSet {
src_label: "User".into(),
edge_type: "WORKS_AT".into(),
dst_label: "Company".into(),
src_id_column: "user_id".into(),
dst_id_column: "company_id".into(),
edge_key_props,
replace_outgoing: false,
batch: RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, false),
])),
vec![
Arc::new(Int64Array::from(vec![1])),
Arc::new(Int64Array::from(vec![2])),
],
)
.unwrap(),
}
}
/// Parallel / property-keyed edges. With NO merge-key props the relationship
/// pattern is the historical `-[e:`WORKS_AT`]->` (one edge per endpoint
/// pair). With merge-key props the values are folded INTO the relationship
/// MERGE key so distinct property combinations survive as parallel edges —
/// and `SET e += r.props` still writes them. RED-when-broken: drop the
/// `edge_key_props` handling and parallels silently collapse again.
#[test]
fn edge_merge_key_controls_parallel_edges() {
let plain = edge_upsert_query(&eset(vec![]), &["{src: 1, dst: 2, props: {}}".into()]);
assert_eq!(
plain,
"UNWIND [{src: 1, dst: 2, props: {}}] AS r \
MERGE (a:`User` {`user_id`: r.src}) \
MERGE (b:`Company` {`company_id`: r.dst}) \
MERGE (a)-[e:`WORKS_AT`]->(b) \
SET e += r.props",
"no merge-key props ⇒ byte-identical to the historical collapsing MERGE",
);
let keyed = edge_upsert_query(
&eset(vec!["since".into(), "role".into()]),
&["{src: 1, dst: 2, props: {`since`: '2021', `role`: 'eng'}}".into()],
);
assert!(
keyed.contains(
"MERGE (a)-[e:`WORKS_AT` {`since`: r.props.`since`, `role`: r.props.`role`}]->(b)"
),
"merge-key props fold into the relationship pattern: {keyed}"
);
assert!(
keyed.ends_with("SET e += r.props"),
"props still written: {keyed}"
);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"sink::edge_merge_key_controls_parallel_edges",
plain.contains("-[e:`WORKS_AT`]->")
&& keyed.contains(
"-[e:`WORKS_AT` {`since`: r.props.`since`, `role`: r.props.`role`}]->",
),
"empty ⇒ collapse; keyed ⇒ parallel-edge merge key",
);
}
#[test]
fn delete_ids_read_only_the_id_column() {
// A DELETE/UPDATE_BEFORE row carries stale props; only the id is used.
let set = VertexDelete {
label: "Person".into(),
id_column: "user_id".into(),
batch: RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("user_id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(vec![7, 9])),
Arc::new(StringArray::from(vec!["stale", "gone"])),
],
)
.unwrap(),
};
let ids = delete_ids_as_cypher(&set).unwrap();
assert_eq!(ids, vec!["7", "9"]);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"sink::delete_ids_read_only_the_id_column",
ids == vec!["7", "9"],
&format!("ids={ids:?} (stale props ignored)"),
);
}
/// Helper: a vertex batch with (optionally) a `_gar_id` column.
fn vset(label: &str, with_gar: bool) -> VertexSet {
let (schema, cols): (Schema, Vec<Arc<dyn Array>>) = if with_gar {
(
Schema::new(vec![
Field::new("_gar_id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
]),
vec![
Arc::new(Int64Array::from(vec![0, 1])),
Arc::new(StringArray::from(vec!["a", "b"])),
],
)
} else {
(
Schema::new(vec![
Field::new("user_id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
]),
vec![
Arc::new(Int64Array::from(vec![0, 1])),
Arc::new(StringArray::from(vec!["a", "b"])),
],
)
};
VertexSet {
label: label.into(),
id_column: if with_gar {
"_gar_id".into()
} else {
"user_id".into()
},
batch: RecordBatch::try_new(Arc::new(schema), cols).unwrap(),
}
}
/// The DB-free half of the empty-graph fast-lane gate. A vertex-only load
/// whose batches all carry `_gar_id` and whose labels are unique is eligible;
/// every other shape (edges, deletes, a missing `_gar_id`, a duplicate label,
/// or no vertices) must fall back to the idempotent MERGE path. This is the
/// byte-identity + safety predicate — RED-when-broken on any relaxation.
#[test]
fn bulk_lane_eligibility_predicate() {
// Vertex-only, `_gar_id` present, unique labels → eligible.
let ok = GraphMutation {
edge_deletes: Vec::new(),
vertices: vec![vset("Person", true), vset("Loan", true)],
edges: vec![],
deletes: vec![],
};
assert!(
bulk_lane_eligible(&ok),
"the demo's vertex-only `_gar_id` load qualifies"
);
// No vertices → nothing to bulk.
assert!(!bulk_lane_eligible(&GraphMutation::default()));
// A batch without `_gar_id` would gain a property MERGE never wrote.
let no_gar = GraphMutation {
edge_deletes: Vec::new(),
vertices: vec![vset("Person", false)],
edges: vec![],
deletes: vec![],
};
assert!(
!bulk_lane_eligible(&no_gar),
"missing `_gar_id` → MERGE (byte-identity)"
);
// A duplicate label — build_bulk_payload would reject it.
let dup = GraphMutation {
edge_deletes: Vec::new(),
vertices: vec![vset("Person", true), vset("Person", true)],
edges: vec![],
deletes: vec![],
};
assert!(!bulk_lane_eligible(&dup), "duplicate label → MERGE");
// Any edge → MERGE (bulk endpoints want 0-based `_gar_id`, not user ids).
let with_edge = GraphMutation {
edge_deletes: Vec::new(),
vertices: vec![vset("Person", true)],
edges: vec![EdgeSet {
src_label: "Person".into(),
edge_type: "KNOWS".into(),
dst_label: "Person".into(),
src_id_column: "_gar_id".into(),
dst_id_column: "_gar_id".into(),
edge_key_props: vec![],
replace_outgoing: false,
batch: RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, false),
])),
vec![
Arc::new(Int64Array::from(vec![0])),
Arc::new(Int64Array::from(vec![1])),
],
)
.unwrap(),
}],
deletes: vec![],
};
assert!(!bulk_lane_eligible(&with_edge), "edges present → MERGE");
// Any delete → MERGE (bulk cannot remove).
let with_delete = GraphMutation {
edge_deletes: Vec::new(),
vertices: vec![vset("Person", true)],
edges: vec![],
deletes: vec![VertexDelete {
label: "Person".into(),
id_column: "_gar_id".into(),
batch: RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new(
"_gar_id",
DataType::Int64,
false,
)])),
vec![Arc::new(Int64Array::from(vec![0]))],
)
.unwrap(),
}],
};
assert!(!bulk_lane_eligible(&with_delete), "deletes present → MERGE");
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"sink::bulk_lane_eligibility_predicate",
bulk_lane_eligible(&ok)
&& !bulk_lane_eligible(&no_gar)
&& !bulk_lane_eligible(&dup)
&& !bulk_lane_eligible(&with_edge)
&& !bulk_lane_eligible(&with_delete),
"vertex-only + `_gar_id` + unique labels ⇒ bulk; else MERGE",
);
}
#[test]
fn in_place_update_does_not_delete_the_upserted_node() {
// An in-place UPDATE on a stable key: UPDATE_AFTER(id=7) lands in
// `vertices`, UPDATE_BEFORE(id=7) lands in `deletes`. Without net-effect
// resolution the sink would DETACH-DELETE id=7 last, destroying the node
// it just upserted. id=9 is a *true* delete (no matching upsert) and
// must still be removed.
let upsert = VertexSet {
label: "Person".into(),
id_column: "user_id".into(),
batch: RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("user_id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(vec![7])),
Arc::new(StringArray::from(vec!["new-name"])),
],
)
.unwrap(),
};
let delete = VertexDelete {
label: "Person".into(),
id_column: "user_id".into(),
batch: RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new(
"user_id",
DataType::Int64,
false,
)])),
vec![Arc::new(Int64Array::from(vec![7, 9]))],
)
.unwrap(),
};
let mutation = GraphMutation {
edge_deletes: Vec::new(),
vertices: vec![upsert],
edges: vec![],
deletes: vec![delete.clone()],
};
let upserted = upserted_vertex_keys(&mutation).unwrap();
let surviving = surviving_delete_ids(&delete, &upserted).unwrap();
// id 7 was upserted → suppressed; id 9 is a real delete → survives.
let pass = surviving == vec!["9"];
assert!(pass, "expected only id 9 to be deleted, got {surviving:?}");
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"sink::in_place_update_does_not_delete_the_upserted_node",
pass,
&format!("surviving_deletes={surviving:?} (id 7 upsert wins over delete)"),
);
}
fn knows(src: &[Option<i64>], dst: &[Option<i64>], id: &[i64]) -> EdgeSet {
EdgeSet {
src_label: "person".into(),
edge_type: "knows".into(),
dst_label: "person".into(),
src_id_column: "id".into(),
dst_id_column: "id".into(),
edge_key_props: vec!["id".into()],
replace_outgoing: false,
batch: RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, true),
Field::new("dst", DataType::Int64, true),
Field::new("id", DataType::Int64, false),
])),
vec![
Arc::new(Int64Array::from(src.to_vec())),
Arc::new(Int64Array::from(dst.to_vec())),
Arc::new(Int64Array::from(id.to_vec())),
],
)
.unwrap(),
}
}
#[test]
fn an_edge_delete_matches_its_endpoints_type_and_key() {
let set = knows(&[Some(1)], &[Some(2)], &[7]);
let (rows, skipped) = surviving_edge_deletes(&set, &HashSet::new()).unwrap();
assert_eq!((rows.as_slice(), skipped), (["{src: 1, dst: 2, props: {`id`: 7}}".to_string()].as_slice(), 0));
assert_eq!(
edge_delete_query(&set, &rows),
"UNWIND [{src: 1, dst: 2, props: {`id`: 7}}] AS r \
MATCH (a:`person` {`id`: r.src})-[e:`knows` {`id`: r.props.`id`}]->(b:`person` {`id`: r.dst}) DELETE e"
);
}
#[test]
fn an_edge_upserted_in_the_round_survives_and_a_null_endpoint_is_skipped() {
// An UPDATE of edge 7 (delete + insert, same endpoints and key) keeps it;
// edge 8 is really deleted; the key-only row cannot name its edge.
let upserts = GraphMutation { edges: vec![knows(&[Some(1)], &[Some(2)], &[7])], ..Default::default() };
let upserted = upserted_edge_keys(&upserts).unwrap();
let deletes = knows(&[Some(1), Some(1), None], &[Some(2), Some(3), None], &[7, 8, 9]);
let (rows, skipped) = surviving_edge_deletes(&deletes, &upserted).unwrap();
assert_eq!(rows, vec!["{src: 1, dst: 3, props: {`id`: 8}}".to_string()]);
assert_eq!(skipped, 1);
}
#[test]
fn an_edge_delete_keeps_the_bulk_lane_shut() {
let m = GraphMutation { edge_deletes: vec![knows(&[Some(1)], &[Some(2)], &[7])], ..Default::default() };
assert!(!bulk_lane_eligible(&m));
}
fn people_set(ids: &[Option<i64>], names: &[&str]) -> VertexSet {
VertexSet {
label: "Person".into(),
id_column: "id".into(),
batch: RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, true), Field::new("name", DataType::Utf8, false)])),
vec![Arc::new(Int64Array::from(ids.to_vec())), Arc::new(StringArray::from(names.to_vec()))],
)
.unwrap(),
}
}
#[test]
fn a_first_load_keeps_the_last_row_per_id_and_drops_null_ids() {
let a = people_set(&[Some(1), Some(2), None], &["a", "b", "nobody"]);
let b = people_set(&[Some(1), Some(3)], &["a2", "c"]);
let batch = first_load_rows(&[&a, &b]).unwrap().unwrap();
let ids: Vec<i64> = batch.column(0).as_any().downcast_ref::<Int64Array>().unwrap().values().to_vec();
let names: Vec<&str> = batch.column(1).as_any().downcast_ref::<StringArray>().unwrap().iter().flatten().collect();
assert_eq!(ids, vec![1, 2, 3], "first-seen order");
assert_eq!(names, vec!["a2", "b", "c"], "the last row of id 1 wins");
}
#[test]
fn a_first_load_of_mismatched_sets_is_left_to_merge() {
let a = people_set(&[Some(1)], &["a"]);
let mut b = people_set(&[Some(2)], &["b"]);
b.id_column = "name".into();
assert!(first_load_rows(&[&a, &b]).unwrap().is_none());
assert!(first_load_rows(&[&people_set(&[None], &["x"])]).unwrap().is_none(), "nothing left");
}
fn works_at(src: &[Option<i64>], dst: &[Option<i64>]) -> EdgeSet {
EdgeSet {
src_label: "User".into(),
edge_type: "WORKS_AT".into(),
dst_label: "Company".into(),
src_id_column: "user_id".into(),
dst_id_column: "id".into(),
edge_key_props: Vec::new(),
replace_outgoing: true,
batch: RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new("src", DataType::Int64, true), Field::new("dst", DataType::Int64, true)])),
vec![Arc::new(Int64Array::from(src.to_vec())), Arc::new(Int64Array::from(dst.to_vec()))],
)
.unwrap(),
}
}
#[test]
fn a_replace_set_keeps_the_last_row_per_source_and_a_cleared_key() {
// User 1 moved twice in the window (10, then 20); user 2's key was cleared.
let set = last_row_per_src(&works_at(&[Some(1), Some(2), Some(1), None], &[Some(10), None, Some(20), Some(9)])).unwrap();
let rows = replace_rows(&set).unwrap();
assert_eq!(rows, vec!["{src: 1, dst: 20}".to_string(), "{src: 2, dst: null}".to_string()]);
assert_eq!(
replace_outgoing_query(&set, &rows[..1]),
"UNWIND [{src: 1, dst: 20}] AS r \
MATCH (a:`User` {`user_id`: r.src})-[e:`WORKS_AT`]->(b:`Company`) \
WHERE r.dst IS NULL OR b.`id` <> r.dst DELETE e"
);
}
#[test]
fn a_key_only_edge_delete_is_named_by_its_key() {
// Edge 9 arrives with its key alone; edge 7 has endpoints (the other path).
let set = knows(&[None, Some(1)], &[None, Some(2)], &[9, 7]);
let rows = key_only_edge_rows(&set).unwrap();
assert_eq!(rows, vec!["{props: {`id`: 9}}".to_string()]);
assert_eq!(
edge_delete_by_key_query(&set, &rows),
"UNWIND [{props: {`id`: 9}}] AS r MATCH (:`person`)-[e:`knows` {`id`: r.props.`id`}]->(:`person`) DELETE e"
);
// No key: nothing can name it.
assert!(key_only_edge_rows(&works_at(&[None], &[None])).unwrap().is_empty());
}
}