knut-bifrost 0.1.9

//! **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::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.
pub struct FalkorDbSink {
    loader: FalkorDbLoader,
    batch_size: usize,
}

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,
        })
    }

    pub fn with_batch_size(mut self, n: usize) -> Self {
        self.batch_size = n.max(1);
        self
    }

    async fn upsert_vertices(&mut self, set: &VertexSet) -> Result<usize> {
        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> {
        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)?;
        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)
    }

    /// 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)?;
        // Vertices strictly before edges: edges MATCH their endpoints.
        for set in &mutation.vertices {
            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;
        }
        for set in &mutation.edges {
            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).
        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 = if set.edge_key_props.is_empty() {
        String::new()
    } else {
        let parts: Vec<String> = set
            .edge_key_props
            .iter()
            .map(|p| format!("`{p}`: r.props.`{p}`"))
            .collect();
        format!(" {{{}}}", parts.join(", "))
    };
    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,
    )
}

/// 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() {
        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,
            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 {
            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 {
            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 {
            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 {
            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![],
                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 {
            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 {
            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)"),
        );
    }
}