diff --git a/Cargo.lock b/Cargo.lock index ed330df75..44d45118b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3964,6 +3964,7 @@ dependencies = [ "nexar", "nodedb-array", "nodedb-bridge", + "nodedb-client", "nodedb-cluster", "nodedb-codec", "nodedb-columnar", diff --git a/nodedb-client/src/graph_dsl/mod.rs b/nodedb-client/src/graph_dsl/mod.rs index 3535685ea..db1f2e7c4 100644 --- a/nodedb-client/src/graph_dsl/mod.rs +++ b/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}; diff --git a/nodedb-client/src/graph_dsl/pagerank.rs b/nodedb-client/src/graph_dsl/pagerank.rs index 54feb6fbb..aca6179ce 100644 --- a/nodedb-client/src/graph_dsl/pagerank.rs +++ b/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 { .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 { + 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 { + 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 { + 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}" + ); + } } diff --git a/nodedb-client/src/native/client/dispatch.rs b/nodedb-client/src/native/client/dispatch.rs index 147583f20..546ff0cd8 100644 --- a/nodedb-client/src/native/client/dispatch.rs +++ b/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, diff --git a/nodedb-client/src/native/client/graph.rs b/nodedb-client/src/native/client/graph.rs index 17cf5e02e..e5926bcb5 100644 --- a/nodedb-client/src/native/client/graph.rs +++ b/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, diff --git a/nodedb-client/src/remote/client/dispatch.rs b/nodedb-client/src/remote/client/dispatch.rs index 7bd355d80..c44930694 100644 --- a/nodedb-client/src/remote/client/dispatch.rs +++ b/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, diff --git a/nodedb-client/src/remote/client/graph.rs b/nodedb-client/src/remote/client/graph.rs index fab3ba9d5..5a53b54eb 100644 --- a/nodedb-client/src/remote/client/graph.rs +++ b/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, diff --git a/nodedb-graph/src/csr/index/lookup.rs b/nodedb-graph/src/csr/index/lookup.rs index b96507d69..b4e650a6f 100644 --- a/nodedb-graph/src/csr/index/lookup.rs +++ b/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()); diff --git a/nodedb-graph/src/csr/index/scoped.rs b/nodedb-graph/src/csr/index/scoped.rs index 8eb780d71..eeeb19e22 100644 --- a/nodedb-graph/src/csr/index/scoped.rs +++ b/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(), diff --git a/nodedb-physical/src/physical_plan/graph/batch_edge.rs b/nodedb-physical/src/physical_plan/graph/batch_edge.rs index 14f2f007d..40a56b1f9 100644 --- a/nodedb-physical/src/physical_plan/graph/batch_edge.rs +++ b/nodedb-physical/src/physical_plan/graph/batch_edge.rs @@ -2,13 +2,18 @@ //! `BatchEdge`: one edge in an `EdgePutBatch` / `EdgeDeleteBatch`. -use nodedb_types::{QualifiedCollection, Surrogate}; +use nodedb_types::{QualifiedCollection, RlsWriteCheck, Surrogate}; /// One edge in an `EdgePutBatch` / `EdgeDeleteBatch`. /// /// `src_surrogate` / `dst_surrogate` carry the global row identity for the /// edge endpoints (resolved at construction time via the surrogate assigner). /// Neither is ever `Surrogate::ZERO`: an edge that carries it is refused. +/// +/// `write_gate` is the edge's own write-policy check, the same slot +/// `GraphOp::EdgeDelete` carries as `rls_write_check`. An `EdgeDeleteBatch` +/// decides it per edge against the stored image, before any tombstone is +/// written. An `EdgePutBatch` ignores it: a put is decided at injection. #[derive( Debug, Clone, @@ -25,4 +30,5 @@ pub struct BatchEdge { pub dst_id: String, pub src_surrogate: Surrogate, pub dst_surrogate: Surrogate, + pub write_gate: RlsWriteCheck, } diff --git a/nodedb-physical/src/physical_plan/rls_write_check_accessor.rs b/nodedb-physical/src/physical_plan/rls_write_check_accessor.rs index 9d9bb37b7..05f7365e8 100644 --- a/nodedb-physical/src/physical_plan/rls_write_check_accessor.rs +++ b/nodedb-physical/src/physical_plan/rls_write_check_accessor.rs @@ -179,6 +179,13 @@ impl PhysicalPlan { { return vec![source_rls_write_check, dest_rls_write_check]; } + // An edge batch carries one check per edge. + if let PhysicalPlan::Graph( + GraphOp::EdgePutBatch { edges } | GraphOp::EdgeDeleteBatch { edges }, + ) = self + { + return edges.iter().map(|edge| &edge.write_gate).collect(); + } // Every other op carries at most one check (`sole_rls_write_check` // is itself exhaustive over `PhysicalPlan`). self.sole_rls_write_check().into_iter().collect() diff --git a/nodedb-sql/src/ddl_ast/graph_parse/batch_shape.rs b/nodedb-sql/src/ddl_ast/graph_parse/batch_shape.rs new file mode 100644 index 000000000..dd6d904fd --- /dev/null +++ b/nodedb-sql/src/ddl_ast/graph_parse/batch_shape.rs @@ -0,0 +1,150 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! Shape check of the `VALUES` tail of a batched edge statement. +//! +//! The tokenizer drops `(`, `)` and `,`, so the token list cannot tell +//! `('a','b','L'), ('c','d','L')` from `('a','b','L') ('c','d','L')` or from +//! three bare words. This check reads the statement's own text instead. + +/// Whether the text after the first unquoted `VALUES` is a comma-separated +/// list of parenthesized triples, with an optional trailing `;`. +pub(super) fn values_tail_is_triples(sql: &str) -> bool { + let Some(tail) = after_values(sql) else { + return false; + }; + let mut chars = tail.chars().peekable(); + loop { + skip_space(&mut chars); + if chars.next() != Some('(') { + return false; + } + for field in 0..3 { + skip_space(&mut chars); + if !read_field(&mut chars) { + return false; + } + skip_space(&mut chars); + let expected = if field == 2 { ')' } else { ',' }; + if chars.next() != Some(expected) { + return false; + } + } + skip_space(&mut chars); + match chars.next() { + Some(',') => continue, + Some(';') => { + skip_space(&mut chars); + return chars.next().is_none(); + } + None => return true, + Some(_) => return false, + } + } +} + +type Chars<'a> = std::iter::Peekable>; + +fn skip_space(chars: &mut Chars<'_>) { + while chars.peek().is_some_and(|c| c.is_whitespace()) { + chars.next(); + } +} + +/// Read one quoted string (`''` escapes a quote) or one bare word. +fn read_field(chars: &mut Chars<'_>) -> bool { + match chars.peek().copied() { + Some(quote @ ('\'' | '"')) => { + chars.next(); + loop { + match chars.next() { + None => return false, + Some(c) if c == quote => { + if chars.peek() == Some("e) { + chars.next(); + } else { + return true; + } + } + Some(_) => {} + } + } + } + Some(c) if !c.is_whitespace() && !matches!(c, ',' | '(' | ')' | ';') => { + while chars + .peek() + .is_some_and(|c| !c.is_whitespace() && !matches!(c, ',' | '(' | ')' | ';')) + { + chars.next(); + } + true + } + _ => false, + } +} + +/// The text after the first `VALUES` word that sits outside a quoted string. +fn after_values(sql: &str) -> Option<&str> { + let bytes = sql.as_bytes(); + let mut quote: Option = None; + let mut i = 0; + while i < bytes.len() { + let b = bytes[i]; + match quote { + Some(q) => { + if b == q { + quote = None; + } + } + None => { + if b == b'\'' || b == b'"' { + quote = Some(b); + } else if sql.is_char_boundary(i) + && sql[i..].len() >= 6 + && sql.is_char_boundary(i + 6) + && sql[i..i + 6].eq_ignore_ascii_case("VALUES") + && (i == 0 || !bytes[i - 1].is_ascii_alphanumeric()) + && bytes.get(i + 6).is_none_or(|n| !n.is_ascii_alphanumeric()) + { + return Some(&sql[i + 6..]); + } + } + } + i += 1; + } + None +} + +#[cfg(test)] +mod tests { + use super::values_tail_is_triples as ok; + + #[test] + fn accepts_comma_separated_triples() { + assert!(ok("IN 'e' VALUES ('a','b','L')")); + assert!(ok("IN 'e' VALUES ('a','b','L'), ('c','d','L');")); + assert!(ok("IN 'e' VALUES (a, b, L)")); + assert!(ok("IN 'e' VALUES ('a,(x)','it''s','L')")); + } + + #[test] + fn refuses_a_missing_comma_between_triples() { + assert!(!ok("IN 'e' VALUES ('a','b','L') ('c','d','L')")); + } + + #[test] + fn refuses_bare_words_without_parentheses() { + assert!(!ok("IN 'e' VALUES a b L")); + } + + #[test] + fn refuses_a_trailing_comma_and_a_short_triple() { + assert!(!ok("IN 'e' VALUES ('a','b','L'),")); + assert!(!ok("IN 'e' VALUES ('a','b')")); + assert!(!ok("IN 'e' VALUES ('a','b','L','c')")); + } + + #[test] + fn a_values_word_inside_the_collection_name_is_not_the_clause() { + assert!(ok("IN 'values' VALUES ('a','b','L')")); + } +} diff --git a/nodedb-sql/src/ddl_ast/graph_parse/cursor.rs b/nodedb-sql/src/ddl_ast/graph_parse/cursor.rs index 40ad8c061..0ebaa7912 100644 --- a/nodedb-sql/src/ddl_ast/graph_parse/cursor.rs +++ b/nodedb-sql/src/ddl_ast/graph_parse/cursor.rs @@ -33,8 +33,13 @@ impl<'t, 'a> Cursor<'t, 'a> { self.statement } - fn peek(&self) -> Option<&'t Tok<'a>> { + /// The next token that is not a comma. Every reader goes through here, + /// so a comma outside a list is as if it were absent. + fn peek(&mut self) -> Option<&'t Tok<'a>> { let toks: &'t [Tok<'a>] = self.toks; + while matches!(toks.get(self.pos), Some(Tok::Comma)) { + self.pos += 1; + } toks.get(self.pos) } @@ -65,6 +70,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 = @@ -144,10 +155,11 @@ impl<'t, 'a> Cursor<'t, 'a> { } } - /// One or more quoted strings: `'a', 'b'` or `('a', 'b')`. + /// One or more quoted strings: `'a', 'b'` or `('a', 'b')`. Exactly one + /// comma separates two items. A comma after the last item is skipped. pub(super) fn quoted_list(&mut self, clause: &str) -> Result, SqlError> { let mut out = Vec::new(); - while let Some(Tok::Quoted(text)) = self.peek() { + if let Some(Tok::Quoted(text)) = self.peek() { out.push(text.clone().into_owned()); self.pos += 1; } @@ -157,7 +169,26 @@ impl<'t, 'a> Cursor<'t, 'a> { self.statement ))); } - Ok(out) + loop { + let toks: &'t [Tok<'a>] = self.toks; + let commas = toks + .get(self.pos..) + .unwrap_or_default() + .iter() + .take_while(|tok| matches!(tok, Tok::Comma)) + .count(); + let Some(Tok::Quoted(text)) = toks.get(self.pos + commas) else { + return Ok(out); + }; + if commas != 1 { + return Err(parse_error(format!( + "{} {clause} needs a comma between labels, as {clause} 'a', 'b'", + self.statement + ))); + } + out.push(text.clone().into_owned()); + self.pos += 2; + } } /// A `{...}` object literal, braces included. @@ -266,6 +297,7 @@ fn describe(tok: &Tok<'_>) -> String { Tok::Ident(text) => format!("identifier \"{text}\""), Tok::Object(_) => "an object literal".to_owned(), Tok::List(_) => "a [...] list".to_owned(), + Tok::Comma => "','".to_owned(), } } diff --git a/nodedb-sql/src/ddl_ast/graph_parse/entry.rs b/nodedb-sql/src/ddl_ast/graph_parse/entry.rs index 151946d4e..afdc536c9 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) + } else if upper.starts_with("GRAPH DELETE EDGES ") { + variants::parse_delete_edges(&toks, sql) + } 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,138 @@ 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_a_missing_comma_between_triples() { + let error = try_parse("GRAPH INSERT EDGES IN 'edges' VALUES ('a','b','L') ('c','d','L')") + .expect("input is graph DSL") + .expect_err("two triples without a comma are malformed"); + assert!( + error.to_string().contains("triples"), + "the error must name the tuple shape: {error}" + ); + } + + #[test] + fn batch_edge_rejects_bare_words_without_parentheses() { + try_parse("GRAPH DELETE EDGES IN 'edges' VALUES a b L") + .expect("input is graph DSL") + .expect_err("three bare words are not a parenthesized triple"); + } + + #[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. @@ -675,4 +817,113 @@ mod tests { let params = params_of("GRAPH RAG FUSION ON entities QUERY ARRAY[0.1]"); assert_eq!(params.bm25_field, None); } + + // ── Missing separators in lists ─────────────────────────────────────── + + /// Two quoted labels with no comma between them are one malformed list. + #[test] + fn a_label_list_without_a_comma_is_refused_on_traverse() { + let sql = "GRAPH TRAVERSE IN 'c' FROM 'x' DEPTH 2 LABEL 'a' 'b'"; + let result = try_parse(sql).expect("graph DSL"); + assert!( + result.is_err(), + "a LABEL list missing its comma must be refused, got {result:?}" + ); + } + + /// Two quoted labels with no comma between them are one malformed list. + #[test] + fn a_label_list_without_a_comma_is_refused_on_neighbors() { + let sql = "GRAPH NEIGHBORS IN 'c' OF 'x' LABEL 'a' 'b' DIRECTION both"; + let result = try_parse(sql).expect("graph DSL"); + assert!( + result.is_err(), + "a LABEL list missing its comma must be refused, got {result:?}" + ); + } + + /// Two quoted labels with no comma between them are one malformed list. + #[test] + fn a_label_list_without_a_comma_is_refused_on_path() { + let sql = "GRAPH PATH IN 'c' FROM 'x' TO 'y' MAX_DEPTH 3 LABEL 'a' 'b'"; + let result = try_parse(sql).expect("graph DSL"); + assert!( + result.is_err(), + "a LABEL list missing its comma must be refused, got {result:?}" + ); + } + + /// Two quoted node labels with no comma between them are refused. + #[test] + fn a_label_set_without_a_comma_is_refused_on_graph_label() { + let sql = "GRAPH LABEL 'alice' AS 'Person' 'User'"; + let result = try_parse(sql).expect("graph DSL"); + assert!( + result.is_err(), + "an AS list missing its comma must be refused, got {result:?}" + ); + } + + /// An `IN (...)` list in `EDGE WHERE` needs commas between its values. + #[test] + fn an_edge_where_list_without_a_comma_is_refused() { + let sql = "GRAPH TRAVERSE IN 'c' FROM 'x' DEPTH 2 EDGE WHERE kind IN ('a' 'b')"; + let result = try_parse(sql).expect("graph DSL"); + assert!( + result.is_err(), + "an EDGE WHERE list missing its comma must be refused, got {result:?}" + ); + } + + /// The comma-separated `EDGE WHERE ... IN (...)` list is the accepted form. + #[test] + fn an_edge_where_list_with_commas_parses() { + let sql = "GRAPH TRAVERSE IN 'c' FROM 'x' DEPTH 2 EDGE WHERE kind IN ('a', 'b')"; + let result = try_parse(sql).expect("graph DSL"); + assert!( + result.is_ok(), + "a comma-separated list must parse: {result:?}" + ); + } + + /// The tokenizer keeps a comma as its own token. + #[test] + fn tokenizer_emits_a_comma_token() { + let toks = tokenizer::tokenize("a, b"); + assert!(matches!( + toks.as_slice(), + [ + tokenizer::Tok::Word("a"), + tokenizer::Tok::Comma, + tokenizer::Tok::Word("b") + ] + )); + } + + /// Two commas between labels are a malformed list. + #[test] + fn a_label_list_with_two_commas_is_refused() { + let sql = "GRAPH NEIGHBORS IN 'c' OF 'x' LABEL 'a',, 'b'"; + let result = try_parse(sql).expect("graph DSL"); + assert!( + result.is_err(), + "a doubled comma must be refused, got {result:?}" + ); + } + + /// A comma after the last label, before a clause keyword, is skipped. + #[test] + fn a_trailing_label_comma_before_a_clause_parses() { + let sql = "GRAPH NEIGHBORS IN 'c' OF 'x' LABEL 'a', DIRECTION out"; + let result = try_parse(sql).expect("graph DSL"); + assert!(result.is_ok(), "a trailing comma must parse: {result:?}"); + } + + /// A comma between two clauses is skipped as it was before commas were tokens. + #[test] + fn a_stray_comma_between_clauses_parses() { + let sql = "GRAPH NEIGHBORS IN 'c', OF 'x', DIRECTION out"; + let result = try_parse(sql).expect("graph DSL"); + assert!(result.is_ok(), "a stray comma must parse: {result:?}"); + } } diff --git a/nodedb-sql/src/ddl_ast/graph_parse/mod.rs b/nodedb-sql/src/ddl_ast/graph_parse/mod.rs index fcd635068..3afe5ea6c 100644 --- a/nodedb-sql/src/ddl_ast/graph_parse/mod.rs +++ b/nodedb-sql/src/ddl_ast/graph_parse/mod.rs @@ -17,7 +17,13 @@ //! 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. +//! +//! The tokenizer drops `(`, `)`, and `,`, so the batched edge statements check +//! the shape of their `VALUES` tail on the statement text (`batch_shape`): a +//! comma-separated list of parenthesized triples. A missing comma or a bare +//! word list is a parse error. +mod batch_shape; mod cursor; mod edge_predicate; mod entry; diff --git a/nodedb-sql/src/ddl_ast/graph_parse/tokenizer.rs b/nodedb-sql/src/ddl_ast/graph_parse/tokenizer.rs index 2a2d57683..65c38fbc2 100644 --- a/nodedb-sql/src/ddl_ast/graph_parse/tokenizer.rs +++ b/nodedb-sql/src/ddl_ast/graph_parse/tokenizer.rs @@ -14,6 +14,8 @@ pub(super) enum Tok<'a> { Object(&'a str), /// Bracket-balanced `[...]` content, brackets excluded. List(&'a str), + /// A `,` separator. + Comma, } /// `sql` with every comment replaced by one space, so no comment becomes a @@ -42,16 +44,21 @@ pub(super) fn blank_comments(sql: &str) -> Cow<'_, str> { /// Split graph DSL text into tokens. /// -/// `,`, `;`, `(` and `)` separate tokens and produce none, so `LABEL 'a', 'b'` -/// and `LABEL ('a', 'b')` read the same. A stray `]` is a word token, so the -/// clause parser reports it. +/// `,` produces a `Comma` token. `;`, `(` and `)` separate tokens and produce +/// none, so `LABEL 'a', 'b'` and `LABEL ('a', 'b')` read the same. A stray `]` +/// is a word token, so the clause parser reports it. pub(super) fn tokenize(sql: &str) -> Vec> { let bytes = sql.as_bytes(); let mut out = Vec::new(); let mut i = 0; while i < bytes.len() { let b = bytes[i]; - if b.is_ascii_whitespace() || b == b',' || b == b';' || b == b'(' || b == b')' { + if b == b',' { + out.push(Tok::Comma); + i += 1; + continue; + } + if b.is_ascii_whitespace() || b == b';' || b == b'(' || b == b')' { i += 1; continue; } diff --git a/nodedb-sql/src/ddl_ast/graph_parse/variants.rs b/nodedb-sql/src/ddl_ast/graph_parse/variants.rs index 58697687f..ed8af8d92 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,126 @@ pub(super) fn parse_insert_edge(toks: &[Tok<'_>]) -> Result], sql: &str) -> Result { + let (collection, edges) = parse_edge_batch(toks, sql, "GRAPH INSERT EDGES")?; + Ok(NodedbStatement::Graph(GraphStmt::GraphInsertEdges { + collection, + edges, + })) +} + +pub(super) fn parse_delete_edges(toks: &[Tok<'_>], sql: &str) -> Result { + let (collection, edges) = parse_edge_batch(toks, sql, "GRAPH DELETE EDGES")?; + Ok(NodedbStatement::Graph(GraphStmt::GraphDeleteEdges { + collection, + edges, + })) +} + +/// Shared body of the batch parsers. +/// +/// Grammar: `IN '' VALUES ('','','