From b755f22f9674d7feecb94fa655de3372ab452142 Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Wed, 21 Oct 2026 06:21:52 +0800 Subject: [PATCH 01/24] feat(graph): batch edge inserts and deletes in one statement A loader with N edges sent one GRAPH INSERT EDGE or GRAPH DELETE EDGE per edge. Each cost a round trip and its own Calvin transaction. GRAPH INSERT EDGES and GRAPH DELETE EDGES now carry the edges of one statement in a VALUES list. The handler builds one task with every edge and submits it as one sequenced Calvin transaction. The submit path already slices a batch per home pair and sums the owned-edge counts (edge_slices, ReplyFold::SumOwnedEdges). Every edge lands or none does. A deadline that ends the wait leaves an unknown but all-or-nothing outcome, and a retry is idempotent. Surrogates for all endpoints of a statement are assigned with one assign_surrogates_routed call. Write policy is evaluated per edge with resolve_edge_write_rls. A delete under a write policy keeps the per-edge path: the allowed edges apply, the denied edge stays, and the statement returns an error. A batch inside a transaction block validates the whole list before it stages any tuple. The bounds are protocol constants, MAX_EDGES_PER_BATCH_STATEMENT and MAX_BATCH_TEXT_BYTES, so the parser and the client agree: 1000 edges or 1 MiB of SQL text, whichever the statement crosses first. The error names the limit. --- nodedb-sql/src/ddl_ast/graph_parse/cursor.rs | 6 + nodedb-sql/src/ddl_ast/graph_parse/entry.rs | 128 ++++++- nodedb-sql/src/ddl_ast/graph_parse/mod.rs | 8 + .../src/ddl_ast/graph_parse/variants.rs | 117 ++++++ nodedb-sql/src/ddl_ast/graph_types.rs | 9 + nodedb-sql/src/ddl_ast/mod.rs | 2 +- nodedb-sql/src/ddl_ast/statement/mod.rs | 2 +- .../src/ddl_ast/statement/types/graph.rs | 18 +- nodedb-types/src/protocol/batch.rs | 16 + nodedb-types/src/protocol/mod.rs | 2 +- .../shared/ddl/neutral/graph_ops/dispatch.rs | 10 +- .../shared/ddl/neutral/graph_ops/edge.rs | 15 +- .../ddl/neutral/graph_ops/edge_batch.rs | 359 ++++++++++++++++++ .../neutral/graph_ops/edge_batch_resolve.rs | 349 +++++++++++++++++ .../shared/ddl/neutral/graph_ops/mod.rs | 2 + 15 files changed, 1035 insertions(+), 8 deletions(-) create mode 100644 nodedb/src/control/server/shared/ddl/neutral/graph_ops/edge_batch.rs create mode 100644 nodedb/src/control/server/shared/ddl/neutral/graph_ops/edge_batch_resolve.rs diff --git a/nodedb-sql/src/ddl_ast/graph_parse/cursor.rs b/nodedb-sql/src/ddl_ast/graph_parse/cursor.rs index 40ad8c061..6c51bee5d 100644 --- a/nodedb-sql/src/ddl_ast/graph_parse/cursor.rs +++ b/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 = diff --git a/nodedb-sql/src/ddl_ast/graph_parse/entry.rs b/nodedb-sql/src/ddl_ast/graph_parse/entry.rs index 151946d4e..b02095679 100644 --- a/nodedb-sql/src/ddl_ast/graph_parse/entry.rs +++ b/nodedb-sql/src/ddl_ast/graph_parse/entry.rs @@ -62,7 +62,11 @@ fn parse_graph(sql: &str) -> Result { }); } - if upper.starts_with("GRAPH INSERT EDGE ") { + if upper.starts_with("GRAPH INSERT EDGES ") { + variants::parse_insert_edges(&toks, sql.len()) + } else if upper.starts_with("GRAPH DELETE EDGES ") { + variants::parse_delete_edges(&toks, sql.len()) + } else if upper.starts_with("GRAPH INSERT EDGE ") { variants::parse_insert_edge(&toks) } else if upper.starts_with("GRAPH DELETE EDGE ") { variants::parse_delete_edge(&toks) @@ -88,7 +92,13 @@ fn parse_graph(sql: &str) -> Result { /// The command words of a `GRAPH` statement, for an error message: /// `GRAPH NEIGHBORS`, `GRAPH INSERT EDGE`, `GRAPH RAG FUSION`. fn graph_command(upper: &str) -> String { - const MULTI_WORD: [&str; 3] = ["GRAPH INSERT EDGE", "GRAPH DELETE EDGE", "GRAPH RAG FUSION"]; + const MULTI_WORD: [&str; 5] = [ + "GRAPH INSERT EDGES", + "GRAPH DELETE EDGES", + "GRAPH INSERT EDGE", + "GRAPH DELETE EDGE", + "GRAPH RAG FUSION", + ]; if let Some(command) = MULTI_WORD.iter().find(|c| upper.starts_with(*c)) { return (*command).to_owned(); } @@ -151,6 +161,120 @@ mod tests { } } + #[test] + fn parse_graph_insert_edges_batch() { + let stmt = + parsed("GRAPH INSERT EDGES IN 'edges' VALUES ('a','b','CALLS'), ('c','d','IMPORTS')"); + match stmt { + NodedbStatement::Graph(GraphStmt::GraphInsertEdges { collection, edges }) => { + assert_eq!(collection, "edges"); + assert_eq!(edges.len(), 2); + assert_eq!(edges[0].src, "a"); + assert_eq!(edges[0].dst, "b"); + assert_eq!(edges[0].label, "CALLS"); + assert_eq!(edges[1].src, "c"); + assert_eq!(edges[1].label, "IMPORTS"); + } + other => panic!("expected GraphInsertEdges, got {other:?}"), + } + } + + #[test] + fn parse_graph_delete_edges_batch_accepts_bare_words() { + let stmt = parsed("GRAPH DELETE EDGES IN 'edges' VALUES (a, b, CALLS)"); + match stmt { + NodedbStatement::Graph(GraphStmt::GraphDeleteEdges { collection, edges }) => { + assert_eq!(collection, "edges"); + assert_eq!(edges.len(), 1); + assert_eq!(edges[0].src, "a"); + assert_eq!(edges[0].dst, "b"); + assert_eq!(edges[0].label, "CALLS"); + } + other => panic!("expected GraphDeleteEdges, got {other:?}"), + } + } + + #[test] + fn batch_edge_cap_is_enforced() { + let mut sql = String::from("GRAPH INSERT EDGES IN 'edges' VALUES "); + for i in 0..=variants::MAX_EDGES_PER_BATCH { + if i > 0 { + sql.push(','); + } + sql.push_str(&format!("('s{i}','d{i}','L')")); + } + let error = try_parse(&sql) + .expect("input is graph DSL") + .expect_err("a batch over the cap must not produce a statement"); + assert!( + error.to_string().contains("at most 1000 edges"), + "the error must name the cap: {error}" + ); + } + + #[test] + fn batch_edge_text_cap_is_enforced_before_the_count_cap() { + // Three edges, far below the 1000-edge cap, but the field values are + // long enough to push the statement past 1 MiB. The text cap must be + // what rejects it, since the count says nothing about the work. + let long = "x".repeat(400 * 1024); + let sql = format!( + "GRAPH INSERT EDGES IN 'edges' VALUES ('{long}','{long}','L'), \ + ('{long}','{long}','L'), ('{long}','{long}','L')" + ); + assert!( + sql.len() > variants::MAX_BATCH_TEXT_BYTES, + "the fixture must exceed the text cap, was {} bytes", + sql.len() + ); + let error = try_parse(&sql) + .expect("input is graph DSL") + .expect_err("a batch over the text cap must not produce a statement"); + assert!( + error.to_string().contains("bytes of SQL text"), + "the error must name the text cap: {error}" + ); + } + + #[test] + fn batch_edge_at_the_text_cap_is_accepted() { + // The cap is a maximum, not an exclusive bound: a statement exactly at + // it must still parse. Built by padding the collection name, which is a + // single quoted field. + let base = "GRAPH INSERT EDGES IN '' VALUES ('a','b','L')"; + let pad = variants::MAX_BATCH_TEXT_BYTES - base.len(); + let sql = format!( + "GRAPH INSERT EDGES IN '{}' VALUES ('a','b','L')", + "c".repeat(pad) + ); + assert_eq!(sql.len(), variants::MAX_BATCH_TEXT_BYTES); + try_parse(&sql) + .expect("input is graph DSL") + .expect("a statement exactly at the text cap must parse"); + } + + #[test] + fn batch_edge_rejects_malformed_triples() { + let error = try_parse("GRAPH INSERT EDGES IN 'edges' VALUES ('a','b')") + .expect("input is graph DSL") + .expect_err("a partial triple must not produce a statement"); + assert!( + error.to_string().contains("triples"), + "the error must name the tuple shape: {error}" + ); + } + + #[test] + fn batch_edge_rejects_per_edge_properties() { + let error = try_parse("GRAPH INSERT EDGES IN 'edges' VALUES ('a','b','L') PROPERTIES '{}'") + .expect("input is graph DSL") + .expect_err("per-edge PROPERTIES is not supported in the batch form"); + assert!( + error.to_string().contains("PROPERTIES"), + "the error must name the rejected clause: {error}" + ); + } + /// A missing required clause is a malformed graph statement, not a /// non-graph one. `None` here would send it to the SQL parser, which /// reports only that `GRAPH` is not SQL. diff --git a/nodedb-sql/src/ddl_ast/graph_parse/mod.rs b/nodedb-sql/src/ddl_ast/graph_parse/mod.rs index fcd635068..7d8fdd05f 100644 --- a/nodedb-sql/src/ddl_ast/graph_parse/mod.rs +++ b/nodedb-sql/src/ddl_ast/graph_parse/mod.rs @@ -17,6 +17,14 @@ //! required clause that is absent, or a clause value outside its own //! vocabulary, means no statement reaches the boundary at all. Both are //! reported as a parse error naming what is wrong. +//! +//! Accepted-grammar limitation for the batched edge statements: the tokenizer +//! drops `(`, `)`, and `,`, so `GRAPH INSERT EDGES` reads the `VALUES` tail as +//! a flat field list and only checks that its length is a multiple of three. +//! `VALUES ('a','b','L'), ('c','d','L')` and `VALUES ('a','b','L') +//! ('c','d','L')` produce the same two triples, and a trailing token is read as +//! another triple's field rather than rejected. Closing that needs a tokenizer +//! change, so it is tracked as a follow-up rather than fixed here. mod cursor; mod edge_predicate; diff --git a/nodedb-sql/src/ddl_ast/graph_parse/variants.rs b/nodedb-sql/src/ddl_ast/graph_parse/variants.rs index 58697687f..7488e8259 100644 --- a/nodedb-sql/src/ddl_ast/graph_parse/variants.rs +++ b/nodedb-sql/src/ddl_ast/graph_parse/variants.rs @@ -14,6 +14,7 @@ use super::{ fusion_params::{FusionParams, RAG_FUSION_KEYWORDS}, tokenizer::Tok, }; +use crate::ddl_ast::graph_types::GraphEdgeTuple; use crate::error::SqlError; use nodedb_types::filter::MetadataFilter; @@ -42,6 +43,122 @@ pub(super) fn parse_insert_edge(toks: &[Tok<'_>]) -> Result], + sql_len: usize, +) -> Result { + let (collection, edges) = parse_edge_batch(toks, sql_len, "GRAPH INSERT EDGES")?; + Ok(NodedbStatement::Graph(GraphStmt::GraphInsertEdges { + collection, + edges, + })) +} + +pub(super) fn parse_delete_edges( + toks: &[Tok<'_>], + sql_len: usize, +) -> Result { + let (collection, edges) = parse_edge_batch(toks, sql_len, "GRAPH DELETE EDGES")?; + Ok(NodedbStatement::Graph(GraphStmt::GraphDeleteEdges { + collection, + edges, + })) +} + +/// Shared body of the batch parsers. +/// +/// Grammar: `IN '' VALUES ('','','