From 15956dfbd0fb67130dfb42a7eb8be035b21d03ee Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 24 Oct 2026 09:33:10 +0530 Subject: [PATCH 01/10] feat(parser): :limit N and :limit $slot, PRS-115 (#306) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0169pNQsLcCEn48te9X9WuZW --- docs/ERROR_REFERENCE.md | 16 +++++++ src/error.rs | 9 ++++ src/query/datalog/executor.rs | 2 + src/query/datalog/parser.rs | 85 +++++++++++++++++++++++++++++++++++ src/query/datalog/types.rs | 12 +++++ tests/error_codes_prs_test.rs | 14 ++++++ 6 files changed, 138 insertions(+) diff --git a/docs/ERROR_REFERENCE.md b/docs/ERROR_REFERENCE.md index 8bd05bbad..0ff160a1f 100644 --- a/docs/ERROR_REFERENCE.md +++ b/docs/ERROR_REFERENCE.md @@ -135,6 +135,7 @@ with no `CodedError` anywhere in its chain. | PRS-112 | Empty query section | Parser | | PRS-113 | Unknown query keyword | Parser | | PRS-114 | Rule head argument is not a variable | Parser | +| PRS-115 | Query option is not a non-negative integer | Parser | | QRY-001 | Invalid entity | Query Execution | | QRY-002 | Attribute must be a keyword | Query Execution | | QRY-003 | Cannot transact a pseudo-attribute | Query Execution | @@ -2135,6 +2136,21 @@ with `PRS-047`. ``` *`0` is not a variable* +### PRS-115 Query option is not a non-negative integer + +**Error text**: `` `{}` must be a non-negative integer or a bind slot, got {} `` + +**Cause**: The value after `:limit` is not an integer `>= 0` or a `$slot`, or a prepared query's `:limit` slot was bound to a negative integer (#306). + +**Resolution**: +- Give `:limit` a whole number of rows, `0` or more, or a bind slot filled with `BindValue::Val(Value::Integer(n))`. + +**Example**: +```datalog +(query [:find ?e :where [?e :person/name ?n] :limit -1]) +``` +*`-1` is negative* + --- ## QRY — Query Execution Errors diff --git a/src/error.rs b/src/error.rs index 57a5d4294..ffe1b6aa1 100644 --- a/src/error.rs +++ b/src/error.rs @@ -161,6 +161,7 @@ pub(crate) enum ErrorCode { Prs112, Prs113, Prs114, + Prs115, Qry001, Qry002, Qry003, @@ -1005,6 +1006,12 @@ pub(crate) const REGISTRY: &[(ErrorCode, &str, &str, ErrorCategory)] = &[ "Rule head argument must be a variable, got {}", ErrorCategory::Parser, ), + ( + ErrorCode::Prs115, + "PRS-115", + "`{}` must be a non-negative integer or a bind slot, got {}", + ErrorCategory::Parser, + ), ( ErrorCode::Qry001, "QRY-001", @@ -2270,6 +2277,7 @@ mod tests { | ErrorCode::Prs112 | ErrorCode::Prs113 | ErrorCode::Prs114 + | ErrorCode::Prs115 | ErrorCode::Qry001 | ErrorCode::Qry002 | ErrorCode::Qry003 @@ -2532,6 +2540,7 @@ mod tests { ErrorCode::Prs112, ErrorCode::Prs113, ErrorCode::Prs114, + ErrorCode::Prs115, ErrorCode::Qry001, ErrorCode::Qry002, ErrorCode::Qry003, diff --git a/src/query/datalog/executor.rs b/src/query/datalog/executor.rs index 9e7f06def..a3ac5fabc 100644 --- a/src/query/datalog/executor.rs +++ b/src/query/datalog/executor.rs @@ -1446,6 +1446,7 @@ mod tests { with_vars: Vec::new(), max_derived_facts: None, max_results: None, + limit: None, })) .unwrap(); @@ -1504,6 +1505,7 @@ mod tests { with_vars: Vec::new(), max_derived_facts: None, max_results: None, + limit: None, })) .unwrap(); diff --git a/src/query/datalog/parser.rs b/src/query/datalog/parser.rs index 6ba88dcfb..ca8f286e6 100644 --- a/src/query/datalog/parser.rs +++ b/src/query/datalog/parser.rs @@ -787,6 +787,7 @@ fn parse_query(elements: &[EdnValue]) -> Result { let mut query_valid_at: Option = None; let mut query_max_derived_facts: Option = None; let mut query_max_results: Option = None; + let mut query_limit: Option = None; let mut i = 0; while i < query_vector.len() { @@ -908,6 +909,28 @@ fn parse_query(elements: &[EdnValue]) -> Result { i += 1; continue; } + ":limit" => { + current_clause = None; + if query_limit.is_some() { + bail_coded!(ErrorCode::Int012, ":limit"); + } + i += 1; + if !value_at(i) { + bail_coded!(ErrorCode::Int013, ":limit"); + } + let val = query_vector + .get(i) + .ok_or_else(|| err_coded!(ErrorCode::Int013, ":limit"))?; + query_limit = Some(match val { + EdnValue::Integer(n) if *n >= 0 => { + Count::Literal(usize::try_from(*n).unwrap_or(usize::MAX)) + } + EdnValue::BindSlot(name) => Count::Slot(name.clone()), + other => bail_coded!(ErrorCode::Prs115, ":limit", format!("{other:?}")), + }); + i += 1; + continue; + } ":with" => { // Collect ?-prefixed symbols until the next keyword or end of vector if !value_at(i + 1) { @@ -1072,6 +1095,7 @@ fn parse_query(elements: &[EdnValue]) -> Result { query.with_vars = with_vars; query.max_derived_facts = query_max_derived_facts; query.max_results = query_max_results; + query.limit = query_limit; Ok(DatalogCommand::Query(query)) } @@ -2916,6 +2940,67 @@ mod tests { ); } + fn parse_limit(input: &str) -> Result> { + match parse_datalog_command(input)? { + DatalogCommand::Query(q) => Ok(q.limit), + _ => panic!("expected Query"), + } + } + + fn code_of(input: &str) -> &'static str { + let e = parse_datalog_command(input).expect_err("should be rejected"); + crate::MinigrafError::from(e).code() + } + + #[test] + fn limit_literal_zero_and_slot() { + let q = |tail: &str| format!("(query [:find ?x :where [?x :a ?y] {tail}])"); + assert_eq!( + parse_limit(&q(":limit 10")).unwrap(), + Some(Count::Literal(10)) + ); + assert_eq!( + parse_limit(&q(":limit 0")).unwrap(), + Some(Count::Literal(0)) + ); + assert_eq!( + parse_limit(&q(":limit $n")).unwrap(), + Some(Count::Slot("n".to_string())) + ); + assert_eq!(parse_limit(&q("")).unwrap(), None); + } + + #[test] + fn limit_in_any_section_order() { + for input in [ + "(query [:limit 3 :find ?x :where [?x :a ?y]])", + "(query [:find ?x :limit 3 :where [?x :a ?y]])", + "(query [:find ?x :where [?x :a ?y] :limit 3])", + "(query [:find ?x :where [?x :a ?y] :limit 3 :as-of 1])", + ] { + assert_eq!( + parse_limit(input).unwrap(), + Some(Count::Literal(3)), + "{input}" + ); + } + } + + #[test] + fn limit_errors() { + let q = |tail: &str| format!("(query [:find ?x :where [?x :a ?y] {tail}])"); + assert_eq!(code_of(&q(":limit 1 :limit 2")), "INT-012"); + assert_eq!(code_of(&q(":limit")), "INT-013"); + assert_eq!( + code_of("(query [:find ?x :limit :where [?x :a ?y]])"), + "INT-013" + ); + assert_eq!(code_of(&q(":limit :ten")), "INT-013"); + for bad in ["-1", "1.5", "\"ten\"", "?x", "nil", "true", "[1]"] { + assert_eq!(code_of(&q(&format!(":limit {bad}"))), "PRS-115", "{bad}"); + } + } + #[test] fn test_parse_max_derived_facts_non_integer_rejected() { let input = r#"(query [:find ?x :where [?x :a :b] :max-derived-facts "a lot"])"#; diff --git a/src/query/datalog/types.rs b/src/query/datalog/types.rs index 531f002ca..f4e41c436 100644 --- a/src/query/datalog/types.rs +++ b/src/query/datalog/types.rs @@ -557,6 +557,16 @@ pub struct DatalogQuery { /// Per-query `:max-results`, a deprecated alias of `:max-derived-facts`; /// with both, the lower applies (#553). It never limited returned rows. pub max_results: Option, + /// `:limit`: the most rows the query returns (#306). + pub limit: Option, +} + +/// The value of `:limit` (#306): a literal, or a bind slot a prepared query +/// fills. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Count { + Literal(usize), + Slot(String), } impl DatalogQuery { @@ -569,6 +579,7 @@ impl DatalogQuery { with_vars: Vec::new(), max_derived_facts: None, max_results: None, + limit: None, } } @@ -583,6 +594,7 @@ impl DatalogQuery { with_vars: Vec::new(), max_derived_facts: None, max_results: None, + limit: None, } } diff --git a/tests/error_codes_prs_test.rs b/tests/error_codes_prs_test.rs index 0cce7ddfc..c0ffe144c 100644 --- a/tests/error_codes_prs_test.rs +++ b/tests/error_codes_prs_test.rs @@ -809,6 +809,19 @@ fn prs_114_rule_head_argument_not_a_variable() { assert_execute_code("(rule [(r ?a 0) [?a :edge ?b]])", "PRS-114"); } +#[test] +fn prs_115_limit_not_a_non_negative_integer() { + assert_code_and_message( + "(query [:find ?a :where [?a :edge ?b] :limit -1])", + "PRS-115", + ":limit", + ); + assert_execute_code( + "(query [:find ?a :where [?a :edge ?b] :limit 2.5])", + "PRS-115", + ); +} + #[test] fn prs_105_rule_name_is_a_variable_or_an_operator() { assert_execute_code("(rule [(?r ?a) [?a :edge ?b]])", "PRS-105"); @@ -824,6 +837,7 @@ fn prs_061_value_after_a_query_option() { r#"(query [:find ?a :valid-at "2024-01-01T00:00:00Z" ?b :where [?a :edge ?b]])"#, "(query [:find ?a :max-results 5 ?b :where [?a :edge ?b]])", "(query [:find ?a :where [?a :edge ?b] :max-derived-facts 5 [?b :edge ?c]])", + "(query [:find ?a :limit 5 ?b :where [?a :edge ?b]])", ] { assert_execute_code(q, "PRS-061"); } From b37104f125e87b34596e362a71ac1155ce2f92f9 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 24 Oct 2026 09:33:33 +0530 Subject: [PATCH 02/10] fix: PRS-115 text without backticks (doc check) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0169pNQsLcCEn48te9X9WuZW --- docs/ERROR_REFERENCE.md | 2 +- src/error.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/ERROR_REFERENCE.md b/docs/ERROR_REFERENCE.md index 0ff160a1f..ef67a4b1b 100644 --- a/docs/ERROR_REFERENCE.md +++ b/docs/ERROR_REFERENCE.md @@ -2138,7 +2138,7 @@ with `PRS-047`. ### PRS-115 Query option is not a non-negative integer -**Error text**: `` `{}` must be a non-negative integer or a bind slot, got {} `` +**Error text**: `{} must be a non-negative integer or a bind slot, got {}` **Cause**: The value after `:limit` is not an integer `>= 0` or a `$slot`, or a prepared query's `:limit` slot was bound to a negative integer (#306). diff --git a/src/error.rs b/src/error.rs index ffe1b6aa1..9ae62a8e0 100644 --- a/src/error.rs +++ b/src/error.rs @@ -1009,7 +1009,7 @@ pub(crate) const REGISTRY: &[(ErrorCode, &str, &str, ErrorCategory)] = &[ ( ErrorCode::Prs115, "PRS-115", - "`{}` must be a non-negative integer or a bind slot, got {}", + "{} must be a non-negative integer or a bind slot, got {}", ErrorCategory::Parser, ), ( From 979b0b03f204c5685e96ce464ab08e66f8cc1231 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 24 Oct 2026 09:34:31 +0530 Subject: [PATCH 03/10] feat(prepared): :limit $slot takes Val(Integer) >= 0 (#306) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0169pNQsLcCEn48te9X9WuZW --- src/query/datalog/prepared.rs | 83 ++++++++++++++++++++++++++++++++++- 1 file changed, 82 insertions(+), 1 deletion(-) diff --git a/src/query/datalog/prepared.rs b/src/query/datalog/prepared.rs index 5108230b7..e9817a32f 100644 --- a/src/query/datalog/prepared.rs +++ b/src/query/datalog/prepared.rs @@ -7,7 +7,7 @@ use crate::query::datalog::executor::{DatalogExecutor, QueryResult}; use crate::query::datalog::functions::FunctionRegistry; use crate::query::datalog::rules::RuleRegistry; use crate::query::datalog::types::{ - AsOf, AttributeSpec, DatalogCommand, DatalogQuery, EdnValue, Expr, ValidAt, WhereClause, + AsOf, AttributeSpec, Count, DatalogCommand, DatalogQuery, EdnValue, Expr, ValidAt, WhereClause, }; use anyhow::Result; use std::collections::{HashMap, HashSet}; @@ -245,6 +245,9 @@ fn collect_slot_names(query: &DatalogQuery) -> Vec { if let Some(ValidAt::Slot(name)) = &query.valid_at { names.insert(name.clone()); } + if let Some(Count::Slot(name)) = &query.limit { + names.insert(name.clone()); + } collect_slots_from_clauses(&query.where_clauses, &mut names); let mut result: Vec = names.into_iter().collect(); result.sort(); @@ -322,6 +325,10 @@ fn substitute( query.valid_at = Some(resolve_valid_at_slot(name, bindings)?); } + if let Some(Count::Slot(name)) = &query.limit { + query.limit = Some(resolve_count_slot(name, ":limit", bindings)?); + } + for clause in &mut query.where_clauses { substitute_where_clause(clause, bindings)?; } @@ -471,6 +478,28 @@ fn resolve_as_of_slot(name: &str, bindings: &HashMap<&str, &BindValue>) -> Resul } } +/// A `:limit` slot (#306): `Val(Integer(n))` with `n >= 0`. +fn resolve_count_slot( + name: &str, + option: &str, + bindings: &HashMap<&str, &BindValue>, +) -> Result { + match bindings.get(name) { + Some(BindValue::Val(Value::Integer(n))) if *n >= 0 => { + Ok(Count::Literal(usize::try_from(*n).unwrap_or(usize::MAX))) + } + Some(BindValue::Val(Value::Integer(n))) => bail_coded!(ErrorCode::Prs115, option, n), + Some(other) => bail_coded!( + ErrorCode::Int035, + name, + option, + "Val(Integer)", + bind_value_type_name(other) + ), + None => bail_coded!(ErrorCode::Int033, name), + } +} + fn resolve_valid_at_slot(name: &str, bindings: &HashMap<&str, &BindValue>) -> Result { match bindings.get(name) { Some(BindValue::Timestamp(t)) => Ok(ValidAt::Timestamp(*t)), @@ -697,6 +726,58 @@ mod tests { } } + fn make_query_limit_slot() -> DatalogQuery { + let mut q = make_query_as_of_slot(); + q.as_of = None; + q.limit = Some(Count::Slot("n".to_string())); + q + } + + fn limit_code(bv: Option) -> &'static str { + let q = make_query_limit_slot(); + let bindings: HashMap<&str, &BindValue> = bv.iter().map(|b| ("n", b)).collect(); + let e = substitute(&q, &bindings).expect_err("should be rejected"); + crate::MinigrafError::from(e).code() + } + + #[test] + fn limit_slot_is_collected_and_rejected_unbound() { + let q = make_query_limit_slot(); + assert_eq!(collect_slot_names(&q), vec!["n".to_string()]); + let e = reject_unbound_slots(&q).expect_err("unbound"); + assert_eq!(crate::MinigrafError::from(e).code(), "API-010"); + } + + #[test] + fn limit_slot_takes_a_non_negative_integer() { + let q = make_query_limit_slot(); + for n in [0, 1, 7] { + let bv = BindValue::Val(Value::Integer(n)); + let bindings: HashMap<&str, &BindValue> = [("n", &bv)].into(); + let filled = substitute(&q, &bindings).unwrap(); + assert_eq!(filled.limit, Some(Count::Literal(n as usize))); + } + // A literal is kept. + let mut lit = make_query_limit_slot(); + lit.limit = Some(Count::Literal(4)); + let filled = substitute(&lit, &HashMap::new()).unwrap(); + assert_eq!(filled.limit, Some(Count::Literal(4))); + } + + #[test] + fn limit_slot_errors() { + assert_eq!( + limit_code(Some(BindValue::Val(Value::Integer(-1)))), + "PRS-115" + ); + assert_eq!( + limit_code(Some(BindValue::Val(Value::Float(1.0)))), + "INT-035" + ); + assert_eq!(limit_code(Some(BindValue::TxCount(3))), "INT-035"); + assert_eq!(limit_code(None), "INT-033"); + } + #[test] fn test_substitute_as_of_counter() { let q = make_query_as_of_slot(); From f6a0edfbf55eb309e8de0d1aa1cf075791be9190 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 24 Oct 2026 09:35:56 +0530 Subject: [PATCH 04/10] feat(engine): Limit operator (#306) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0169pNQsLcCEn48te9X9WuZW --- src/query/engine/batch.rs | 11 ++ src/query/engine/filter.rs | 6 +- src/query/engine/limit.rs | 228 +++++++++++++++++++++++++++++++++++++ src/query/engine/mod.rs | 2 + src/query/engine/plan.rs | 3 + 5 files changed, 248 insertions(+), 2 deletions(-) create mode 100644 src/query/engine/limit.rs diff --git a/src/query/engine/batch.rs b/src/query/engine/batch.rs index 2cba1a3b1..e922f32c4 100644 --- a/src/query/engine/batch.rs +++ b/src/query/engine/batch.rs @@ -37,6 +37,17 @@ impl Batch { self.len = keep.iter().filter(|&&k| k).count(); } + /// Keep the rows `skip..skip + take`. + pub(crate) fn slice(&mut self, skip: usize, take: usize) { + let end = skip.saturating_add(take).min(self.len); + let skip = skip.min(end); + for col in &mut self.cols { + col.truncate(end); + col.drain(..skip); + } + self.len = end - skip; + } + /// Set the column of `slot`: the next new slot, or one already filled. pub(crate) fn set_col(&mut self, slot: usize, col: Vec) { match self.cols.get_mut(slot) { diff --git a/src/query/engine/filter.rs b/src/query/engine/filter.rs index 2b27dd5bc..a3b2d05b9 100644 --- a/src/query/engine/filter.rs +++ b/src/query/engine/filter.rs @@ -33,6 +33,7 @@ impl Apply { Step::Distinct(_) => "Distinct", Step::Aggregate { .. } => "Aggregate", Step::Window { .. } => "Window", + Step::Limit { .. } => "Limit", }; Apply { child, @@ -131,13 +132,14 @@ fn apply(step: &Step, mut batch: Batch, ctx: &mut Ctx<'_>) -> Result { batch.set_col(*slot, out); } // Joins are operators of their own (`join::Join`, `anti::AntiJoin`, - // `union::Union`, `distinct::Distinct`). + // `union::Union`, `distinct::Distinct`, `limit::Limit`). Step::Join(_) | Step::Anti(_) | Step::Union(_) | Step::Distinct(_) | Step::Aggregate { .. } - | Step::Window { .. } => {} + | Step::Window { .. } + | Step::Limit { .. } => {} } Ok(batch) } diff --git a/src/query/engine/limit.rs b/src/query/engine/limit.rs new file mode 100644 index 000000000..799ad4cb2 --- /dev/null +++ b/src/query/engine/limit.rs @@ -0,0 +1,228 @@ +//! `Limit` (#473 spec §6.1, #306 spec §4.1): skips the first `offset` rows +//! of its child, passes on the next `limit`, then closes the child, so the +//! operators below it read nothing more. + +use super::batch::Batch; +use super::cell::Cell; +use super::{Counters, Ctx, Operator}; +use anyhow::Result; + +pub(crate) struct Limit { + child: Box, + offset: usize, + limit: Option, + /// Rows still to skip. + skip: usize, + /// Rows still to pass on; `None` passes on every row. + left: Option, + counters: Counters, +} + +impl Limit { + pub(crate) fn new(child: Box, offset: usize, limit: Option) -> Self { + Limit { + child, + offset, + limit, + skip: offset, + left: limit, + counters: Counters::new("Limit"), + } + } +} + +impl Operator for Limit { + fn next_batch(&mut self, ctx: &mut Ctx<'_>) -> Result> { + loop { + if self.left == Some(0) { + self.child.close(); + return Ok(None); + } + let Some(mut batch) = self.child.next_batch(ctx)? else { + return Ok(None); + }; + self.counters.rows_in += batch.len as u64; + let skip = self.skip.min(batch.len); + self.skip -= skip; + let take = self + .left + .map_or(batch.len - skip, |l| l.min(batch.len - skip)); + if let Some(l) = &mut self.left { + *l -= take; + } + batch.slice(skip, take); + if batch.len > 0 { + if self.left == Some(0) { + self.child.close(); + } + self.counters.out(&batch); + return Ok(Some(batch)); + } + } + } + + fn rewind(&mut self, seed: &[Cell]) { + self.skip = self.offset; + self.left = self.limit; + self.child.rewind(seed); + } + + fn close(&mut self) { + self.child.close(); + } + + fn counters(&self, out: &mut Vec) { + out.push(self.counters); + self.child.counters(out); + } +} + +#[cfg(all(test, not(target_arch = "wasm32")))] +mod tests { + use super::*; + use crate::graph::types::Value; + use crate::query::engine::cell::Scalar; + use crate::query::engine::{BATCH_ROWS, Stream}; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + + /// Rows `0..rows` in full batches; counts the rows it returns and its + /// closes. + struct Numbers { + rows: usize, + next: usize, + sent: Arc, + closes: Arc, + } + + impl Operator for Numbers { + fn next_batch(&mut self, _: &mut Ctx<'_>) -> Result> { + if self.next >= self.rows { + return Ok(None); + } + let mut batch = Batch::new(1); + while batch.len < BATCH_ROWS && self.next < self.rows { + batch.push(&mut [Some(Cell::Val(Scalar::Int(self.next as i64)))]); + self.next += 1; + } + self.sent.fetch_add(batch.len, Ordering::SeqCst); + Ok(Some(batch)) + } + + fn rewind(&mut self, _: &[Cell]) { + self.next = 0; + } + + fn close(&mut self) { + self.next = self.rows; + self.closes.fetch_add(1, Ordering::SeqCst); + } + + fn counters(&self, _: &mut Vec) {} + } + + struct Run { + rows: Vec, + sent: usize, + closes: usize, + } + + fn run(rows: usize, offset: usize, limit: Option) -> Run { + let (sent, closes) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0))); + let child = Numbers { + rows, + next: 0, + sent: sent.clone(), + closes: closes.clone(), + }; + let snapshot = crate::graph::storage::FactStorage::new() + .snapshot() + .unwrap(); + let mut stream = Stream::for_test( + snapshot, + Box::new(Limit::new(Box::new(child), offset, limit)), + ); + let mut out = Vec::new(); + while let Some(batch) = stream.next_rows().unwrap() { + assert!(!batch.is_empty(), "never an empty batch"); + for row in batch { + match row.as_slice() { + [Value::Integer(n)] => out.push(*n), + _ => panic!("one integer per row"), + } + } + } + Run { + rows: out, + sent: sent.load(Ordering::SeqCst), + closes: closes.load(Ordering::SeqCst), + } + } + + #[test] + fn passes_on_at_most_limit_rows() { + for n in [0, 1, 1023, 1024, 1025, 5000, 6000] { + let r = run(5000, 0, Some(n)); + assert_eq!( + r.rows, + (0..n.min(5000) as i64).collect::>(), + "limit {n}" + ); + } + let r = run(5000, 0, None); + assert_eq!(r.rows.len(), 5000, "no limit"); + } + + #[test] + fn stops_reading_once_it_has_its_rows() { + let r = run(5000, 0, Some(0)); + assert_eq!(r.sent, 0, "limit 0 pulls nothing"); + assert!(r.closes >= 1, "and closes the child"); + + let r = run(5000, 0, Some(10)); + assert_eq!(r.sent, BATCH_ROWS, "one batch for 10 rows"); + assert!(r.closes >= 1); + + let r = run(5000, 0, Some(BATCH_ROWS + 1)); + assert_eq!(r.sent, 2 * BATCH_ROWS, "two batches for one more row"); + } + + #[test] + fn skips_offset_rows_first() { + assert_eq!(run(5000, 3, Some(2)).rows, [3, 4]); + assert_eq!(run(5000, 1023, Some(3)).rows, [1023, 1024, 1025]); + assert_eq!(run(5000, 4999, Some(5)).rows, [4999]); + assert!(run(5000, 5000, Some(5)).rows.is_empty()); + assert!(run(5000, 9000, None).rows.is_empty()); + assert_eq!(run(5000, 4990, None).rows.len(), 10); + } + + #[test] + fn rewind_starts_again() { + let (sent, closes) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0))); + let child = Numbers { + rows: 100, + next: 0, + sent, + closes, + }; + let mut limit = Limit::new(Box::new(child), 2, Some(3)); + limit.skip = 0; + limit.left = Some(0); + limit.rewind(&[]); + assert_eq!( + (limit.skip, limit.left), + (2, Some(3)), + "offset and limit restored" + ); + let snapshot = crate::graph::storage::FactStorage::new() + .snapshot() + .unwrap(); + let mut stream = Stream::for_test(snapshot, Box::new(limit)); + let mut n = 0; + while let Some(b) = stream.next_rows().unwrap() { + n += b.len(); + } + assert_eq!(n, 3, "and the child read again from its start"); + } +} diff --git a/src/query/engine/mod.rs b/src/query/engine/mod.rs index 9709faf74..d16e088f5 100644 --- a/src/query/engine/mod.rs +++ b/src/query/engine/mod.rs @@ -11,6 +11,7 @@ mod distinct; mod filter; mod ids; mod join; +mod limit; mod meter; mod plan; mod project; @@ -319,6 +320,7 @@ pub(crate) fn build( plan::Step::Window { find, named, keep } => { Box::new(window::Window::new(root, find, named, keep, meter)) } + plan::Step::Limit { offset, limit } => Box::new(limit::Limit::new(root, offset, limit)), step => Box::new(filter::Apply::new(root, step)), }; } diff --git a/src/query/engine/plan.rs b/src/query/engine/plan.rs index befb68d0a..c0994bfe7 100644 --- a/src/query/engine/plan.rs +++ b/src/query/engine/plan.rs @@ -181,6 +181,9 @@ pub(crate) enum Step { named: Vec, keep: Vec, }, + /// After the last step: skips `offset` rows, then passes on at most + /// `limit` (#306). + Limit { offset: usize, limit: Option }, } /// A `not` / `not-join` body (#546 spec §4). From c15f4819878b817a02e9072257c4d8576833602c Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 24 Oct 2026 09:40:40 +0530 Subject: [PATCH 05/10] feat(engine): plan :limit after the last step; tests (#306) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0169pNQsLcCEn48te9X9WuZW --- src/cursor.rs | 43 ++++++++ src/query/engine/plan.rs | 16 ++- tests/limit_order_test.rs | 209 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 266 insertions(+), 2 deletions(-) create mode 100644 tests/limit_order_test.rs diff --git a/src/cursor.rs b/src/cursor.rs index e8bf1f316..7bf94b0b5 100644 --- a/src/cursor.rs +++ b/src/cursor.rs @@ -331,6 +331,49 @@ mod tests { assert_eq!(counters[1].rows_out, counters[0].rows_in); } + /// `:limit` closes the operators below it once it has its rows: the + /// scan reads one internal batch for 10 rows, and nothing for none + /// (#306 spec §4.1). + fn limit_stops_the_scan(db: &Minigraf) { + let q = |n: usize| format!("(query [:find ?e ?a :where [?e :person/age ?a] :limit {n}])"); + let mut cursor = db.query(&q(10)).unwrap(); + assert_eq!(cursor.next_batch(10).unwrap().expect("rows").len(), 10); + // Read before the stream ends: an ended cursor drops its counters. + let counters = cursor.counters(); + let ops: Vec<&str> = counters.iter().map(|c| c.op).collect(); + assert_eq!(ops, ["Project", "Limit", "Scan"]); + assert_eq!(counters[1].rows_out, 10); + let read = scan(&counters).rows_in; + assert!(read > 0, "the scan read keys"); + assert!( + read <= 2 * BATCH_ROWS as u64, + "one internal batch, not the attribute" + ); + assert_eq!(scan(&counters).batches, 1); + assert!(cursor.next_batch(10).unwrap().is_none(), "no more rows"); + + let mut cursor = db.query(&q(0)).unwrap(); + let mut stream = cursor.stream.take().expect("an engine stream"); + assert!(stream.next_rows().unwrap().is_none(), "no rows"); + assert_eq!(scan(&stream.counters()).rows_in, 0, "limit 0 reads nothing"); + } + + #[test] + fn limit_stops_the_scan_in_memory() { + let db = Minigraf::in_memory().unwrap(); + people(&db); + limit_stops_the_scan(&db); + } + + #[test] + fn limit_stops_the_scan_on_file() { + let dir = tempfile::tempdir().unwrap(); + let db = Minigraf::open(dir.path().join("limit.graph")).unwrap(); + people(&db); + db.checkpoint().unwrap(); + limit_stops_the_scan(&db); + } + /// Spec §10: a lazy cursor's answer is unchanged by writes and by /// checkpoints that free the pages it has yet to read. #[test] diff --git a/src/query/engine/plan.rs b/src/query/engine/plan.rs index c0994bfe7..f982e80f8 100644 --- a/src/query/engine/plan.rs +++ b/src/query/engine/plan.rs @@ -30,8 +30,8 @@ use crate::query::datalog::optimizer; use crate::query::datalog::rules::RuleRegistry; use crate::query::datalog::stratification::DependencyGraph; use crate::query::datalog::types::{ - AsOf, AttributeSpec, DatalogQuery, EdnValue, Expr, FindSpec, Pattern, PseudoAttr, UnaryOp, - ValidAt, WhereClause, + AsOf, AttributeSpec, Count, DatalogQuery, EdnValue, Expr, FindSpec, Pattern, PseudoAttr, + UnaryOp, ValidAt, WhereClause, }; use anyhow::Result; use std::collections::{HashMap, HashSet}; @@ -543,6 +543,18 @@ fn plan_with(query: &DatalogQuery, evaluated: Option>) -> Result< } let mut plan = b.finish()?; plan.find = find; + // After `finish`, which may still add the steps of a clause list with + // no pattern. + match &query.limit { + None => {} + Some(Count::Literal(n)) => plan.steps.push(Step::Limit { + offset: 0, + limit: Some(*n), + }), + // `execute` rejects an unbound slot (API-010); a prepared query + // fills it. + Some(Count::Slot(_)) => return Err(invariant("a :limit slot reached the planner")), + } plan.vars = query.find.iter().map(FindSpec::display_name).collect(); plan.as_of = as_of; Ok(plan) diff --git a/tests/limit_order_test.rs b/tests/limit_order_test.rs new file mode 100644 index 000000000..fbf8d107f --- /dev/null +++ b/tests/limit_order_test.rs @@ -0,0 +1,209 @@ +//! `:limit` (#306, spec `docs/superpowers/specs/2026-10-24-limit-order-by-design.md`). +//! +//! Without `:order-by`, `:limit n` returns `min(n, |answer|)` rows, and +//! they are a sub-bag of the answer the query gives without it: which rows is +//! unspecified (spec §1, decision 2). Each shape runs through +//! `Minigraf::execute`, a `Minigraf::query` cursor and a prepared `:limit $n`, +//! with clause reordering on and off (`minigraf::__reorder`). +#![cfg(all(feature = "test-hooks", not(target_arch = "wasm32")))] + +use minigraf::{BindValue, Minigraf, QueryResult, Value}; +use std::collections::HashMap; + +/// A value as text, without `Debug` (CodeQL `rust/cleartext-logging`). +fn key(v: &Value) -> String { + match v { + Value::String(s) => format!("s:{s}"), + Value::Integer(i) => format!("i:{i}"), + Value::Float(f) => format!("f:{:016x}", f.to_bits()), + Value::Boolean(b) => format!("b:{b}"), + Value::Ref(u) => format!("r:{}", u.as_hyphenated()), + Value::Keyword(k) => format!("k:{k}"), + Value::Null => "nil".to_string(), + } +} + +fn keys(rows: &[Vec]) -> Vec> { + rows.iter().map(|r| r.iter().map(key).collect()).collect() +} + +fn results(r: QueryResult) -> Vec> { + match r { + QueryResult::QueryResults { results, .. } => keys(&results), + _ => panic!("expected QueryResults"), + } +} + +/// Whether every row of `part` is in `whole` at least as many times. +fn is_sub_bag(part: &[Vec], whole: &[Vec]) -> bool { + let mut left: HashMap<&Vec, usize> = HashMap::new(); + for r in whole { + *left.entry(r).or_default() += 1; + } + part.iter().all(|r| match left.get_mut(r) { + Some(n) if *n > 0 => { + *n -= 1; + true + } + _ => false, + }) +} + +/// 50 people in five `:follows` chains, over two transactions, the first checkpointed on a file +/// database, so answers mix committed and pending facts. +fn people(db: &Minigraf, checkpoint: bool) { + let facts: String = (0..50) + .map(|i| { + format!( + "[:p{i} :name \"n{i}\"] [:p{i} :age {}] [:p{i} :status {}] [:p{i} :follows :p{}]", + i % 7, + if i % 3 == 0 { ":inactive" } else { ":active" }, + // Chains of 10, so `reach` stays small. + if i % 10 == 9 { i } else { i + 1 } + ) + }) + .collect(); + db.execute(&format!("(transact [{facts}])")).unwrap(); + if checkpoint { + db.checkpoint().unwrap(); + } + let more: String = (50..60) + .map(|i| { + format!( + "[:p{i} :name \"n{i}\"] [:p{i} :age {}] [:p{i} :status :active]", + i % 7 + ) + }) + .collect(); + db.execute(&format!("(transact [{more}])")).unwrap(); + db.execute("(retract [[:p1 :age 1]])").unwrap(); + db.execute("(rule [(reach ?x ?y) [?x :follows ?y]])") + .unwrap(); + db.execute("(rule [(reach ?x ?y) [?x :follows ?z] (reach ?z ?y)])") + .unwrap(); +} + +/// Query bodies: every shape `plan_with` builds before its `Limit`. +const SHAPES: &[&str] = &[ + ":find ?e ?a :where [?e :age ?a]", + ":find ?a :where [?e :age ?a]", + ":find ?n ?f :where [?e :name ?n] [?e :follows ?f]", + ":find ?e :where [?e :age ?a] (not [?e :status :inactive])", + ":find ?e :where (or [?e :status :inactive] [?e :age 3])", + ":find ?s (count ?e) :where [?e :status ?s]", + ":find (count ?e) :where [?e :age ?a]", + ":find (count ?e) :where [?e :no-such ?a]", + ":find ?e (rank :over (:order-by ?a)) :where [?e :age ?a]", + ":find ?y :where (reach :p0 ?y)", + ":find ?x ?y :where (reach ?x ?y)", + ":find ?e :as-of 1 :where [?e :age ?a]", + ":find ?e ?a :any-valid-time :where [?e :age ?a]", + ":find ?x :where [(+ 1 2) ?x]", +]; + +fn check(db: &Minigraf) { + for body in SHAPES { + for reorder in [true, false] { + let _g = minigraf::__reorder(reorder); + let answer = results(db.execute(&format!("(query [{body}])")).expect(body)); + let total = answer.len(); + let prepared = db + .prepare(&format!("(query [{body} :limit $n])")) + .expect(body); + for n in [0, 1, 2, total.saturating_sub(1), total, total + 5] { + let q = format!("(query [{body} :limit {n}])"); + let executed = results(db.execute(&q).expect(&q)); + + let mut cursor = db.query(&q).expect(&q); + let mut pulled = Vec::new(); + while let Some(batch) = cursor.next_batch(1).expect("batch") { + pulled.extend(keys(batch.rows())); + } + + let bind = [("n", BindValue::Val(Value::Integer(n as i64)))]; + let from_prepared = results(prepared.execute(&bind).expect(&q)); + let mut cursor = prepared.query(&bind).expect(&q); + let mut prepared_pulled = Vec::new(); + while let Some(batch) = cursor.next_batch(3).expect("batch") { + prepared_pulled.extend(keys(batch.rows())); + } + + for (how, rows) in [ + ("execute", &executed), + ("cursor", &pulled), + ("prepared", &from_prepared), + ("prepared cursor", &prepared_pulled), + ] { + assert_eq!(rows.len(), n.min(total), "{q} via {how}, reorder {reorder}"); + assert!( + is_sub_bag(rows, &answer), + "{q} via {how}, reorder {reorder}: rows not in the answer" + ); + } + } + } + } +} + +#[test] +fn limit_in_memory() { + let db = Minigraf::in_memory().unwrap(); + people(&db, false); + check(&db); +} + +#[test] +fn limit_on_committed_and_pending_facts() { + let dir = tempfile::tempdir().unwrap(); + let db = Minigraf::open(dir.path().join("limit.graph")).unwrap(); + people(&db, true); + check(&db); +} + +/// A `count`-only query with no input gives its zero row; `:limit 0` +/// removes it. +#[test] +fn limit_applies_to_the_zero_row() { + let db = Minigraf::in_memory().unwrap(); + people(&db, false); + let q = |n: usize| format!("(query [:find (count ?e) :where [?e :no-such ?a] :limit {n}])"); + assert_eq!(results(db.execute(&q(1)).unwrap()), [["i:0"]]); + assert!(results(db.execute(&q(0)).unwrap()).is_empty()); +} + +/// A write transaction's own staged facts count toward its answer. +#[test] +fn limit_in_a_write_transaction() { + let db = Minigraf::in_memory().unwrap(); + people(&db, false); + let mut tx = db.begin_write().unwrap(); + tx.execute("(transact [[:q0 :age 100] [:q1 :age 100] [:q2 :age 100]])") + .unwrap(); + let all = results( + tx.execute("(query [:find ?e :where [?e :age 100]])") + .unwrap(), + ); + assert_eq!(all.len(), 3); + let some = results( + tx.execute("(query [:find ?e :where [?e :age 100] :limit 2])") + .unwrap(), + ); + assert_eq!(some.len(), 2); + assert!(is_sub_bag(&some, &all)); + tx.commit().unwrap(); +} + +/// An unbound `:limit $n` in `execute` is `API-010`; a bad bind value is +/// rejected at execute. +#[test] +fn limit_slot_errors() { + let db = Minigraf::in_memory().unwrap(); + people(&db, false); + let q = "(query [:find ?e :where [?e :age ?a] :limit $n])"; + assert_eq!(db.execute(q).expect_err("unbound").code(), "API-010"); + let p = db.prepare(q).unwrap(); + let code = |b: BindValue| p.execute(&[("n", b)]).expect_err("rejected").code(); + assert_eq!(code(BindValue::Val(Value::Integer(-2))), "PRS-115"); + assert_eq!(code(BindValue::Val(Value::String("2".into()))), "INT-035"); + assert_eq!(code(BindValue::TxCount(2)), "INT-035"); +} From 7c469bb75d74be853248be0063daf62911a85fb1 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 24 Oct 2026 09:44:45 +0530 Subject: [PATCH 06/10] test: :limit grammar, verifier, 10M acceptance, envelope shape (#306) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0169pNQsLcCEn48te9X9WuZW --- .github/workflows/smoke.yml | 18 ++++ docs/GRAMMAR.md | 4 +- examples/perf_envelope.rs | 6 ++ src/query/datalog/parser/verifier.rs | 19 ++++- tests/grammar/grammar.pest | 4 +- .../invalid/semantic/query_limit_negative.edn | 3 + .../invalid/semantic/query_limit_twice.edn | 4 + .../invalid/syntax/query_limit_missing.edn | 3 + .../invalid/syntax/query_limit_string.edn | 3 + tests/grammar/valid/query_limit.edn | 3 + tests/grammar/valid/query_limit_slot.edn | 3 + tests/grammar/valid/query_limit_zero.edn | 3 + tests/limit_acceptance_test.rs | 83 +++++++++++++++++++ 13 files changed, 151 insertions(+), 5 deletions(-) create mode 100644 tests/grammar/invalid/semantic/query_limit_negative.edn create mode 100644 tests/grammar/invalid/semantic/query_limit_twice.edn create mode 100644 tests/grammar/invalid/syntax/query_limit_missing.edn create mode 100644 tests/grammar/invalid/syntax/query_limit_string.edn create mode 100644 tests/grammar/valid/query_limit.edn create mode 100644 tests/grammar/valid/query_limit_slot.edn create mode 100644 tests/grammar/valid/query_limit_zero.edn create mode 100644 tests/limit_acceptance_test.rs diff --git a/.github/workflows/smoke.yml b/.github/workflows/smoke.yml index 748bdc68c..1a70b2927 100644 --- a/.github/workflows/smoke.yml +++ b/.github/workflows/smoke.yml @@ -66,6 +66,24 @@ jobs: env: RUST_BACKTRACE: 1 + # #306: `:limit 10` over 10M facts stops once it has its rows, in a + # 64 KiB query memory limit. + limit-acceptance: + name: Limit acceptance + runs-on: ubuntu-latest + timeout-minutes: 30 + + steps: + - uses: actions/checkout@v4 + + - name: Install Rust stable + uses: dtolnay/rust-toolchain@stable + + - name: Run the acceptance test + run: cargo test --profile bench --test limit_acceptance_test -- --ignored --nocapture + env: + RUST_BACKTRACE: 1 + # #550: a 1M-fact transact holds at most one chunk of facts and commits as # a checkpoint; peak RSS is printed. test-hooks: the statement reader's # high-water mark. diff --git a/docs/GRAMMAR.md b/docs/GRAMMAR.md index c0eb83b1a..7c5e4a8a3 100644 --- a/docs/GRAMMAR.md +++ b/docs/GRAMMAR.md @@ -19,6 +19,7 @@ kw-valid-at ::= ":valid-at" (* not followed by word-cont *) (* token * kw-any-valid-time ::= ":any-valid-time" (* not followed by word-cont *) (* token *) kw-max-derived-facts ::= ":max-derived-facts" (* not followed by word-cont *) (* token *) kw-max-results ::= ":max-results" (* not followed by word-cont *) (* token *) +kw-limit ::= ":limit" (* not followed by word-cont *) (* token *) kw-over ::= ":over" (* not followed by word-cont *) (* token *) kw-partition-by ::= ":partition-by" (* not followed by word-cont *) (* token *) kw-order-by ::= ":order-by" (* not followed by word-cont *) (* token *) @@ -73,7 +74,7 @@ query-cmd ::= "(" "query" query-vector ")" query-vector ::= "[" query-section+ "]" query-section ::= find-section | where-section | as-of-section | valid-at-section | any-valid-time-section | with-section | max-derived-facts-section - | max-results-section + | max-results-section | limit-section (* ── FIND SECTION ── *) find-section ::= kw-find find-spec+ @@ -116,6 +117,7 @@ any-valid-time-section ::= kw-any-valid-time with-section ::= kw-with variable+ max-derived-facts-section ::= kw-max-derived-facts integer-lit max-results-section ::= kw-max-results integer-lit +limit-section ::= kw-limit (integer-lit | bind-slot) (* ── RULE ── *) rule-cmd ::= "(" "rule" rule-vector ")" diff --git a/examples/perf_envelope.rs b/examples/perf_envelope.rs index e842caa8a..983344819 100644 --- a/examples/perf_envelope.rs +++ b/examples/perf_envelope.rs @@ -660,6 +660,12 @@ fn shapes(e: usize, tx_half: u64, tx_half_time: &str, hot_tx_half: u64) -> Vec Vec { a } -const TOKENS: [&str; 32] = [ +const TOKENS: [&str; 33] = [ "(", ")", "[", @@ -88,6 +88,7 @@ const TOKENS: [&str; 32] = [ ":any-valid-time", ":over", ":order-by", + ":limit", ":valid-from", ":foo", "?x", @@ -205,6 +206,17 @@ const SEMANTIC: &[Semantic] = &[ code: "PRS-084", witness: "(query [:find ?x (sum ?y :over ()) :where [?x :a ?y]])", }, + // parse_query: a query option given twice (`query_section+` allows any repeat); + // `:limit 1 :limit 1` is a 4-token filler (#306). + Semantic { + code: "INT-012", + witness: "(query [:find ?x :where [?x :a ?y] :limit 1 :limit 1])", + }, + // parse_query: a negative `:limit` (`integer_lit` takes a sign, #306). + Semantic { + code: "PRS-115", + witness: "(query [:find ?x :where [?x :a ?y] :limit -1])", + }, // parse_rule: a rule body with no pattern or rule call (`rule_vector` allows // `where_clause*`). Semantic { @@ -497,8 +509,9 @@ fn filler_counts_match_the_spec() { } n }; - assert_eq!(count(3), 18_516); - assert_eq!(count(4), 487_678); + // 33 tokens since `:limit` (#306); the #643 spec counted 32 (18,516 and 487,678). + assert_eq!(count(3), 20_686); + assert_eq!(count(4), 565_267); } #[test] diff --git a/tests/grammar/grammar.pest b/tests/grammar/grammar.pest index 5622ab761..5cbc53748 100644 --- a/tests/grammar/grammar.pest +++ b/tests/grammar/grammar.pest @@ -19,6 +19,7 @@ kw_valid_at = @{ ":valid-at" ~ !word_cont } kw_any_valid_time = @{ ":any-valid-time" ~ !word_cont } kw_max_derived_facts = @{ ":max-derived-facts" ~ !word_cont } kw_max_results = @{ ":max-results" ~ !word_cont } +kw_limit = @{ ":limit" ~ !word_cont } kw_over = @{ ":over" ~ !word_cont } kw_partition_by = @{ ":partition-by" ~ !word_cont } kw_order_by = @{ ":order-by" ~ !word_cont } @@ -123,7 +124,7 @@ query_vector = { "[" ~ query_section+ ~ "]" } query_section = { find_section | where_section | as_of_section | valid_at_section | any_valid_time_section | with_section | - max_derived_facts_section | max_results_section + max_derived_facts_section | max_results_section | limit_section } // ── FIND SECTION ────────────────────────────────────────────────────────── @@ -197,6 +198,7 @@ any_valid_time_section = { kw_any_valid_time } with_section = { kw_with ~ variable+ } max_derived_facts_section = { kw_max_derived_facts ~ integer_lit } max_results_section = { kw_max_results ~ integer_lit } +limit_section = { kw_limit ~ (integer_lit | bind_slot) } // ── RULE ────────────────────────────────────────────────────────────────── // (rule [(head ?args) body-clause ...]) diff --git a/tests/grammar/invalid/semantic/query_limit_negative.edn b/tests/grammar/invalid/semantic/query_limit_negative.edn new file mode 100644 index 000000000..63093e895 --- /dev/null +++ b/tests/grammar/invalid/semantic/query_limit_negative.edn @@ -0,0 +1,3 @@ +(query [:find ?e + :where [?e :person/name ?n] + :limit -1]) diff --git a/tests/grammar/invalid/semantic/query_limit_twice.edn b/tests/grammar/invalid/semantic/query_limit_twice.edn new file mode 100644 index 000000000..0f60148d3 --- /dev/null +++ b/tests/grammar/invalid/semantic/query_limit_twice.edn @@ -0,0 +1,4 @@ +(query [:find ?e + :where [?e :person/name ?n] + :limit 1 + :limit 2]) diff --git a/tests/grammar/invalid/syntax/query_limit_missing.edn b/tests/grammar/invalid/syntax/query_limit_missing.edn new file mode 100644 index 000000000..dfc2e468c --- /dev/null +++ b/tests/grammar/invalid/syntax/query_limit_missing.edn @@ -0,0 +1,3 @@ +(query [:find ?e + :where [?e :person/name ?n] + :limit]) diff --git a/tests/grammar/invalid/syntax/query_limit_string.edn b/tests/grammar/invalid/syntax/query_limit_string.edn new file mode 100644 index 000000000..19dce4663 --- /dev/null +++ b/tests/grammar/invalid/syntax/query_limit_string.edn @@ -0,0 +1,3 @@ +(query [:find ?e + :where [?e :person/name ?n] + :limit "10"]) diff --git a/tests/grammar/valid/query_limit.edn b/tests/grammar/valid/query_limit.edn new file mode 100644 index 000000000..d7f970a13 --- /dev/null +++ b/tests/grammar/valid/query_limit.edn @@ -0,0 +1,3 @@ +(query [:find ?e + :where [?e :person/name ?n] + :limit 10]) diff --git a/tests/grammar/valid/query_limit_slot.edn b/tests/grammar/valid/query_limit_slot.edn new file mode 100644 index 000000000..261f6bc9b --- /dev/null +++ b/tests/grammar/valid/query_limit_slot.edn @@ -0,0 +1,3 @@ +(query [:find ?e + :where [?e :person/name ?n] + :limit $n]) diff --git a/tests/grammar/valid/query_limit_zero.edn b/tests/grammar/valid/query_limit_zero.edn new file mode 100644 index 000000000..60e7cadd7 --- /dev/null +++ b/tests/grammar/valid/query_limit_zero.edn @@ -0,0 +1,3 @@ +(query [:find ?e + :where [?e :person/name ?n] + :limit 0]) diff --git a/tests/limit_acceptance_test.rs b/tests/limit_acceptance_test.rs new file mode 100644 index 000000000..760d0e129 --- /dev/null +++ b/tests/limit_acceptance_test.rs @@ -0,0 +1,83 @@ +//! Acceptance test for `:limit` (#306 spec §5, moved from #473): over 10M +//! facts, `:limit 10` stops the query once it has its rows, so its time and +//! memory do not grow with the facts. +//! +//! Ignored by default; `.github/workflows/smoke.yml` runs it nightly: +//! +//! ```text +//! cargo test --profile bench --test limit_acceptance_test -- --ignored --nocapture +//! ``` +//! +//! Peak RSS comes from `/proc/self/status` (Linux only): the high-water mark +//! is reset just before the query. +// +// File-backed (Minigraf::open, tempfile): native only, like the other file tests. +#![cfg(not(target_arch = "wasm32"))] + +use minigraf::{Minigraf, OpenOptions, QueryResult}; + +const FACTS: usize = 10_000_000; +const CHUNK: usize = 100_000; +/// What the query may hold: a `Limit` holds nothing and a scan one batch of +/// ids. Holding even one id per fact would need megabytes. +const MEMORY_LIMIT: usize = 64 << 10; +/// Peak RSS the query may add: the reader's pages and one batch. +const MAX_RSS: usize = 32 << 20; +/// Far below a scan of 10M facts (seconds), well above one batch (well +/// under a millisecond on a desktop), for slow CI runners. +const MAX_TIME: std::time::Duration = std::time::Duration::from_millis(250); + +/// A `kB` field of `/proc/self/status`; 0 where it does not exist. +fn proc_status_kb(field: &str) -> usize { + std::fs::read_to_string("/proc/self/status") + .ok() + .and_then(|s| { + s.lines() + .find_map(|l| l.strip_prefix(field)?.strip_prefix(':').map(str::to_owned)) + }) + .and_then(|v| v.trim().trim_end_matches("kB").trim().parse().ok()) + .unwrap_or(0) +} + +#[test] +#[ignore] +fn limit_10_over_10m_facts() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("limit.graph"); + { + let db = Minigraf::open(&path).unwrap(); + for chunk in 0..FACTS / CHUNK { + let facts: String = (chunk * CHUNK..(chunk + 1) * CHUNK) + .map(|i| format!("[:e{i} :a {i}]")) + .collect(); + db.execute(&format!("(transact [{facts}])")).unwrap(); + db.checkpoint().unwrap(); + } + } + let db = OpenOptions::new() + .query_memory_limit(MEMORY_LIMIT) + .path(&path) + .open() + .unwrap(); + let query = "(query [:find ?e ?v :where [?e :a ?v] :limit 10])"; + let reset = std::fs::write("/proc/self/clear_refs", "5").is_ok(); + let before_kb = proc_status_kb("VmRSS"); + let start = std::time::Instant::now(); + let result = db.execute(query).unwrap(); + let elapsed = start.elapsed(); + let peak_kb = proc_status_kb("VmHWM"); + let added = peak_kb.saturating_sub(before_kb) * 1024; + eprintln!( + "limit_10_over_10m_facts: {elapsed:.2?}; peak RSS {} MiB, {} KiB added by the query", + peak_kb / 1024, + added / 1024 + ); + let QueryResult::QueryResults { results, .. } = result else { + panic!("expected query results"); + }; + assert_eq!(results.len(), 10, "ten rows"); + assert!(elapsed < MAX_TIME, "stops once it has its rows"); + if reset && peak_kb > 0 { + assert!(added < MAX_RSS, "holds no rows"); + } +} From cc40d796715d84e6acee4ffe4c435eec4904e531 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 24 Oct 2026 13:06:25 +0530 Subject: [PATCH 07/10] test(limit): build the 10M fixture with LogWriter (42 min via transact) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0169pNQsLcCEn48te9X9WuZW --- CHANGELOG.md | 1 + CLAUDE.md | 6 +++--- README.md | 1 + ROADMAP.md | 2 +- docs/TEST_COVERAGE.md | 6 ++++++ tests/limit_acceptance_test.rs | 33 ++++++++++++++++++++++----------- 6 files changed, 34 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3ca142f69..30b8afe6f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,6 +48,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **`:limit N` bounds the rows a query returns** (#306, PR 1 of 2). `[:find … :where … :limit 10]` returns at most 10 rows of the answer, and the engine stops reading once it has them: over 10M facts `:limit 10` reads one internal batch. Without `:order-by` (PR 2) which rows is unspecified, and may differ between plans; they are always rows of the unlimited answer. `:limit 0` returns no rows. It applies after aggregates and windows, to rule queries and to cursors, and a prepared query takes `:limit $n`, bound with `BindValue::Val(Value::Integer(n))`. A value that is not an integer `>= 0` or a slot fails with the new `PRS-115`; a repeated `:limit` with `INT-012` and a missing value with `INT-013`, as for the other query options. `:max-results` and `:max-derived-facts` still limit only the facts rules derive. - **Every query runs on a streaming engine** (#473: #544, #545, #546, #547, #548, #549, #551). The executor, which turned every live fact of each pattern's attribute into a `Fact` before matching, held each row's bindings in a map and copied the database into a `FactStorage` for every rule query, no longer runs queries: its query path, `PatternMatcher` and the old rule evaluators are gone (#551), and the release binary is 1,153 KiB instead of 1,286 KiB (CI's size check). A query is now a tree of operators that pass batches of up to 1,024 rows of ids, read from the index keys, and turn ids into values only for the rows they return (#530). A pattern reads the narrowest key range its constants allow (`[?e :a C]` reads the AVET range of `C`, #518); a later pattern is an `IndexNestedLoopJoin` when its entity is bound and a `HashJoin` otherwise, built on the side with the smaller estimate (with no shared variable, the smaller product, #575); per-fact pseudo-attributes read their source's metadata column (#594). `not` / `not-join` is an `AntiJoin` and `or` / `or-join` a `Union` and `Distinct`: a body or branch with the same rows for every outer row runs once into a hash table, any other runs per outer row from a one-row seed. Aggregates keep one entry per group (one set of accumulators when ungrouped), windows hold their rows, and rule queries derive each stratum's relations semi-naively, with the magic-set rewrite, then read them as patterns. `:as-of` with a time bounds the transaction counter when transaction times rise with it, and otherwise keeps the versions of transactions at or before the time. Cursors (`Minigraf::query`, `PreparedQuery::query`, `BrowserDb.query`) pull one batch at a time, and closing or dropping one stops the work; a cursor still answers from the database as it was when it opened. Every operator that keeps state counts it against `OpenOptions::query_memory_limit` (#549, below). Answers are the executor's, as a bag, except for the changes under Breaking changes: until #551 every query in the test suite was also run by the executor and the answers had to agree, and the answers of the suites that relied on that are recorded in `tests/answers/`. Measured with the #394 envelope at 1M facts, executor → engine, p50 and peak RSS (`docs/BENCHMARKS.md` has every shape at 100K, 1M and 10M): `[?e :name "…"]` 288 ms → 22 µs; an attribute scan of 166,666 rows drained with `next_batch(1000)` 58 → 7 MiB (458 → 7 MiB at 10M); `[:p :follows ?f] [?f :name ?n]` 796 ms → 50 µs; an entity-bound `not` 760 → 27 ms; `[:find (count ?v) :where [?e ?a ?v]]` 3.01 s → 457 ms and 728 → 7 MiB (6.6 GiB → 7 MiB at 10M); an `or-join` of two hashed branches 1.44 s → 80 ms; `(boss :p ?b)` 43.9 s → 2.14 ms and 2.6 GiB → 4 MiB; the whole transitive closure of `:manager` 51.0 s → 3.21 s and 4.7 GiB → 318 MiB; `:as-of` with a time, a point query, 1.65 s → 0.8 ms and 435 → 4 MiB (17.2 s and 3.6 GiB → 0.9 ms and 5 MiB at 10M). - **Transactions of any size on a file database, in bounded memory** (#473, part 8: #550). `transact` and `retract` are read one fact at a time: the whole command is never tokenized, parsed or materialised at once, and a statement reports the same error code as before whatever its size. A transaction becomes *large* when its facts would pass the new `OpenOptions::transaction_chunk_facts` (default 65,536, kept between 1 and 1,000,000): `Minigraf::execute` of a larger statement, or the `WriteTransaction` statement that takes its staged facts past it. A large transaction holds at most one chunk of facts in memory. It writes them a chunk at a time into copy-on-write trees that become the next generation, and commits as a checkpoint: no WAL entry, and the uncheckpointed facts of earlier transactions go into the same checkpoint. What a large transaction stores is what a small one stores, with two differences you can see: its `tx_id` is the time it became large, not its commit time (a default `:valid-from` takes that time too), and errors from different chunks of one statement come in chunk order. In a large `WriteTransaction` a rejected statement still leaves the transaction usable and the earlier statements' facts as they were; a fact too large to store is rejected at its statement (`WAL-003`) rather than at `commit`; a handle that has failed is refused when the transaction becomes large (`STG-046`) rather than at `commit`; and inside the transaction `:as-of` its own transaction number sees every finished statement, since they share one number (staged statements of a small transaction each have their own until `commit`). A crash at any point before a large transaction's commit leaves the database as it was; the WAL file exists, header only, from before its first page write, so a new file reopens too. The first checkpoint after a WAL is found at open, or after a large transaction is rolled back, now writes a meta page before it deletes the WAL. Measured at 1M facts against v3 before this change (one `execute`, or a `WriteTransaction` of 10 statements): peak memory above the input text goes from 1.3–2.4 GiB to at most 55 MiB at the default (under 10 MiB at 16,384); with sequential entities it is faster (3.9 s from 9.1 s for short values), with random entities about as fast for short values (9.5 s from 11.4 s in one `execute`) and slower for long ones over 10 statements (34 s from 22 s; 16,384 takes up to twice as long as the default with random entities), and a transaction of several statements over random entities can leave the file up to 1.8× larger, since each statement keeps the pages its base references until it ends; later checkpoints reuse that space. - **Queries inside a `WriteTransaction` read through a snapshot** (#550). They no longer copy every committed and pending fact into a new list: they read the committed indexes and the pending facts, with the transaction's staged facts read with the pending ones (or, in a large transaction, its working trees), and run on the streaming engine where it plans them, counted by `query_memory_limit`. diff --git a/CLAUDE.md b/CLAUDE.md index adb7db12f..a4da86d8e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -142,12 +142,12 @@ cargo run < demos/demo_negation.txt - `evaluator.rs`: `rule_invocation_to_pattern`, the rule limits and their errors - `stratification.rs`: `DependencyGraph`, `stratify()` — negative edges + cycle detection; a positive edge to a rule with `not`/`not-join`/`or`/`or-join` also raises the caller, off a cycle (#628) - `rules.rs`: `RuleRegistry` — thread-safe rule management - - `types.rs`: `EdnValue`, `Pattern`, `DatalogQuery`, `AsOf`, `ValidAt`, `WhereClause` (incl. `Not`, `NotJoin`); `PseudoAttr` enum, `AttributeSpec` wrapper + - `types.rs`: `EdnValue`, `Pattern`, `DatalogQuery`, `AsOf`, `ValidAt`, `Count` (`:limit` literal or slot, #306), `WhereClause` (incl. `Not`, `NotJoin`); `PseudoAttr` enum, `AttributeSpec` wrapper - `optimizer.rs`: Selectivity-based join reordering on every build (the `wasm` feature is a deprecated no-op). `reorder()` gates every reordering site; tests turn it off per thread (`set_reorder`, `minigraf::__reorder` under `test-hooks`, #596); `plan_seeded` orders a body with its imported variables counted as bound, `plan_after` orders a rule body's clauses after its delta, each joining on a variable bound before it where one can (#694), and `subplans()` lets tests force the engine's negation bodies and `or` branches hashed or correlated (`set_subplans`, `minigraf::__subplans`, #546) - - `prepared.rs`: `BindValue`, `PreparedQuery` — parse-once/execute-many with named `$slot` bind slots; `prepare_query`, `substitute` + - `prepared.rs`: `BindValue`, `PreparedQuery` — parse-once/execute-many with named `$slot` bind slots; `prepare_query`, `substitute` (a `:limit $n` slot takes `Val(Integer(n >= 0))`, `resolve_count_slot`) - `magic_sets.rs`: Magic-sets rewriting for demand-driven recursive rule evaluation (not applied to rules with `not`/`not-join`) -3a. **`src/query/engine/`** — Streaming query engine (#473), runs every query (#551): `plan.rs` (the operator tree of every query, or the executor's error code for it: patterns, expression clauses, `not`/`not-join` and `or`/`or-join`, and rule invocations, read as patterns of their relations (`plan_rules`, #548; a magic-set head constant is `HeadArg::Const`), in `optimizer::plan` order, then the `or`s in text order, then the deferred `not`s; an `or` branch or a `not` body is planned as a `:where` is; a clause list with no pattern first starts from `Seed`; a constant no cell equals is `Term::Never`, a scan that reads nothing; `:as-of` resolved when the query opens (`Snapshot::tx_bound`: a counter bound when `tx_id` rises with `tx_count`, else `TxBound::Timestamp`: committed `tx_count` ranges and pending `tx_id <= t`); a per-fact pseudo-attribute pattern reads the metadata column its source's scan fills (the hidden variable the parser named, #594), and `:db/valid-at` joins as `[e _ _]`; plain `:find` variables, aggregates with `:with`, and window functions, over an aggregate's groups too), `scan.rs` (one pattern over committed key ranges via `CommittedReader::key_source`/`live_keys`, AVET ranges for cross-kind constants, then the pending range by whole runs; one entity per probe for a join), `join.rs` (`IndexNestedLoopJoin` when the pattern's entity is bound, or, under a constant attribute and entity not constant, its value (#721: per outer row the AVET ranges of the values that join-equal it and the attribute's pending facts with its key, read once per query in `IdSpace::probes`; an entity cell needs the keyword naming it, known only where the query made the entity from a keyword (rule heads, magic seeds), else the join becomes the planner's `HashJoin`); `HashJoin` otherwise, and on a relation; `JoinKey`: floats by IEEE `==`, a keyword equals a ref to the entity it names), `filter.rs` (`Filter`/`Bind` and pseudo-attribute steps over batches; a `Bind` on a bound slot keeps the rows where the result is join-equal, #596), `anti.rs` (`AntiJoin`, #546: a body with the same rows for every outer row runs once into a hash set of its imported-variable keys; any other body, or one whose first pattern's entity is imported, runs per outer row from `seed.rs`'s one-row `Seed`, via `Operator::rewind`; never deduplicates), `aggregate.rs` (`Aggregate`, #547: a hash table of groups keyed on cells as `Value`'s `Ord` compares values, one entry when ungrouped; `count`/`count-distinct` on cells, other aggregates on values via `datalog::aggregate::Agg`; groups sorted by value and finished in that order; a float group shows its least value by `value_order`), `window.rs` (`Window`, #547: holds each input row's passed-on cells and its named variables' values, then computes every window with `datalog::aggregate::windows`, the executor's code, and returns the rows in input order), `union.rs`/`distinct.rs` (`Union` and `Distinct`, #546: each `or` branch planned as a `not` body is, hashed into a table or run per outer row; then one row per distinct binding, as the executor's `Value`s compare), `cell.rs`/`batch.rs` (id cells in columnar batches of 1,024), `ids.rs` (DICT and provisional ids, `keyword_eid`), `project.rs` (ids → values for returned rows only), `relation.rs`/`recursive.rs` (#548: rule queries; a relation per evaluated predicate holds its derived rows, a set by `encode_value` identity, appended per round so `Old`/`Delta`/`Full` are ranges; `Recursive` runs each stratum's semi-naive rounds, one run per same-stratum top-level invocation, a rule reading its stratum otherwise run in full every round; scans read derived rows after committed and pending ones, with `nil` metadata), `meter.rs` (#549: one `Meter` per query, shared by every holder of state through a `Held` — hash join build sides, `not`/`or` tables, distinct sets, aggregate groups, window rows, rule relations and their temporaries, `IdSpace`; each entry charged by a per-entry estimate before it is inserted, containers grown with `try_reserve`, released on `rewind`/`close`/drop; over `OpenOptions::query_memory_limit`, or a refused allocation, is `QRY-012`; `Stream::close` drops the operator tree, keeping its counters). `mod.rs`: `Operator` (`next_batch`/`rewind`/`close`/`counters`), `Stream` (the operator tree over an owned `Snapshot`, pulled one batch at a time). `DatalogExecutor::open` / `execute_query` run every query on it: a cursor gets the `Stream` (`Answer::Stream`), `execute` drains it. +3a. **`src/query/engine/`** — Streaming query engine (#473), runs every query (#551): `plan.rs` (the operator tree of every query, or the executor's error code for it: patterns, expression clauses, `not`/`not-join` and `or`/`or-join`, and rule invocations, read as patterns of their relations (`plan_rules`, #548; a magic-set head constant is `HeadArg::Const`), in `optimizer::plan` order, then the `or`s in text order, then the deferred `not`s; an `or` branch or a `not` body is planned as a `:where` is; a clause list with no pattern first starts from `Seed`; a constant no cell equals is `Term::Never`, a scan that reads nothing; `:as-of` resolved when the query opens (`Snapshot::tx_bound`: a counter bound when `tx_id` rises with `tx_count`, else `TxBound::Timestamp`: committed `tx_count` ranges and pending `tx_id <= t`); a per-fact pseudo-attribute pattern reads the metadata column its source's scan fills (the hidden variable the parser named, #594), and `:db/valid-at` joins as `[e _ _]`; plain `:find` variables, aggregates with `:with`, and window functions, over an aggregate's groups too), `scan.rs` (one pattern over committed key ranges via `CommittedReader::key_source`/`live_keys`, AVET ranges for cross-kind constants, then the pending range by whole runs; one entity per probe for a join), `join.rs` (`IndexNestedLoopJoin` when the pattern's entity is bound, or, under a constant attribute and entity not constant, its value (#721: per outer row the AVET ranges of the values that join-equal it and the attribute's pending facts with its key, read once per query in `IdSpace::probes`; an entity cell needs the keyword naming it, known only where the query made the entity from a keyword (rule heads, magic seeds), else the join becomes the planner's `HashJoin`); `HashJoin` otherwise, and on a relation; `JoinKey`: floats by IEEE `==`, a keyword equals a ref to the entity it names), `filter.rs` (`Filter`/`Bind` and pseudo-attribute steps over batches; a `Bind` on a bound slot keeps the rows where the result is join-equal, #596), `anti.rs` (`AntiJoin`, #546: a body with the same rows for every outer row runs once into a hash set of its imported-variable keys; any other body, or one whose first pattern's entity is imported, runs per outer row from `seed.rs`'s one-row `Seed`, via `Operator::rewind`; never deduplicates), `aggregate.rs` (`Aggregate`, #547: a hash table of groups keyed on cells as `Value`'s `Ord` compares values, one entry when ungrouped; `count`/`count-distinct` on cells, other aggregates on values via `datalog::aggregate::Agg`; groups sorted by value and finished in that order; a float group shows its least value by `value_order`), `limit.rs` (`Limit`, #306: skips `offset` rows, passes on at most `limit`, then closes its child so nothing below reads more; planned after the last step, after `Builder::finish`), `window.rs` (`Window`, #547: holds each input row's passed-on cells and its named variables' values, then computes every window with `datalog::aggregate::windows`, the executor's code, and returns the rows in input order), `union.rs`/`distinct.rs` (`Union` and `Distinct`, #546: each `or` branch planned as a `not` body is, hashed into a table or run per outer row; then one row per distinct binding, as the executor's `Value`s compare), `cell.rs`/`batch.rs` (id cells in columnar batches of 1,024), `ids.rs` (DICT and provisional ids, `keyword_eid`), `project.rs` (ids → values for returned rows only), `relation.rs`/`recursive.rs` (#548: rule queries; a relation per evaluated predicate holds its derived rows, a set by `encode_value` identity, appended per round so `Old`/`Delta`/`Full` are ranges; `Recursive` runs each stratum's semi-naive rounds, one run per same-stratum top-level invocation, a rule reading its stratum otherwise run in full every round; scans read derived rows after committed and pending ones, with `nil` metadata), `meter.rs` (#549: one `Meter` per query, shared by every holder of state through a `Held` — hash join build sides, `not`/`or` tables, distinct sets, aggregate groups, window rows, rule relations and their temporaries, `IdSpace`; each entry charged by a per-entry estimate before it is inserted, containers grown with `try_reserve`, released on `rewind`/`close`/drop; over `OpenOptions::query_memory_limit`, or a refused allocation, is `QRY-012`; `Stream::close` drops the operator tree, keeping its counters). `mod.rs`: `Operator` (`next_batch`/`rewind`/`close`/`counters`), `Stream` (the operator tree over an owned `Snapshot`, pulled one batch at a time). `DatalogExecutor::open` / `execute_query` run every query on it: a cursor gets the `Stream` (`Answer::Stream`), `execute` drains it. 4. **`src/temporal.rs`** — UTC-only timestamp parsing (avoids chrono CVE GHSA-wcg3-cvx6-7396) diff --git a/README.md b/README.md index 5b86846ff..dd8c56277 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,7 @@ Minigraf is a **single-file embedded graph database** that lets you: - ✅ **Query relationships with Datalog** - Recursive rules, natural graph traversal - ✅ **Time travel through history** - Bi-temporal queries (transaction time + valid time) - ✅ **Window functions** - `sum/count/min/max/avg/rank/row-number :over (partition-by … :order-by …)` in `:find` clauses +- ✅ **Bounded results** - `:limit N` returns at most N rows and stops the work once it has them - ✅ **Prepared statements** - Parse + plan once with `$slot` bind tokens, execute thousands of times - ✅ **Embed anywhere** - Native, WASM, mobile, IoT - one `.graph` file - ✅ **Zero configuration** - Just `Minigraf::open("data.graph")` and you're done diff --git a/ROADMAP.md b/ROADMAP.md index aaf26a707..682828486 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -94,7 +94,7 @@ Scope — format and storage integrity: Scope — query API: - Cursor API: `Minigraf::query` and `PreparedQuery::query` return an owned `Cursor` that delivers rows in batches, with the answer fixed at open (#432; Rust API done on `v3`). The binding cursor objects follow. Behind the cursor, the streaming engine (#473) runs every query and produces rows as they are pulled: patterns, expressions and pseudo-attributes (#544), joins and `:db/valid-at` (#545), `not` and `or` (#546), aggregates and windows (#547), rules (#548), and transaction reads (#550); the executor's query path is gone (#551; done on `v3`). Operators do not spill: memory follows operator state, with an opt-in per-query limit (`QRY-012`, #549). -- Native `:limit N` / `:offset` for plain `:find` queries (#306), with the 10M-fact acceptance test moved from #473. +- Native `:limit N` (done on `v3`, with the 10M-fact acceptance test moved from #473), then `:order-by` and `:offset` with a top-k `Sort` (#306; spec `docs/superpowers/specs/2026-10-24-limit-order-by-design.md`). - Fact-log iterator: `Minigraf::fact_log(&FactFilter)` streams every fact version (assertions and retractions, with `tx_count`, `tx_id` and valid-time bounds) in transaction order or storage order, filtered by attribute, attribute prefix, entity and tx range, with memory bounded by a window rather than the database (#430; Rust API done on `v3`). An open log pins its generation: checkpoints run but do not reuse its pages (#543; `API-013` deprecated). A raw log writer (#431) completes the extract-transform-load path; the binding objects follow with #462. - Read-only open: `OpenOptions::read_only` takes a shared lock, so several readers can hold a file at once, writes nothing to the file or its WAL (a WAL is applied in memory, a v7 file is read without migrating), and refuses writes with `API-014` (#429; Rust API done on `v3`). The bindings' `open_with_options` follows. - Raw log writer: `LogWriter` builds a new database from `FactRecord`s with their `tx_count`, `tx_id`, valid-time windows and assertion flags kept, gaps allowed (a purged transaction stays a hole for `:as-of`), at `.partial` renamed into place by `finish` (#431; Rust API done on `v3`). Completes extract (#429, #430), transform, load. The binding objects follow. diff --git a/docs/TEST_COVERAGE.md b/docs/TEST_COVERAGE.md index 8de85e9b2..0eb68beb5 100644 --- a/docs/TEST_COVERAGE.md +++ b/docs/TEST_COVERAGE.md @@ -353,6 +353,11 @@ These tests compare the database against an independent answer rather than a sin - Every planner error's code, and the error order. - The two answer changes of the spec (§4), and an unknown predicate in a `not` body, were checked against both the engine and the executor; their tests keep the engine's answer. - The envelope harness's `-irregular` fixture keeps its source's records, with `tx_id`s that fall and rise (`cargo test` runs the example's tests). +- **`:limit`** (#306) + - `tests/limit_order_test.rs` (5 tests, `test-hooks`): 14 shapes (patterns, joins, `not`, `or`, grouped and ungrouped aggregates and the zero row, a window, rule queries, `:as-of`, `:any-valid-time`, an expression with no pattern), in memory and over committed and pending facts, with limits 0, 1, 2, the answer size less one, the answer size and five more: through `execute`, a cursor pulled one row at a time, a prepared `:limit $n` and its cursor, with reordering on and off, each returns `min(n, |answer|)` rows that are a sub-bag of the unlimited answer. A write transaction's own staged facts; slot errors (`API-010`, `PRS-115`, `INT-035`). + - `src/query/engine/limit.rs` (4 unit tests): at most `limit` rows across batch boundaries, `limit 0` pulls nothing, one batch read for 10 rows and two for `BATCH_ROWS + 1`, offsets across batches and past the end, `rewind`. + - `src/cursor.rs` (2 tests): through a cursor, in memory and on file, `:limit 10` leaves the scan at one internal batch and `:limit 0` at none. + - Parser (3 unit tests), prepared slots (3 unit tests), `prs_115_…` in `tests/error_codes_prs_test.rs`, the grammar corpus (3 valid, 2 syntax, 2 semantic files) and two verifier `SEMANTIC` witnesses (`INT-012`, `PRS-115`). - **Query memory limit** (`tests/memory_limit_test.rs`, 7 tests; #549) - `OpenOptions::query_memory_limit` is off by default and set by its builder. - `QRY-012` through `execute`, a cursor's first pull, a prepared query and a rule query (from `execute`, and from `query` when the rules run at open). The database answers after a query fails. @@ -478,6 +483,7 @@ The thirteen ignored tests (fourteen with `test-hooks`) are intentionally exclud - 1M aggregate groups return 1M rows, and with `query_memory_limit` at 32 MiB stop with `QRY-012` naming `Aggregate`, after which the database answers; - 13 patterns sharing no variable over the fuzz harness's 3 facts stream 3^13 rows with peak RSS under 64 MiB, and with a 64-byte limit stop with `QRY-012` naming `HashJoin`; - `count` over every one of 1M facts counts them under a 4 KiB `query_memory_limit` (`count_holds_no_rows`, the #473 constant-memory `count` item). +- One `:limit` acceptance test (`tests/limit_acceptance_test.rs`, #306, the #473 `:limit` item), run by the nightly smoke workflow: `:limit 10` over 10M checkpointed facts returns 10 rows in under 250 ms under a 64 KiB `query_memory_limit`, adding under 32 MiB of peak RSS. - One large-transaction acceptance test (`tests/large_transaction_acceptance_test.rs`, #550; `test-hooks` only), run by the nightly smoke workflow (see [Large transactions](#large-transactions-550)). - One measurement test (`meter_peak_against_rss`, `src/query/engine/tests.rs`, #549): prints the meter's peak against the peak RSS for the same two shapes, opening the engine directly. diff --git a/tests/limit_acceptance_test.rs b/tests/limit_acceptance_test.rs index 760d0e129..df45c39ba 100644 --- a/tests/limit_acceptance_test.rs +++ b/tests/limit_acceptance_test.rs @@ -14,10 +14,11 @@ // File-backed (Minigraf::open, tempfile): native only, like the other file tests. #![cfg(not(target_arch = "wasm32"))] -use minigraf::{Minigraf, OpenOptions, QueryResult}; +use minigraf::{FactRecord, LogWriter, OpenOptions, QueryResult, Value}; const FACTS: usize = 10_000_000; -const CHUNK: usize = 100_000; +/// Facts per transaction. +const TX_FACTS: usize = 10_000; /// What the query may hold: a `Limit` holds nothing and a scan one batch of /// ids. Holding even one id per fact would need megabytes. const MEMORY_LIMIT: usize = 64 << 10; @@ -44,16 +45,26 @@ fn proc_status_kb(field: &str) -> usize { fn limit_10_over_10m_facts() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("limit.graph"); - { - let db = Minigraf::open(&path).unwrap(); - for chunk in 0..FACTS / CHUNK { - let facts: String = (chunk * CHUNK..(chunk + 1) * CHUNK) - .map(|i| format!("[:e{i} :a {i}]")) - .collect(); - db.execute(&format!("(transact [{facts}])")).unwrap(); - db.checkpoint().unwrap(); - } + // `LogWriter` writes the records straight into the trees in batched + // checkpoints: the `transact` path took 42 minutes for 10M facts. + let start = std::time::Instant::now(); + let mut w = LogWriter::create(&path, OpenOptions::new()).unwrap(); + for i in 0..FACTS { + let tx = (i / TX_FACTS) as u64 + 1; + w.append(&FactRecord { + entity: uuid::Uuid::from_u128(i as u128 + 1), + attribute: ":a".to_string(), + value: Value::Integer(i as i64), + tx_count: tx, + tx_id: 1_700_000_000_000 + tx, + valid_from: (1_700_000_000_000 + tx) as i64, + valid_to: i64::MAX, + asserted: true, + }) + .unwrap(); } + w.finish().unwrap(); + eprintln!("limit_10_over_10m_facts: built in {:.2?}", start.elapsed()); let db = OpenOptions::new() .query_memory_limit(MEMORY_LIMIT) .path(&path) From 507b9f9dff99aa8c6ddf4b763b3b990683a59ed9 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 24 Oct 2026 13:06:59 +0530 Subject: [PATCH 08/10] docs: :limit in CHANGELOG/README/ROADMAP/CLAUDE.md, test counts (#306) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0169pNQsLcCEn48te9X9WuZW --- CLAUDE.md | 2 +- docs/TEST_COVERAGE.md | 12 +++++++----- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index a4da86d8e..b576a834d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -212,7 +212,7 @@ Auto-migrates v7 → v8 on open (spec §9, with a backup meta page). v1–v6 are ## Test Coverage -**1716 tests** (1703 passing, 13 ignored; unit + integration + doc). Engine answers once checked by the executor are frozen in `tests/answers/` (#551). +**1730 tests** (1716 passing, 14 ignored; unit + integration + doc). Engine answers once checked by the executor are frozen in `tests/answers/` (#551). See `docs/TEST_COVERAGE.md` for the full per-file breakdown. **Testing conventions** — see the Testing Conventions section below before writing any tests. diff --git a/docs/TEST_COVERAGE.md b/docs/TEST_COVERAGE.md index 0eb68beb5..90034e249 100644 --- a/docs/TEST_COVERAGE.md +++ b/docs/TEST_COVERAGE.md @@ -4,11 +4,13 @@ | | Default features | With `--features test-hooks` | |---|---|---| -| **Passing** | 1,703 | 1,719 | -| **Ignored** | 13 | 14 | -| **Total** | 1,716 | 1,733 | +| **Passing** | 1,716 | 1,737 | +| **Ignored** | 14 | 15 | +| **Total** | 1,730 | 1,752 | -**Verified**: 2026-10-24 with `cargo test --quiet`. +**Verified**: 2026-10-24 with `cargo test --quiet` (and `--features test-hooks`). + +#306 PR 1 (`:limit`) added 13 tests, 5 more under `test-hooks`, and one ignored acceptance test. #551 PR 2 deleted the executor's query path, with the 73 unit tests of its internals and the oracle. Two engine tests replace shapes no other test covered, and a storage unit test covers the reads only the executor made. @@ -483,7 +485,7 @@ The thirteen ignored tests (fourteen with `test-hooks`) are intentionally exclud - 1M aggregate groups return 1M rows, and with `query_memory_limit` at 32 MiB stop with `QRY-012` naming `Aggregate`, after which the database answers; - 13 patterns sharing no variable over the fuzz harness's 3 facts stream 3^13 rows with peak RSS under 64 MiB, and with a 64-byte limit stop with `QRY-012` naming `HashJoin`; - `count` over every one of 1M facts counts them under a 4 KiB `query_memory_limit` (`count_holds_no_rows`, the #473 constant-memory `count` item). -- One `:limit` acceptance test (`tests/limit_acceptance_test.rs`, #306, the #473 `:limit` item), run by the nightly smoke workflow: `:limit 10` over 10M checkpointed facts returns 10 rows in under 250 ms under a 64 KiB `query_memory_limit`, adding under 32 MiB of peak RSS. +- One `:limit` acceptance test (`tests/limit_acceptance_test.rs`, #306, the #473 `:limit` item), run by the nightly smoke workflow: `:limit 10` over 10M checkpointed facts returns 10 rows in under 250 ms under a 64 KiB `query_memory_limit`, adding under 32 MiB of peak RSS; the fixture is written with `LogWriter`. - One large-transaction acceptance test (`tests/large_transaction_acceptance_test.rs`, #550; `test-hooks` only), run by the nightly smoke workflow (see [Large transactions](#large-transactions-550)). - One measurement test (`meter_peak_against_rss`, `src/query/engine/tests.rs`, #549): prints the meter's peak against the peak RSS for the same two shapes, opening the engine directly. From 99bc710bda00a9eb6b2c944334b4ba45c5a3edab Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 24 Oct 2026 18:21:27 +0530 Subject: [PATCH 09/10] docs(benchmarks): :limit envelope and acceptance numbers (#306) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0169pNQsLcCEn48te9X9WuZW --- docs/BENCHMARKS.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/docs/BENCHMARKS.md b/docs/BENCHMARKS.md index 6833fd4bb..5bb4aac1e 100644 --- a/docs/BENCHMARKS.md +++ b/docs/BENCHMARKS.md @@ -54,6 +54,7 @@ Every query runs on the streaming engine since #551. The figures in this table a | `attribute_value` | `[?e :name "p{m}"]` (1 row) | 24.5 ms / 25.6 ms | 288 ms / 297 ms | 3.16 s / 3.16 s | | `attribute_scan` | `[?e :age ?a]` (N/6 rows) | 43.4 ms / 49.4 ms | 462 ms / 484 ms | 4.83 s / 5.18 s | | `count_grouped` | `[:find ?g (count ?e) :where [?e :group ?g]]` | 46.3 ms / 48.6 ms | 521 ms / 524 ms | 5.66 s / 6.20 s | +| `limit` | `[:find ?e ?a :where [?e :age ?a] :limit 10]` (#306) | 307 µs / 502 µs | 317 µs / 348 µs | 318 µs / 391 µs | | `join_2` | `[:p{m} :follows ?f] [?f :name ?n]` (1 row) | 54.0 ms / 63.0 ms | 606 ms / 610 ms | 6.50 s / 7.02 s | | `join_3_selective` | `[:p{m} :manager ?b] [?s :manager ?b] [?s :name ?n]` (10 rows) | 117 ms / 125 ms | 1.24 s / 1.28 s | 13.5 s / 13.8 s | | `join_3_scan` | `[?e :follows ?f] [?f :group 7] [?f :name ?n]` | 150 ms / 155 ms | 1.92 s / 1.96 s | 20.9 s / 21.2 s | @@ -189,6 +190,10 @@ The meter's estimate is within the spec's factor of two of the RSS a query adds Measuring this found that `Aggregate` built its sorted map of groups uncharged while its hash table was still alive: the meter read 282 MiB against 651 MiB of RSS. The map is now charged, and the table goes first, which also cut the real peak to the figure above. +### `:limit` (#306) + +`limit` is `attribute_scan` with `:limit 10`: `Limit` closes the scan after its first internal batch, so the time does not grow with the facts (p50 307 µs, 317 µs and 318 µs at 100K, 1M and 10M, 5 MiB peak RSS each; `attribute_scan` reads the whole attribute in 43 ms, 462 ms and 4.83 s). Same idle server, one NUMA node, 2026-10-24. The acceptance test (`cargo test --profile bench --test limit_acceptance_test -- --ignored --nocapture`, nightly) runs `:limit 10` over 10M facts of one attribute in 0.8 ms under a 64 KiB `query_memory_limit`, adding 132 KiB of RSS. + ### Executor removal (#551) Since #551 every query runs on the engine and the executor's query path is gone. Before it, the queries the engine could not plan ran on the executor, which reads every fact of the patterns' attributes into memory; the envelope's case is `:as-of` with a time. `as_of_ts` is the `as_of` point query with `:as-of "