knut-bifrost 0.2.0

//! **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());
    }
}