feat(graph): batch edge inserts and deletes, with neighbors, label list and DDL commit fixes

nodedb-client/src/graph_dsl/mod.rs

@@ -10,5 +10,7 @@ pub(crate) mod pagerank;
 pub(crate) mod traverse;
 
 pub(crate) use decode::{decode_path_result, decode_traverse_result};
-pub(crate) use pagerank::{build_pagerank_sql, parse_pagerank_rows};
+pub(crate) use pagerank::{
+    build_delete_edges_sql, build_insert_edges_sql, build_pagerank_sql, parse_pagerank_rows,
+};
 pub(crate) use traverse::{build_graph_path_sql, build_graph_traverse_sql};

nodedb-client/src/graph_dsl/pagerank.rs

@@ -5,6 +5,7 @@
 use std::collections::HashMap;
 
 use nodedb_types::error::{NodeDbError, NodeDbResult};
+use nodedb_types::id::NodeId;
 use nodedb_types::value::Value;
 
 use crate::sql_escape::quote_string_literal;
@@ -83,6 +84,87 @@ fn value_as_f64(v: &Value) -> Option<f64> {
         .or_else(|| v.as_str().and_then(|s| s.parse().ok()))
 }
 
+/// Largest number of edges one batch statement may carry.
+///
+/// Read from the protocol bound, which the statement parser also reads, so the
+/// client and the server cannot hold different numbers. Checked here before
+/// building the statement so an oversized batch fails with the limit named
+/// instead of a round trip and a parse error.
+pub(crate) use nodedb_types::protocol::{MAX_BATCH_TEXT_BYTES, MAX_EDGES_PER_BATCH_STATEMENT};
+
+/// Build one `GRAPH INSERT EDGES` statement carrying every edge.
+///
+/// The batched statement is one round trip for the whole list, which is what
+/// `batch_graph_insert_edges` promises. Its server-side form is property-less:
+/// `BatchEdge` carries no property object, so a caller that needs per-edge
+/// properties stays on the singular statement.
+///
+/// A batch over [`MAX_EDGES_PER_BATCH_STATEMENT`] is refused here rather than
+/// sent to be rejected.
+pub(crate) fn build_insert_edges_sql(
+    collection: &str,
+    edges: &[(&str, &str, &str)],
+) -> NodeDbResult<String> {
+    build_edges_sql("INSERT", "batch_graph_insert_edges", collection, edges)
+}
+
+/// Build one `GRAPH DELETE EDGES` statement for a list of `(src, dst, label)`
+/// triples. The bounds are the ones of [`build_insert_edges_sql`].
+pub(crate) fn build_delete_edges_sql(
+    collection: &str,
+    edges: &[(&str, &str, &str)],
+) -> NodeDbResult<String> {
+    build_edges_sql("DELETE", "batch_graph_delete_edges", collection, edges)
+}
+
+fn build_edges_sql(
+    verb: &str,
+    caller: &str,
+    collection: &str,
+    edges: &[(&str, &str, &str)],
+) -> NodeDbResult<String> {
+    if edges.is_empty() {
+        return Err(NodeDbError::storage(format!(
+            "{caller} requires at least one edge"
+        )));
+    }
+    if edges.len() > MAX_EDGES_PER_BATCH_STATEMENT {
+        return Err(NodeDbError::storage(format!(
+            "{caller} carries at most {MAX_EDGES_PER_BATCH_STATEMENT} edges per statement, got {}",
+            edges.len()
+        )));
+    }
+    let mut sql = format!(
+        "GRAPH {verb} EDGES IN {} VALUES ",
+        quote_string_literal(collection)
+    );
+    for (i, (from, to, label)) in edges.iter().enumerate() {
+        for node in [from, to] {
+            NodeId::try_new(*node)
+                .map_err(|e| NodeDbError::storage(format!("invalid node id: {e}")))?;
+        }
+        if i > 0 {
+            sql.push(',');
+        }
+        sql.push_str(&format!(
+            "({},{},{})",
+            quote_string_literal(from),
+            quote_string_literal(to),
+            quote_string_literal(label)
+        ));
+    }
+    // The count is not the only bound: a few edges with long node ids exceed
+    // the text cap while staying far below it. Check the statement that was
+    // actually built, so the client refuses the same statements the parser does.
+    if sql.len() > MAX_BATCH_TEXT_BYTES {
+        return Err(NodeDbError::storage(format!(
+            "{caller} builds at most {MAX_BATCH_TEXT_BYTES} bytes of SQL text per statement, got {}",
+            sql.len()
+        )));
+    }
+    Ok(sql)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -149,4 +231,117 @@ mod tests {
         let rows = vec![vec![Value::String("a".into())]];
         assert!(parse_pagerank_rows(&columns, &rows).is_err());
     }
+
+    #[test]
+    fn insert_edges_sql_carries_every_triple_in_one_statement() {
+        let sql = build_insert_edges_sql("c", &[("a", "b", "L"), ("b", "c", "L")]).unwrap();
+        assert_eq!(
+            sql,
+            "GRAPH INSERT EDGES IN 'c' VALUES ('a','b','L'),('b','c','L')"
+        );
+        // One statement, one VALUES clause: a per-edge loop would build two.
+        assert_eq!(sql.matches("VALUES").count(), 1);
+        assert_eq!(sql.matches("GRAPH INSERT EDGES").count(), 1);
+    }
+
+    #[test]
+    fn delete_edges_sql_carries_every_triple_in_one_statement() {
+        let sql = build_delete_edges_sql("c", &[("a", "b", "L"), ("b", "c", "L")]).unwrap();
+        assert_eq!(
+            sql,
+            "GRAPH DELETE EDGES IN 'c' VALUES ('a','b','L'),('b','c','L')"
+        );
+    }
+
+    #[test]
+    fn delete_edges_sql_refuses_an_empty_batch() {
+        let error = build_delete_edges_sql("c", &[]).expect_err("an empty batch has nothing to do");
+        assert!(
+            error.to_string().contains("batch_graph_delete_edges"),
+            "the error must name the caller: {error}"
+        );
+    }
+
+    #[test]
+    fn insert_edges_sql_escapes_every_field() {
+        // A quote in any field must not break out of its literal.
+        let sql = build_insert_edges_sql("c'x", &[("a'1", "b'2", "L'3")]).unwrap();
+        assert!(
+            sql.contains("'a''1'") && sql.contains("'b''2'") && sql.contains("'L''3'"),
+            "every field must be escaped: {sql}"
+        );
+        assert!(
+            sql.contains("'c''x'"),
+            "the collection must be escaped: {sql}"
+        );
+    }
+
+    #[test]
+    fn insert_edges_sql_refuses_an_empty_batch() {
+        let error = build_insert_edges_sql("c", &[]).expect_err("an empty batch has nothing to do");
+        assert!(
+            error.to_string().contains("at least one edge"),
+            "the error must say why: {error}"
+        );
+    }
+
+    #[test]
+    fn insert_edges_sql_refuses_a_batch_over_the_cap() {
+        let owned: Vec<(String, String, String)> = (0..=MAX_EDGES_PER_BATCH_STATEMENT)
+            .map(|i| (format!("s{i}"), format!("d{i}"), "L".to_string()))
+            .collect();
+        let edges: Vec<(&str, &str, &str)> = owned
+            .iter()
+            .map(|(s, d, l)| (s.as_str(), d.as_str(), l.as_str()))
+            .collect();
+        let error = build_insert_edges_sql("c", &edges)
+            .expect_err("a batch over the cap must not build a statement");
+        assert!(
+            error.to_string().contains("at most 1000 edges"),
+            "the client must name the limit before sending: {error}"
+        );
+    }
+
+    #[test]
+    fn insert_edges_sql_accepts_a_batch_exactly_at_the_cap() {
+        let owned: Vec<(String, String, String)> = (0..MAX_EDGES_PER_BATCH_STATEMENT)
+            .map(|i| (format!("s{i}"), format!("d{i}"), "L".to_string()))
+            .collect();
+        let edges: Vec<(&str, &str, &str)> = owned
+            .iter()
+            .map(|(s, d, l)| (s.as_str(), d.as_str(), l.as_str()))
+            .collect();
+        let sql = build_insert_edges_sql("c", &edges).expect("the cap is a maximum");
+        assert_eq!(sql.matches("VALUES").count(), 1);
+    }
+
+    #[test]
+    fn insert_edges_sql_refuses_a_batch_over_the_text_cap() {
+        // 1000 edges of 600-byte ids: exactly at the count cap and every id is
+        // legal, but the built statement exceeds 1 MiB. The count alone would
+        // let this through.
+        let long = "x".repeat(600);
+        let owned: Vec<(&str, &str, &str)> = (0..MAX_EDGES_PER_BATCH_STATEMENT)
+            .map(|_| (long.as_str(), long.as_str(), "L"))
+            .collect();
+        let error = build_insert_edges_sql("c", &owned)
+            .expect_err("a batch over the text cap must not build a statement");
+        assert!(
+            error.to_string().contains("bytes of SQL text"),
+            "the client must name the text limit before sending: {error}"
+        );
+    }
+
+    #[test]
+    fn insert_edges_sql_refuses_an_invalid_node_id() {
+        // The in-process fallback path validates node ids; the wire builder must
+        // reject the same input instead of sending a statement the parser may
+        // not diagnose.
+        let error = build_insert_edges_sql("c", &[("a", "", "L")])
+            .expect_err("an empty node id must not reach the wire");
+        assert!(
+            error.to_string().contains("invalid node id"),
+            "the client must say which field is bad: {error}"
+        );
+    }
 }

nodedb-client/src/native/client/dispatch.rs

@@ -92,6 +92,14 @@ impl NodeDb for NativeClient {
             .await
     }
 
+    async fn batch_graph_insert_edges(
+        &self,
+        collection: &str,
+        edges: &[(&str, &str, &str)],
+    ) -> NodeDbResult<()> {
+        self.batch_graph_insert_edges_impl(collection, edges).await
+    }
+
     async fn graph_insert_edge(
         &self,
         collection: &str,

nodedb-client/src/native/client/graph.rs

@@ -15,6 +15,38 @@ use crate::native::connection::check_error;
 use crate::sql_escape::quote_string_literal;
 
 impl NativeClient {
+    /// Insert a batch of edges as ONE `GRAPH INSERT EDGES` statement.
+    ///
+    /// The statement carries the whole list over the existing `Sql` opcode, so
+    /// the batch costs one round trip instead of one per edge — no new wire
+    /// operation is needed for the transport to be batched. The server-side form
+    /// is property-less, which is what the caller's `(&str, &str, &str)`
+    /// triples carry.
+    pub(super) async fn batch_graph_insert_edges_impl(
+        &self,
+        collection: &str,
+        edges: &[(&str, &str, &str)],
+    ) -> NodeDbResult<()> {
+        let sql = crate::graph_dsl::build_insert_edges_sql(collection, edges)?;
+        let mut conn = self.pool.acquire().await?;
+        conn.execute_sql(&sql).await.map(|_| ())
+    }
+
+    /// Delete a batch of edges as ONE `GRAPH DELETE EDGES` statement.
+    ///
+    /// A delete of an absent edge counts zero, and a refused edge refuses the
+    /// whole batch. This is an inherent method: the `NodeDb` trait file is at
+    /// its line limit.
+    pub async fn batch_graph_delete_edges(
+        &self,
+        collection: &str,
+        edges: &[(&str, &str, &str)],
+    ) -> NodeDbResult<()> {
+        let sql = crate::graph_dsl::build_delete_edges_sql(collection, edges)?;
+        let mut conn = self.pool.acquire().await?;
+        conn.execute_sql(&sql).await.map(|_| ())
+    }
+
     pub(super) async fn graph_traverse_impl(
         &self,
         collection: &str,

nodedb-client/src/remote/client/dispatch.rs

@@ -89,6 +89,14 @@ impl NodeDb for NodeDbRemote {
             .await
     }
 
+    async fn batch_graph_insert_edges(
+        &self,
+        collection: &str,
+        edges: &[(&str, &str, &str)],
+    ) -> NodeDbResult<()> {
+        self.batch_graph_insert_edges_impl(collection, edges).await
+    }
+
     async fn graph_insert_edge(
         &self,
         collection: &str,

nodedb-client/src/remote/client/graph.rs

@@ -14,6 +14,34 @@ use super::core::NodeDbRemote;
 use crate::sql_escape::quote_string_literal;
 
 impl NodeDbRemote {
+    /// Insert a batch of edges as ONE `GRAPH INSERT EDGES` statement.
+    ///
+    /// The statement carries the whole list, so the batch costs one round trip
+    /// instead of one per edge. The server-side form is property-less, which is
+    /// what the caller's `(&str, &str, &str)` triples carry.
+    pub(super) async fn batch_graph_insert_edges_impl(
+        &self,
+        collection: &str,
+        edges: &[(&str, &str, &str)],
+    ) -> NodeDbResult<()> {
+        let sql = crate::graph_dsl::build_insert_edges_sql(collection, edges)?;
+        self.simple_query_raw(&sql).await.map(|_| ())
+    }
+
+    /// Delete a batch of edges as ONE `GRAPH DELETE EDGES` statement.
+    ///
+    /// A delete of an absent edge counts zero, and a refused edge refuses the
+    /// whole batch. This is an inherent method: the `NodeDb` trait file is at
+    /// its line limit.
+    pub async fn batch_graph_delete_edges(
+        &self,
+        collection: &str,
+        edges: &[(&str, &str, &str)],
+    ) -> NodeDbResult<()> {
+        let sql = crate::graph_dsl::build_delete_edges_sql(collection, edges)?;
+        self.simple_query_raw(&sql).await.map(|_| ())
+    }
+
     pub(super) async fn graph_traverse_impl(
         &self,
         collection: &str,

nodedb-graph/src/csr/index/lookup.rs

@@ -67,7 +67,9 @@ impl CsrIndex {
         }
         if matches!(direction, Direction::In | Direction::Both) {
             for (lid, src) in self.dense_iter_in(node_id) {
-                if labels.keeps(lid) {
+                // A self-loop is one edge. `Both` read it in the outbound pass.
+                let repeats_self_loop = direction == Direction::Both && src == node_id;
+                if labels.keeps(lid) && !repeats_self_loop {
                     result.push((
                         self.id_to_label[lid as usize].clone(),
                         self.id_to_node[src as usize].clone(),
@@ -412,6 +414,35 @@ mod tests {
             .collect()
     }
 
+    #[test]
+    fn both_directions_list_a_self_loop_once() {
+        use crate::csr::Direction;
+
+        let mut csr = CsrIndex::new(test_memory());
+        csr.add_edge("a", "L", "b").unwrap();
+        csr.add_edge("a", "L", "a").unwrap();
+        csr.add_edge("c", "L", "d").unwrap();
+        csr.add_edge("d", "L", "c").unwrap();
+        for compacted in [false, true] {
+            if compacted {
+                csr.compact().unwrap();
+            }
+            let mut both = csr.neighbors("a", &[], Direction::Both);
+            both.sort();
+            assert_eq!(
+                both,
+                vec![
+                    ("L".to_string(), "a".to_string()),
+                    ("L".to_string(), "b".to_string())
+                ],
+                "compacted: {compacted}"
+            );
+            assert_eq!(csr.neighbors("a", &[], Direction::In).len(), 1);
+            // Two distinct edges to one neighbour stay two entries.
+            assert_eq!(csr.neighbors("c", &[], Direction::Both).len(), 2);
+        }
+    }
+
     #[test]
     fn live_iterators_preserve_dense_then_buffer_order() {
         let mut csr = CsrIndex::new(test_memory());

nodedb-graph/src/csr/index/scoped.rs

@@ -49,7 +49,9 @@ impl CsrIndex {
         }
         if matches!(direction, Direction::In | Direction::Both) {
             for (lid, src) in self.iter_in_edges_raw_in(node_id, collection_id) {
-                if keep(lid) {
+                // A self-loop is one edge. `Both` read it in the outbound pass.
+                let repeats_self_loop = direction == Direction::Both && src == node_id;
+                if keep(lid) && !repeats_self_loop {
                     result.push((
                         self.id_to_label[lid as usize].clone(),
                         self.id_to_node[src as usize].clone(),