feat(graph): batch edge inserts and deletes in one statement

nodedb-client/src/graph_dsl/mod.rs

@@ -10,5 +10,5 @@ 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_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,71 @@ 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> {
+    if edges.is_empty() {
+        return Err(NodeDbError::storage(
+            "batch_graph_insert_edges requires at least one edge",
+        ));
+    }
+    if edges.len() > MAX_EDGES_PER_BATCH_STATEMENT {
+        return Err(NodeDbError::storage(format!(
+            "batch_graph_insert_edges carries at most {MAX_EDGES_PER_BATCH_STATEMENT} edges per \
+             statement, got {}",
+            edges.len()
+        )));
+    }
+    let mut sql = format!(
+        "GRAPH INSERT 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!(
+            "batch_graph_insert_edges 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 +215,99 @@ 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 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,23 @@ 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(|_| ())
+    }
+
     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,20 @@ 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(|_| ())
+    }
+
     pub(super) async fn graph_traverse_impl(
         &self,
         collection: &str,

nodedb-sql/src/ddl_ast/graph_parse/cursor.rs

@@ -65,6 +65,12 @@ impl<'t, 'a> Cursor<'t, 'a> {
         }
     }
 
+    /// The tokens not yet consumed.
+    pub(super) fn rest(&self) -> &'t [Tok<'a>] {
+        let toks: &'t [Tok<'a>] = self.toks;
+        toks.get(self.pos..).unwrap_or_default()
+    }
+
     /// Consume the next token when it is the bare word `keyword`.
     pub(super) fn eat_keyword(&mut self, keyword: &str) -> bool {
         let hit =