From c12c6a021edd247eeaea334249f8324bf6a4fd8f Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sun, 25 Oct 2026 03:49:46 +0530 Subject: [PATCH 01/13] docs(plan): detail #306 PR 2 steps 10-17 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01TcpoiFyFeQQQELL872LfQF --- .../plans/2026-10-24-306-limit-order-by.md | 186 ++++++++++++++++-- 1 file changed, 169 insertions(+), 17 deletions(-) diff --git a/docs/superpowers/plans/2026-10-24-306-limit-order-by.md b/docs/superpowers/plans/2026-10-24-306-limit-order-by.md index 0588e583b..d7a5a66ae 100644 --- a/docs/superpowers/plans/2026-10-24-306-limit-order-by.md +++ b/docs/superpowers/plans/2026-10-24-306-limit-order-by.md @@ -135,20 +135,172 @@ machine (`ssh 192.168.1.16`, `PATH=$HOME/.cargo/bin:$PATH`); the laptop only as --features browser`), release binary size against the 1,200 KiB gate. Codex review (`codex exec -s read-only … < /dev/null`). PR to `v3` "Part of #306". -## PR 2 — `:order-by`, `:offset`, `Sort` (outline; detailed after PR 1 merges) - -10. `Direction`, `DatalogQuery.order_by`/`offset`; parser for `:order-by [..]` (matching - `FindSpec` by `==`) and `:offset`; `PRS-116`–`118`. -11. Prepared `:offset` slots (`resolve_count_slot(…, ":offset", …)`). -12. `aggregate.rs`: a row comparator over the full order (`value_order` per column, keys' - directions, then the other columns ascending). -13. `engine/sort.rs`: full sort and top-k heap, metered (`QRY-012`), `keep = 0` closes without - pulling. -14. `plan_with`: `Step::Sort` then `Step::Limit { offset, limit }`. -15. Tests: ordered lists equal with reordering on/off; every `value_order` class; ties; - aggregates by `(count ?e) :desc`; windows; rule queries; cursors; prepared slots; - `QRY-012` in `Sort`; grammar, verifier `SEMANTIC` witnesses for `PRS-116`, `117`, `118`. -16. Acceptance `top_k` (memory independent of the fact count) and envelope shape `top_k`. -17. Docs: CHANGELOG, ERROR_REFERENCE (`PRS-116`–`118`), README, CLAUDE.md, TEST_COVERAGE, - ROADMAP; docs site `content/datalog-reference.md` and `content/comparison.md` - (`@since v3.0.0`), PR to minigraf-docs merged after PR 2. Close #306 by hand. +## PR 2 — `:order-by`, `:offset`, `Sort` + +Branch `feat/306-order-by` from `v3` 59ce7f24 (PR 1 merged as #745), worktree +`.worktrees/306-order-by`. + +### Decisions for PR 2 (not fixed by the spec) + +7. **No new `Direction` type.** `types::Order { Asc, Desc }`, which a window's `:over` + already uses, is the direction: `DatalogQuery.order_by: Vec<(FindSpec, Order)>` + (deviation 2 from spec §4, which names a `Direction`). +8. **Entries are matched after the section loop.** `:order-by` may come before `:find` + (spec §3), so `parse_query` keeps the parsed entries and checks them against + `find_specs` once every section is read: shape errors (`PRS-118`, and `PRS-117` for an + entry that is not a variable or a list) as the vector is read, `PRS-117` for an entry no + `:find` element equals and `PRS-118` for a repeat after the loop, then `PRS-116`. A list + entry is read with `parse_aggregate`, as `:find` reads one; an entry it rejects is + `PRS-117` (it is not a `:find` element), so the grammar-accepted `(foo ?x)` is semantic. +9. **`Sort` holds values, not cells.** Each held row is the `Vec` of its `:find` + columns (the full order compares all of them), turned back into cells with + `Cell::of_value` only for the rows it returns, as `Window` does for its results. Holding + both cells and values would double the memory for no gain; a value's cell round-trips + (`Iid` ↔ keyword, `Eid` ↔ ref, a long string back to its value ref). +10. **`Sort` returns the `:find` columns only.** `Step::Sort { cols, order, keep }` reads the + slots `cols` (= `plan.find`), returns batches of width `cols.len()`, and `plan.find` + becomes `0..cols.len()`, so `Limit` and `Project` after it see only those columns. + `order` is the full order as `(column index, desc)`: the order keys, then every other + `:find` index ascending. +11. **Top-k is a hand-written binary heap** over `Vec>` with the full-order + comparator (worst row on top): `std::collections::BinaryHeap` needs `Ord` on the element, + which would carry the order spec in every row. ~30 lines, unit-tested against a full + sort. +12. **A `Sort` stream keeps its DICT lookups** (`Stream::hold_ids`, as for `Window`): it turns + every row's ids into values in its first pull. With top-k it holds at most `k` rows' + ids' worth of lookups beyond what the scan touched; without `keep` it holds the answer + anyway. +13. **`:offset` alone (with `:order-by`, no `:limit`)** plans `Sort` without `keep` and + `Limit { offset, limit: None }`. +14. **Error texts**: `PRS-116` `` `:offset` requires `:order-by` ``; `PRS-117` + `` `:order-by` entry {} is not a `:find` element ``; `PRS-118` + `` `:order-by` takes a non-empty vector of `:find` elements, each optionally followed by + `:asc` or `:desc`; {} `` (the fault: `not a vector`, `empty`, `` `?x` twice ``, + `` `:desc` without an element ``, `` `:foo` ``). + +### Step 10. Parser: `:order-by` and `:offset` + +- `types.rs`: `DatalogQuery.order_by: Vec<(FindSpec, Order)>` (empty = none) and + `offset: Option`; `new`/`from_patterns` set them empty/`None`; `Count`'s doc names + `:offset`. +- `parser.rs` `parse_query`: an `:offset` arm as `:limit`'s (`INT-012` repeated, `INT-013` + no value, `PRS-115` otherwise). An `:order-by` arm: repeated `INT-012`; no value (end, or a + keyword next) `INT-013`; not a vector `PRS-118`; empty `PRS-118`; then per item: `?sym` or + a list starts an entry, `:asc`/`:desc` sets the direction of the entry before it (none + before it, or that entry already has one: `PRS-118`), any other keyword `PRS-118`, + anything else `PRS-117`. After the loop (decision 8): each entry must equal a `:find` + element (`PRS-117`) and appear once (`PRS-118`); `offset.is_some() && order_by.is_empty()` + is `PRS-116`. +- `error.rs`: `Prs116`–`Prs118` at the four sites (enum after `Prs115`, registry, parser + category arm, `ALL`); `docs/ERROR_REFERENCE.md` rows and sections. +- Tests (`parser.rs` unit): defaults; `:asc` default; `:desc`; aggregate and window entries; + every section order (`:order-by` before `:find`, between, last; `:offset` before + `:order-by`); `:offset 0`, slot; every row of spec §3.1 for the two sections, incl. + `?y` bound in `:where` not in `:find`, a `:with` variable, `(count ?e)` not in `:find`, + `1`, `"x"`, `(foo ?x)`, `[?x ?x]`, `[(count ?e) (count ?e)]`, `[:desc ?x]`, + `[?x :asc :desc]`, `[?x :foo]`, `[]`, `?x` (not a vector). `tests/error_codes_prs_test.rs`: + `prs_116_…`, `prs_117_…`, `prs_118_…`. + +### Step 11. Prepared `:offset` slots + +- `prepared.rs`: `collect_slot_names` adds `offset`'s slot; `substitute` resolves it with + `resolve_count_slot(name, ":offset", bindings)` (its doc names both options). +- Tests (`prepared.rs` unit): listed in `slot_names`; unbound `API-010`; `n >= 0`, `n < 0` + `PRS-115`, wrong type `API-024`; one slot name used by both `:limit` and `:offset`. + +### Step 12. The full-order comparator + +- `aggregate.rs`: `pub(crate) fn row_order(a: &[Value], b: &[Value], order: &[(usize, bool)]) + -> Ordering`: the first non-equal `value_order` of the listed columns, reversed where + `desc`. +- Tests: each direction; ties falling to the next column; every `value_order` class in one + column (Integer before an equal Float, `-0.0` before `0.0`, NaN after numbers, strings, + booleans, refs, keywords, `nil`, sets), ascending and descending. + +### Step 13. `Sort` operator + +- `engine/sort.rs`: `Sort { child, cols, order, keep: Option, rows: Vec>, + out: Option>>, held: Held, counters }`. + - `keep == Some(0)`: close the child without pulling, return `None`. + - Otherwise read the child to its end; each row's values at `cols` (a missing slot reads + `nil`, as `Window`), charged (`size_of::>()` + `value_bytes`) before it is + held, `try_reserve` for growth. + - Without `keep`: push; at the end sort by `row_order` (`sort_unstable_by`: rows equal + under the full order are equal in every column, so stability is not needed). + - With `keep = k`: a max-heap (worst on top) of at most `k` rows; while below `k`, push and + sift up; else, if the row is better than the top, replace it (release the old row's + charge) and sift down; otherwise drop the row (never charged: compare first, then + charge). At the end, sort the heap's rows. + - Return the sorted rows in batches of `BATCH_ROWS` (`Cell::of_value` per value), + uncharging each row as it leaves. + - `rewind` clears and rewinds the child; `close` releases and closes the child; + `counters` reports `bytes_held`. +- `plan.rs`: `Step::Sort { cols: Vec, order: Vec<(usize, bool)>, keep: Option }`; + `mod.rs` `build` routes it; `hold_ids` also for `Sort` (decision 12). +- Engine unit tests (`sort.rs`, with a test child as `limit.rs` has): sort of 5,000 shuffled + rows equals `sort_by(row_order)`; top-k for k = 1, 10, 1,024, 4,999, 5,000, 6,000 equals + the full sort's first `k`; `keep 0` pulls nothing and closes the child; ties broken by + the second column; descending keys; `bytes_held` with `keep 10` over 100,000 rows stays + under the bytes of ~10 rows (bounded, not growing); `QRY-012` with a small meter limit, + with and without `keep`; `rewind` sorts again. + +### Step 14. Planning + +- `plan_with`, after `finish` and before the `Limit` push: with `order_by` non-empty, map + each entry to its `:find` index (`position(==)`, an invariant error if none: the parser + checked it), add every other index ascending, push `Step::Sort { cols: plan.find.clone(), + order, keep }` with `keep = limit.map(|l| offset.saturating_add(l))`, set + `plan.find = (0..n).collect()`. Then `Step::Limit { offset, limit }` when either is given + (`offset` default 0); a slot in either is `INT-032`. +- Engine tests: ordered plain, join, `not`, `or`, aggregate (`(count ?e) :desc`, grouped and + ungrouped zero row), window (sort by a window column), rule query, `:as-of`, `:valid-at`; + each equals the unlimited answer sorted by the full order, sliced `offset..offset+limit`. + +### Step 15. Integration tests (`tests/limit_order_test.rs`) + +- Ordered shapes (each step-14 shape with `:order-by`): through `execute`, the cursor (one + row at a time), `PreparedQuery::execute`/`query` with `$n`/`$o` slots, reordering on and + off: the list equals `answer` sorted by the full order (computed in the test with its own + comparator over `value_order` classes, on the test's value keys), sliced; all four paths + and both reorder settings give the same list. +- Pages: `:offset 0, 3, 6 … :limit 3` concatenate to the whole ordered answer. +- `:offset` past the answer → no rows; `:limit 0` → none; `:offset` without `:limit`. +- A value of every class in one ordered column, ascending and descending. +- File database with committed and pending facts; `WriteTransaction` reads. +- `QRY-012` from `Sort` with a small `query_memory_limit`, and none with top-k `:limit 10` + under the same limit over enough rows to exceed it unsorted. +- Grammar (step-6 style): `grammar.pest` `kw_offset`, `order_by_section = { kw_order_by ~ + "[" ~ order_entry+ ~ "]" }`, `order_entry = { find_spec ~ (kw_asc | kw_desc)? }`, + `offset_section = { kw_offset ~ (integer_lit | bind_slot) }`, both in `query_section`; + `docs/GRAMMAR.md` regenerated (`UPDATE_GRAMMAR_DOC=1`). Corpus: valid `query_order_by.edn`, + `query_order_by_desc_aggregate.edn`, `query_order_by_offset_limit.edn`, + `query_offset_slot.edn`; syntax `query_order_by_not_vector.edn`, + `query_order_by_empty.edn`, `query_order_by_leading_desc.edn`; semantic + `query_offset_without_order_by.edn`, `query_order_by_not_in_find.edn`, + `query_order_by_twice_entry.edn`, `query_offset_negative.edn`. +- Verifier: `":offset"`, `":asc"`, `":desc"` in `TOKENS` (37); template 16 + `("(query [:find ?x :where [?x :a ?y] :order-by [", "]])")`; `SEMANTIC` `PRS-116` + (`… :offset 1])`), `PRS-117` (`… :order-by [?y]])`), `PRS-118` (`… :order-by [?x ?x]])`). + `KNOWN_DRIFT` stays empty at bound 3 locally, bound 4 on the bench (`:asc`/`:desc` are + new in the window template too: any drift there is a finding, fixed or filed). + +### Step 16. Acceptance and envelope + +- `tests/limit_acceptance_test.rs`: `top_k_10_over_10m_facts` over the same fixture, + `[:find ?e ?v :where [?e :a ?v] :order-by [?v :desc] :limit 10]`: the rows are + `9_999_990..10M` descending; peak RSS added under 32 MB (memory independent of the fact + count; time follows the fact count: the input is read in full). The fixture is built once + for both tests (a `OnceLock` path in a kept temp dir), not twice. +- `examples/perf_envelope.rs`: shape `top_k`, `[:find ?e ?a :where [?e :age ?a] + :order-by [?a :desc] :limit 10]`, rows `10.min(e)`. Bench run at 100K, 1M, 10M + (`--profile envelope`); numbers in the PR body and `BENCHMARKS.md`. + +### Step 17. Docs and PR + +`CHANGELOG.md` (Added: `:order-by`, `:offset`, `PRS-116`–`118`, under PR 1's `:limit` +entry), `docs/ERROR_REFERENCE.md`, `README.md`, `CLAUDE.md` (`sort.rs`, `order_by`/`offset`, +verifier template count 16, test count), `docs/TEST_COVERAGE.md`, `ROADMAP.md` (#306 done). +Docs site (`../minigraf-docs`, worktree): `content/datalog-reference.md` (the three sections, +`@since v3.0.0`), `content/comparison.md`; PR merged after PR 2. Review as step 9 (fmt, +clippy plain and `test-hooks`, bench suite plain and `test-hooks`, wasm32 build, size gate, +Codex). PR to `v3` "Part of #306"; after merge close #306 by hand and tick it in #383. From 3cee0080e29c7560a96ba407a77d84564b5493af Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sun, 25 Oct 2026 03:53:26 +0530 Subject: [PATCH 02/13] feat(parser): :order-by and :offset, PRS-116-118 (#306) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01TcpoiFyFeQQQELL872LfQF --- docs/ERROR_REFERENCE.md | 60 ++++- .../plans/2026-10-24-306-limit-order-by.md | 10 +- src/error.rs | 27 +++ src/query/datalog/executor.rs | 4 + src/query/datalog/parser.rs | 222 +++++++++++++++++- src/query/datalog/types.rs | 14 +- tests/error_codes_prs_test.rs | 38 +++ 7 files changed, 359 insertions(+), 16 deletions(-) diff --git a/docs/ERROR_REFERENCE.md b/docs/ERROR_REFERENCE.md index 23de0fd56..d4565850b 100644 --- a/docs/ERROR_REFERENCE.md +++ b/docs/ERROR_REFERENCE.md @@ -136,6 +136,9 @@ with no `CodedError` anywhere in its chain. | 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 | +| PRS-116 | `:offset` without `:order-by` | Parser | +| PRS-117 | `:order-by` entry is not a `:find` element | Parser | +| PRS-118 | Malformed `:order-by` | Parser | | PRS-119 | Set literal inside a set literal | Parser | | PRS-120 | Set literal element is not a scalar literal | Parser | | PRS-121 | Unterminated set literal | Parser | @@ -2150,10 +2153,10 @@ with `PRS-047`. **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). +**Cause**: The value after `:limit` or `:offset` is not an integer `>= 0` or a `$slot`, or a prepared query's `:limit` or `:offset` 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))`. +- Give `:limit` or `:offset` a whole number of rows, `0` or more, or a bind slot filled with `BindValue::Val(Value::Integer(n))`. **Example**: ```datalog @@ -2163,6 +2166,59 @@ with `PRS-047`. --- +### PRS-116 `:offset` without `:order-by` + +**Error text**: `:offset requires :order-by` + +**Cause**: The query has `:offset` but no `:order-by` (#306). Without an order, which rows an offset skips is unspecified: pages could overlap or miss rows between plans and between writes. + +**Resolution**: +- Add `:order-by` naming the `:find` elements to page by. +- To read a whole answer in pieces, use a cursor (`Minigraf::query`) instead of pages. + +**Example**: +```datalog +(query [:find ?e :where [?e :person/name ?n] :offset 20 :limit 10]) +``` +*No `:order-by`* + +--- + +### PRS-117 `:order-by` entry is not a `:find` element + +**Error text**: `:order-by entry {} is not a :find element` + +**Cause**: An `:order-by` entry does not equal any `:find` element (#306). Entries name `:find` elements: plain variables, or aggregate and window elements written as they appear in `:find`. A variable bound in `:where` but not in `:find`, a `:with` variable, or a value that is neither a variable nor a list is rejected. + +**Resolution**: +- Add the variable or aggregate to `:find`, or order by an element already there. +- Write an aggregate entry exactly as in `:find`, e.g. `(count ?e)`. + +**Example**: +```datalog +(query [:find ?e :where [?e :person/age ?age] :order-by [?age]]) +``` +*`?age` is not in `:find`* + +--- + +### PRS-118 Malformed `:order-by` + +**Error text**: `:order-by takes a non-empty vector of :find elements, each optionally followed by :asc or :desc; {}` + +**Cause**: The value after `:order-by` is not a vector, or is empty, or names the same element twice, or has `:asc`/`:desc` with no element before it (or two directions for one element), or holds another keyword (#306). The fault is named after the semicolon. + +**Resolution**: +- Write `:order-by [?a :desc ?b]`: each `:find` element once, each optionally followed by one direction (default `:asc`). + +**Example**: +```datalog +(query [:find ?n :where [?e :person/name ?n] :order-by [:desc ?n]]) +``` +*`:desc` before its element* + +--- + ### PRS-119 Set literal inside a set literal **Error text**: `a set literal cannot hold a set` diff --git a/docs/superpowers/plans/2026-10-24-306-limit-order-by.md b/docs/superpowers/plans/2026-10-24-306-limit-order-by.md index d7a5a66ae..4df69a2d7 100644 --- a/docs/superpowers/plans/2026-10-24-306-limit-order-by.md +++ b/docs/superpowers/plans/2026-10-24-306-limit-order-by.md @@ -172,11 +172,11 @@ Branch `feat/306-order-by` from `v3` 59ce7f24 (PR 1 merged as #745), worktree anyway. 13. **`:offset` alone (with `:order-by`, no `:limit`)** plans `Sort` without `keep` and `Limit { offset, limit: None }`. -14. **Error texts**: `PRS-116` `` `:offset` requires `:order-by` ``; `PRS-117` - `` `:order-by` entry {} is not a `:find` element ``; `PRS-118` - `` `:order-by` takes a non-empty vector of `:find` elements, each optionally followed by - `:asc` or `:desc`; {} `` (the fault: `not a vector`, `empty`, `` `?x` twice ``, - `` `:desc` without an element ``, `` `:foo` ``). +14. **Error texts** (no backticks: no registry template has any, and the doc check reads + the text between single backticks): `PRS-116` `:offset requires :order-by`; `PRS-117` + `:order-by entry {} is not a :find element`; `PRS-118` `:order-by takes a non-empty + vector of :find elements, each optionally followed by :asc or :desc; {}` (the fault: + `not a vector`, `empty`, `?x twice`, `:desc without an element`, `another keyword :foo`). ### Step 10. Parser: `:order-by` and `:offset` diff --git a/src/error.rs b/src/error.rs index 2b2de4ea4..ca6cfcbf0 100644 --- a/src/error.rs +++ b/src/error.rs @@ -162,6 +162,9 @@ pub(crate) enum ErrorCode { Prs113, Prs114, Prs115, + Prs116, + Prs117, + Prs118, Prs119, Prs120, Prs121, @@ -1028,6 +1031,24 @@ pub(crate) const REGISTRY: &[(ErrorCode, &str, &str, ErrorCategory)] = &[ "{} must be a non-negative integer or a bind slot, got {}", ErrorCategory::Parser, ), + ( + ErrorCode::Prs116, + "PRS-116", + ":offset requires :order-by", + ErrorCategory::Parser, + ), + ( + ErrorCode::Prs117, + "PRS-117", + ":order-by entry {} is not a :find element", + ErrorCategory::Parser, + ), + ( + ErrorCode::Prs118, + "PRS-118", + ":order-by takes a non-empty vector of :find elements, each optionally followed by :asc or :desc; {}", + ErrorCategory::Parser, + ), ( ErrorCode::Prs119, "PRS-119", @@ -2354,6 +2375,9 @@ mod tests { | ErrorCode::Prs113 | ErrorCode::Prs114 | ErrorCode::Prs115 + | ErrorCode::Prs116 + | ErrorCode::Prs117 + | ErrorCode::Prs118 | ErrorCode::Prs119 | ErrorCode::Prs120 | ErrorCode::Prs121 @@ -2627,6 +2651,9 @@ mod tests { ErrorCode::Prs113, ErrorCode::Prs114, ErrorCode::Prs115, + ErrorCode::Prs116, + ErrorCode::Prs117, + ErrorCode::Prs118, ErrorCode::Prs119, ErrorCode::Prs120, ErrorCode::Prs121, diff --git a/src/query/datalog/executor.rs b/src/query/datalog/executor.rs index 04d51653f..6701a6355 100644 --- a/src/query/datalog/executor.rs +++ b/src/query/datalog/executor.rs @@ -1483,6 +1483,8 @@ mod tests { max_derived_facts: None, max_results: None, limit: None, + offset: None, + order_by: Vec::new(), })) .unwrap(); @@ -1542,6 +1544,8 @@ mod tests { max_derived_facts: None, max_results: None, limit: None, + offset: None, + order_by: Vec::new(), })) .unwrap(); diff --git a/src/query/datalog/parser.rs b/src/query/datalog/parser.rs index 05d88d68a..c19aedd7f 100644 --- a/src/query/datalog/parser.rs +++ b/src/query/datalog/parser.rs @@ -625,6 +625,52 @@ fn parse_whole_command(input: &str) -> Result { /// Parse an aggregate expression list: (func-name ?var) /// e.g., [Symbol("count"), Symbol("?e")] → FindSpec::Aggregate { "count", "?e" } +/// The entries of `:order-by`'s vector (#306): `:find` elements, each +/// optionally followed by `:asc` or `:desc`. Whether each names a `:find` +/// element is checked once every section is read. +fn parse_order_by(val: &EdnValue) -> Result> { + let shape = |fault: String| err_coded!(ErrorCode::Prs118, fault); + let items = val + .as_vector() + .ok_or_else(|| shape("not a vector".to_string()))?; + if items.is_empty() { + return Err(shape("empty".to_string())); + } + let mut entries: Vec<(FindSpec, Order)> = Vec::new(); + // Whether the last entry was given its direction. + let mut directed = true; + for item in items { + match item { + EdnValue::Keyword(k) if k == ":asc" || k == ":desc" => { + let Some(last) = entries.last_mut().filter(|_| !directed) else { + return Err(shape(format!("{k} without an element"))); + }; + last.1 = if k == ":desc" { + Order::Desc + } else { + Order::Asc + }; + directed = true; + } + EdnValue::Keyword(k) => return Err(shape(format!("another keyword {k}"))), + EdnValue::Symbol(s) if s.starts_with('?') => { + entries.push((FindSpec::Variable(s.clone()), Order::Asc)); + directed = false; + } + // A list that is no aggregate or window cannot equal a `:find` + // element. + EdnValue::List(elems) => { + let spec = parse_aggregate(elems) + .map_err(|_| err_coded!(ErrorCode::Prs117, format!("{item:?}")))?; + entries.push((spec, Order::Asc)); + directed = false; + } + other => bail_coded!(ErrorCode::Prs117, format!("{other:?}")), + } + } + Ok(entries) +} + fn parse_aggregate(elems: &[EdnValue]) -> Result { // Detect window function: contains ":over" keyword anywhere in elems let has_over = elems @@ -829,6 +875,8 @@ fn parse_query(elements: &[EdnValue]) -> Result { let mut query_max_derived_facts: Option = None; let mut query_max_results: Option = None; let mut query_limit: Option = None; + let mut query_offset: Option = None; + let mut query_order_by: Option> = None; let mut i = 0; while i < query_vector.len() { @@ -950,28 +998,49 @@ fn parse_query(elements: &[EdnValue]) -> Result { i += 1; continue; } - ":limit" => { + ":limit" | ":offset" => { current_clause = None; - if query_limit.is_some() { - bail_coded!(ErrorCode::Int012, ":limit"); + let target = if keyword == ":limit" { + &mut query_limit + } else { + &mut query_offset + }; + if target.is_some() { + bail_coded!(ErrorCode::Int012, keyword); } i += 1; if !value_at(i) { - bail_coded!(ErrorCode::Int013, ":limit"); + bail_coded!(ErrorCode::Int013, keyword); } let val = query_vector .get(i) - .ok_or_else(|| err_coded!(ErrorCode::Int013, ":limit"))?; - query_limit = Some(match val { + .ok_or_else(|| err_coded!(ErrorCode::Int013, keyword))?; + *target = 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:?}")), + other => bail_coded!(ErrorCode::Prs115, keyword, format!("{other:?}")), }); i += 1; continue; } + ":order-by" => { + current_clause = None; + if query_order_by.is_some() { + bail_coded!(ErrorCode::Int012, ":order-by"); + } + i += 1; + if !value_at(i) { + bail_coded!(ErrorCode::Int013, ":order-by"); + } + let val = query_vector + .get(i) + .ok_or_else(|| err_coded!(ErrorCode::Int013, ":order-by"))?; + query_order_by = Some(parse_order_by(val)?); + i += 1; + continue; + } ":with" => { // Collect ?-prefixed symbols until the next keyword or end of vector if !value_at(i + 1) { @@ -1130,6 +1199,20 @@ fn parse_query(elements: &[EdnValue]) -> Result { bail_coded!(ErrorCode::Prs021); } + // `:order-by` may come before `:find`, so its entries are matched here. + let order_by = query_order_by.unwrap_or_default(); + for (k, (entry, _)) in order_by.iter().enumerate() { + if !find_specs.contains(entry) { + bail_coded!(ErrorCode::Prs117, entry.display_name()); + } + if order_by[..k].iter().any(|(e, _)| e == entry) { + bail_coded!(ErrorCode::Prs118, format!("{} twice", entry.display_name())); + } + } + if query_offset.is_some() && order_by.is_empty() { + bail_coded!(ErrorCode::Prs116); + } + let mut query = DatalogQuery::new(find_specs, where_clauses); query.as_of = query_as_of; query.valid_at = query_valid_at; @@ -1137,6 +1220,8 @@ fn parse_query(elements: &[EdnValue]) -> Result { query.max_derived_facts = query_max_derived_facts; query.max_results = query_max_results; query.limit = query_limit; + query.offset = query_offset; + query.order_by = order_by; Ok(DatalogCommand::Query(query)) } @@ -3139,6 +3224,129 @@ mod tests { } } + fn parse_order(input: &str) -> Result<(Vec<(FindSpec, Order)>, Option)> { + match parse_datalog_command(input)? { + DatalogCommand::Query(q) => Ok((q.order_by, q.offset)), + _ => panic!("expected Query"), + } + } + + fn var(name: &str) -> FindSpec { + FindSpec::Variable(name.to_string()) + } + + #[test] + fn order_by_entries_and_directions() { + let q = |tail: &str| { + format!("(query [:find ?x ?y (count ?z) :where [?x :a ?y] [?x :b ?z] {tail}])") + }; + assert_eq!(parse_order(&q("")).unwrap(), (vec![], None)); + assert_eq!( + parse_order(&q(":order-by [?y]")).unwrap().0, + vec![(var("?y"), Order::Asc)] + ); + let count = FindSpec::Aggregate { + func: "count".to_string(), + var: "?z".to_string(), + }; + assert_eq!( + parse_order(&q(":order-by [(count ?z) :desc ?x :asc ?y]")) + .unwrap() + .0, + vec![ + (count, Order::Desc), + (var("?x"), Order::Asc), + (var("?y"), Order::Asc) + ] + ); + let w = "(query [:find ?x (sum ?y :over (:order-by ?x)) :where [?x :a ?y] \ + :order-by [(sum ?y :over (:order-by ?x)) :desc]])"; + let (order, _) = parse_order(w).unwrap(); + assert!( + matches!(order.as_slice(), [(FindSpec::Window(_), Order::Desc)]), + "a window entry" + ); + } + + #[test] + fn order_by_and_offset_in_any_section_order() { + for input in [ + "(query [:order-by [?x :desc] :offset 2 :find ?x :where [?x :a ?y]])", + "(query [:offset 2 :find ?x :order-by [?x :desc] :where [?x :a ?y]])", + "(query [:find ?x :where [?x :a ?y] :limit 1 :offset 2 :order-by [?x :desc]])", + ] { + assert_eq!( + parse_order(input).unwrap(), + (vec![(var("?x"), Order::Desc)], Some(Count::Literal(2))), + "{input}" + ); + } + let q = |tail: &str| format!("(query [:find ?x :where [?x :a ?y] :order-by [?x] {tail}])"); + assert_eq!( + parse_order(&q(":offset 0")).unwrap().1, + Some(Count::Literal(0)) + ); + assert_eq!( + parse_order(&q(":offset $o")).unwrap().1, + Some(Count::Slot("o".to_string())) + ); + } + + #[test] + fn offset_errors() { + let q = |tail: &str| format!("(query [:find ?x :where [?x :a ?y] :order-by [?x] {tail}])"); + assert_eq!(code_of(&q(":offset 1 :offset 2")), "INT-012"); + assert_eq!(code_of(&q(":offset")), "INT-013"); + assert_eq!(code_of(&q(":offset :ten")), "INT-013"); + for bad in ["-1", "1.5", "\"ten\"", "?x", "nil", "[1]"] { + assert_eq!(code_of(&q(&format!(":offset {bad}"))), "PRS-115", "{bad}"); + } + assert_eq!( + code_of("(query [:find ?x :where [?x :a ?y] :offset 1])"), + "PRS-116" + ); + assert_eq!( + code_of("(query [:find ?x :where [?x :a ?y] :offset 0 :limit 3])"), + "PRS-116" + ); + } + + #[test] + fn order_by_errors() { + let q = |tail: &str| { + format!( + "(query [:find ?x (count ?z) :with ?w :where [?x :a ?y] [?x :b ?z] [?x :c ?w] {tail}])" + ) + }; + assert_eq!(code_of(&q(":order-by [?x] :order-by [?x]")), "INT-012"); + assert_eq!(code_of(&q(":order-by")), "INT-013"); + assert_eq!(code_of(&q(":order-by :desc")), "INT-013"); + for bad in [ + "[?y]", + "[?w]", + "[(count ?y)]", + "[(sum ?z)]", + "[1]", + "[\"x\"]", + "[(foo)]", + "[[?x]]", + ] { + assert_eq!(code_of(&q(&format!(":order-by {bad}"))), "PRS-117", "{bad}"); + } + for bad in [ + "?x", + "[]", + "[?x ?x]", + "[?x :asc ?x :desc]", + "[(count ?z) (count ?z)]", + "[:desc ?x]", + "[?x :asc :desc]", + "[?x :foo]", + ] { + assert_eq!(code_of(&q(&format!(":order-by {bad}"))), "PRS-118", "{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 de69cee79..dd255dad3 100644 --- a/src/query/datalog/types.rs +++ b/src/query/datalog/types.rs @@ -574,10 +574,16 @@ pub struct DatalogQuery { pub max_results: Option, /// `:limit`: the most rows the query returns (#306). pub limit: Option, + /// `:offset`: the rows of the ordered answer skipped before `:limit`'s + /// (#306); the parser requires `:order-by` with it. + pub offset: Option, + /// `:order-by`: `:find` elements and their directions (#306); empty for + /// none. Every `:find` element not named follows, ascending. + pub order_by: Vec<(FindSpec, Order)>, } -/// The value of `:limit` (#306): a literal, or a bind slot a prepared query -/// fills. +/// The value of `:limit` or `:offset` (#306): a literal, or a bind slot a +/// prepared query fills. #[derive(Debug, Clone, PartialEq, Eq)] pub enum Count { Literal(usize), @@ -595,6 +601,8 @@ impl DatalogQuery { max_derived_facts: None, max_results: None, limit: None, + offset: None, + order_by: Vec::new(), } } @@ -610,6 +618,8 @@ impl DatalogQuery { max_derived_facts: None, max_results: None, limit: None, + offset: None, + order_by: Vec::new(), } } diff --git a/tests/error_codes_prs_test.rs b/tests/error_codes_prs_test.rs index c0ffe144c..922aa95c5 100644 --- a/tests/error_codes_prs_test.rs +++ b/tests/error_codes_prs_test.rs @@ -822,6 +822,44 @@ fn prs_115_limit_not_a_non_negative_integer() { ); } +#[test] +fn prs_116_offset_without_order_by() { + assert_code_and_message( + "(query [:find ?a :where [?a :edge ?b] :offset 2 :limit 1])", + "PRS-116", + ":order-by", + ); +} + +#[test] +fn prs_117_order_by_entry_not_in_find() { + assert_code_and_message( + "(query [:find ?a :where [?a :edge ?b] :order-by [?b]])", + "PRS-117", + "?b", + ); + assert_execute_code( + "(query [:find ?a :where [?a :edge ?b] :order-by [(count ?a)]])", + "PRS-117", + ); +} + +#[test] +fn prs_118_malformed_order_by() { + assert_code_and_message( + "(query [:find ?a :where [?a :edge ?b] :order-by [?a ?a]])", + "PRS-118", + "twice", + ); + for q in [ + "(query [:find ?a :where [?a :edge ?b] :order-by ?a])", + "(query [:find ?a :where [?a :edge ?b] :order-by []])", + "(query [:find ?a :where [?a :edge ?b] :order-by [:desc ?a]])", + ] { + assert_execute_code(q, "PRS-118"); + } +} + #[test] fn prs_105_rule_name_is_a_variable_or_an_operator() { assert_execute_code("(rule [(?r ?a) [?a :edge ?b]])", "PRS-105"); From 783689537e21c1c4a649237883044c56de61885f Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sun, 25 Oct 2026 03:53:47 +0530 Subject: [PATCH 03/13] feat(prepared): :offset bind slots (#306) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01TcpoiFyFeQQQELL872LfQF --- src/query/datalog/prepared.rs | 49 ++++++++++++++++++++++++++++++++--- 1 file changed, 46 insertions(+), 3 deletions(-) diff --git a/src/query/datalog/prepared.rs b/src/query/datalog/prepared.rs index 00b960923..1b688f52d 100644 --- a/src/query/datalog/prepared.rs +++ b/src/query/datalog/prepared.rs @@ -250,8 +250,10 @@ 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()); + for count in [&query.limit, &query.offset] { + if let Some(Count::Slot(name)) = count { + names.insert(name.clone()); + } } collect_slots_from_clauses(&query.where_clauses, &mut names); let mut result: Vec = names.into_iter().collect(); @@ -334,6 +336,10 @@ fn substitute( query.limit = Some(resolve_count_slot(name, ":limit", bindings)?); } + if let Some(Count::Slot(name)) = &query.offset { + query.offset = Some(resolve_count_slot(name, ":offset", bindings)?); + } + for clause in &mut query.where_clauses { substitute_where_clause(clause, bindings)?; } @@ -485,7 +491,7 @@ fn resolve_as_of_slot(name: &str, bindings: &HashMap<&str, &BindValue>) -> Resul } } -/// A `:limit` slot (#306): `Val(Integer(n))` with `n >= 0`. +/// A `:limit` or `:offset` slot (#306): `Val(Integer(n))` with `n >= 0`. fn resolve_count_slot( name: &str, option: &str, @@ -786,6 +792,43 @@ mod tests { assert_eq!(limit_code(None), "API-022"); } + #[test] + fn offset_slot_is_collected_and_resolved() { + let mut q = make_query_limit_slot(); + q.offset = Some(Count::Slot("o".to_string())); + assert_eq!( + collect_slot_names(&q), + vec!["n".to_string(), "o".to_string()] + ); + let (n, o) = ( + BindValue::Val(Value::Integer(2)), + BindValue::Val(Value::Integer(5)), + ); + let bindings: HashMap<&str, &BindValue> = [("n", &n), ("o", &o)].into(); + let filled = substitute(&q, &bindings).unwrap(); + assert_eq!(filled.offset, Some(Count::Literal(5))); + assert_eq!(filled.limit, Some(Count::Literal(2))); + let code = |bv: Option| { + let bindings: HashMap<&str, &BindValue> = std::iter::once(("n", &n)) + .chain(bv.iter().map(|b| ("o", b))) + .collect(); + crate::MinigrafError::from(substitute(&q, &bindings).expect_err("rejected")).code() + }; + assert_eq!(code(Some(BindValue::Val(Value::Integer(-1)))), "PRS-115"); + assert_eq!(code(Some(BindValue::Val(Value::Float(1.0)))), "API-024"); + assert_eq!(code(None), "API-022"); + // One slot for both. + let mut both = make_query_limit_slot(); + both.offset = Some(Count::Slot("n".to_string())); + assert_eq!(collect_slot_names(&both), vec!["n".to_string()]); + let bindings: HashMap<&str, &BindValue> = [("n", &o)].into(); + let filled = substitute(&both, &bindings).unwrap(); + assert_eq!( + (filled.limit, filled.offset), + (Some(Count::Literal(5)), Some(Count::Literal(5))) + ); + } + #[test] fn test_substitute_as_of_counter() { let q = make_query_as_of_slot(); From f436d37faa52368bb14cb6426c40865abc51d737 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sun, 25 Oct 2026 03:54:47 +0530 Subject: [PATCH 04/13] feat(aggregate): row_order, the full order of #306 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01TcpoiFyFeQQQELL872LfQF --- src/query/datalog/aggregate.rs | 17 ++++++++++ src/query/datalog/aggregate/tests.rs | 50 ++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+) diff --git a/src/query/datalog/aggregate.rs b/src/query/datalog/aggregate.rs index 8f0419197..120f04183 100644 --- a/src/query/datalog/aggregate.rs +++ b/src/query/datalog/aggregate.rs @@ -65,6 +65,23 @@ pub(crate) fn value_order(a: &Value, b: &Value) -> Ordering { } } +/// Two rows by the columns `order` lists, each `(column, descending)`: +/// the first column whose values differ by [`value_order`] decides (#306 +/// spec §2, the full order). +pub(crate) fn row_order(a: &[Value], b: &[Value], order: &[(usize, bool)]) -> Ordering { + for &(col, desc) in order { + // Every row has every listed column. + let o = a + .get(col) + .zip(b.get(col)) + .map_or(Ordering::Equal, |(x, y)| value_order(x, y)); + if o.is_ne() { + return if desc { o.reverse() } else { o }; + } + } + Ordering::Equal +} + /// `i` against `f` by exact numeric value; a NaN `f` is greater. fn int_float(i: i64, f: f64) -> Ordering { if f.is_nan() { diff --git a/src/query/datalog/aggregate/tests.rs b/src/query/datalog/aggregate/tests.rs index 0582fdf1e..0f8ffd269 100644 --- a/src/query/datalog/aggregate/tests.rs +++ b/src/query/datalog/aggregate/tests.rs @@ -437,3 +437,53 @@ fn value_order_puts_sets_last_and_orders_them_by_elements() { let prefix = set(vec![i(1)]); assert_eq!(value_order(&prefix, &a), Ordering::Less, "prefix first"); } + +#[test] +fn row_order_follows_keys_then_the_next_column() { + let asc = [(0, false), (1, false)]; + let desc_first = [(0, true), (1, false)]; + let (a, b, c) = (vec![i(1), i(9)], vec![i(1), i(2)], vec![i(2), i(0)]); + let mut rows = vec![a.clone(), b.clone(), c.clone()]; + rows.sort_by(|x, y| row_order(x, y, &asc)); + assert_eq!(rows, vec![b.clone(), a.clone(), c.clone()]); + rows.sort_by(|x, y| row_order(x, y, &desc_first)); + assert_eq!(rows, vec![c.clone(), b.clone(), a.clone()]); + // Keys name columns in any order. + rows.sort_by(|x, y| row_order(x, y, &[(1, true), (0, false)])); + assert_eq!(rows, vec![a.clone(), b, c]); + assert_eq!(row_order(&a, &a, &asc), Ordering::Equal); +} + +#[test] +fn row_order_over_every_value_class() { + let set = Value::Set(crate::graph::types::ValueSet::new([i(1)]).unwrap()); + let ascending = vec![ + f(f64::NEG_INFINITY), + i(-1), + f(-0.0), + f(0.0), + i(1), + f(1.0), + f(f64::NAN), + Value::String("a".to_string()), + Value::Boolean(false), + Value::Boolean(true), + Value::Ref(uuid::Uuid::from_u128(1)), + kw(":a"), + Value::Null, + set, + ]; + let mut rows: Vec> = ascending.iter().rev().map(|v| vec![v.clone()]).collect(); + rows.sort_by(|x, y| row_order(x, y, &[(0, false)])); + let same = |rows: &[Vec], want: &[Value]| { + rows.len() == want.len() + && rows + .iter() + .zip(want) + .all(|(r, w)| value_order(&r[0], w).is_eq()) + }; + assert!(same(&rows, &ascending), "ascending"); + rows.sort_by(|x, y| row_order(x, y, &[(0, true)])); + let descending: Vec = ascending.iter().rev().cloned().collect(); + assert!(same(&rows, &descending), "descending"); +} From 6a03d9a4e0a010c7af5d73ebadaa8abfb27eb843 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sun, 25 Oct 2026 03:57:36 +0530 Subject: [PATCH 05/13] feat(engine): Sort with top-k, planned for :order-by/:offset (#306) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01TcpoiFyFeQQQELL872LfQF --- src/query/datalog/parser.rs | 2 +- src/query/engine/filter.rs | 5 +- src/query/engine/mod.rs | 14 +- src/query/engine/plan.rs | 62 +++-- src/query/engine/sort.rs | 448 ++++++++++++++++++++++++++++++++++++ 5 files changed, 512 insertions(+), 19 deletions(-) create mode 100644 src/query/engine/sort.rs diff --git a/src/query/datalog/parser.rs b/src/query/datalog/parser.rs index c19aedd7f..faacce204 100644 --- a/src/query/datalog/parser.rs +++ b/src/query/datalog/parser.rs @@ -1205,7 +1205,7 @@ fn parse_query(elements: &[EdnValue]) -> Result { if !find_specs.contains(entry) { bail_coded!(ErrorCode::Prs117, entry.display_name()); } - if order_by[..k].iter().any(|(e, _)| e == entry) { + if order_by.iter().take(k).any(|(e, _)| e == entry) { bail_coded!(ErrorCode::Prs118, format!("{} twice", entry.display_name())); } } diff --git a/src/query/engine/filter.rs b/src/query/engine/filter.rs index d4a3dfc2a..b1658d216 100644 --- a/src/query/engine/filter.rs +++ b/src/query/engine/filter.rs @@ -34,6 +34,7 @@ impl Apply { Step::Distinct(_) => "Distinct", Step::Aggregate { .. } => "Aggregate", Step::Window { .. } => "Window", + Step::Sort { .. } => "Sort", Step::Limit { .. } => "Limit", }; Apply { @@ -133,7 +134,8 @@ 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`, `collect::CollectSet`, `limit::Limit`). + // `union::Union`, `distinct::Distinct`, `collect::CollectSet`, `sort::Sort`, + // `limit::Limit`). Step::Join(_) | Step::CollectSet(_) | Step::Anti(_) @@ -141,6 +143,7 @@ fn apply(step: &Step, mut batch: Batch, ctx: &mut Ctx<'_>) -> Result { | Step::Distinct(_) | Step::Aggregate { .. } | Step::Window { .. } + | Step::Sort { .. } | Step::Limit { .. } => {} } Ok(batch) diff --git a/src/query/engine/mod.rs b/src/query/engine/mod.rs index 33e806631..b7bd31399 100644 --- a/src/query/engine/mod.rs +++ b/src/query/engine/mod.rs @@ -20,6 +20,7 @@ mod recursive; mod relation; mod scan; mod seed; +mod sort; #[cfg(all(test, not(target_arch = "wasm32")))] mod tests; mod union; @@ -120,9 +121,9 @@ pub(crate) struct Stream { find: Vec, vars: Vec, project: Counters, - /// Keep the DICT lookups until the stream ends: a `Window` turns every - /// row's ids into values in its first pull, and the rows it returns - /// later read the same ids again (#547). + /// Keep the DICT lookups until the stream ends: a `Window` or a `Sort` + /// turns every row's ids into values in its first pull, and the rows it + /// returns later read the same ids again (#547, #306). hold_ids: bool, /// In a rule query, the relations its rules derived (#548). relations: Option, @@ -274,7 +275,9 @@ fn stream( vars, as_of: _, } = plan; - let hold_ids = steps.iter().any(|s| matches!(s, plan::Step::Window { .. })); + let hold_ids = steps + .iter() + .any(|s| matches!(s, plan::Step::Window { .. } | plan::Step::Sort { .. })); let root = build(source, steps, &meter); Stream { snapshot, @@ -322,6 +325,9 @@ pub(crate) fn build( plan::Step::Window { find, named, keep } => { Box::new(window::Window::new(root, find, named, keep, meter)) } + plan::Step::Sort { cols, order, keep } => { + Box::new(sort::Sort::new(root, cols, order, 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 749b84358..5c2ddd651 100644 --- a/src/query/engine/plan.rs +++ b/src/query/engine/plan.rs @@ -30,7 +30,7 @@ use crate::query::datalog::optimizer; use crate::query::datalog::rules::RuleRegistry; use crate::query::datalog::stratification::DependencyGraph; use crate::query::datalog::types::{ - AsOf, AttributeSpec, Count, DatalogQuery, EdnValue, Expr, FindSpec, Pattern, PseudoAttr, + AsOf, AttributeSpec, Count, DatalogQuery, EdnValue, Expr, FindSpec, Order, Pattern, PseudoAttr, UnaryOp, ValidAt, WhereClause, }; use anyhow::Result; @@ -183,8 +183,16 @@ pub(crate) enum Step { named: Vec, keep: Vec, }, - /// After the last step: skips `offset` rows, then passes on at most - /// `limit` (#306). + /// After the last step: the `:find` columns at slots `cols`, sorted by + /// `order`, `(column of cols, descending)` (the full order), holding at + /// most `keep` rows (#306). + Sort { + cols: Vec, + order: Vec<(usize, bool)>, + keep: Option, + }, + /// After the last step, or `Sort`: skips `offset` rows, then passes on at + /// most `limit` (#306). Limit { offset: usize, limit: Option }, } @@ -557,16 +565,44 @@ 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")), + // no pattern. `execute` rejects an unbound slot (API-010); a prepared + // query fills it. + let count = |c: &Option, option: &str| match c { + None => Ok(None), + Some(Count::Literal(n)) => Ok(Some(*n)), + Some(Count::Slot(_)) => Err(invariant(&format!("a {option} slot reached the planner"))), + }; + let (limit, offset) = ( + count(&query.limit, ":limit")?, + count(&query.offset, ":offset")?, + ); + let offset = offset.unwrap_or(0); + if !query.order_by.is_empty() { + // The full order (#306 spec §2): the keys, then every other `:find` + // column ascending. + let mut order = Vec::with_capacity(query.find.len()); + for (spec, dir) in &query.order_by { + let col = query + .find + .iter() + .position(|s| s == spec) + .ok_or_else(|| invariant("an :order-by entry outside :find"))?; + order.push((col, matches!(dir, Order::Desc))); + } + for col in 0..query.find.len() { + if !order.iter().any(|&(c, _)| c == col) { + order.push((col, false)); + } + } + plan.steps.push(Step::Sort { + cols: std::mem::take(&mut plan.find), + order, + keep: limit.map(|l| offset.saturating_add(l)), + }); + plan.find = (0..query.find.len()).collect(); + } + if limit.is_some() || offset > 0 { + plan.steps.push(Step::Limit { offset, limit }); } plan.vars = query.find.iter().map(FindSpec::display_name).collect(); plan.as_of = as_of; diff --git a/src/query/engine/sort.rs b/src/query/engine/sort.rs new file mode 100644 index 000000000..a2ae29184 --- /dev/null +++ b/src/query/engine/sort.rs @@ -0,0 +1,448 @@ +//! `Sort` (#473 spec §6.2, #306 spec §4.2): the rows of its child in the +//! full order of `:order-by`, as the `:find` columns. +//! +//! It holds each row's values of those columns, which the full order +//! compares, and turns them back into cells only for the rows it returns. +//! Under a `:limit` (`keep = k`) it holds at most `k` rows, in a binary heap +//! whose top is the worst row held, so its memory follows `k`, not the +//! answer. Either way it reads its whole input; skipping that when an index +//! returns the rows in order is #641. + +use super::batch::Batch; +use super::cell::{Cell, Scalar}; +use super::meter::{Held, Meter, growth, value_bytes}; +use super::project::value; +use super::{BATCH_ROWS, Counters, Ctx, Operator}; +use crate::graph::types::Value; +use crate::query::datalog::aggregate::row_order; +use anyhow::Result; +use std::cmp::Ordering; +use std::mem::size_of; +use std::sync::Arc; + +pub(crate) struct Sort { + child: Box, + /// The slots of the `:find` columns, in `:find` order. + cols: Vec, + /// The full order, as `(column, descending)` over `cols`' positions. + order: Vec<(usize, bool)>, + /// The most rows to hold, when a `:limit` follows. + keep: Option, + /// The rows held: a heap, worst on top, when `keep` is given. + rows: Vec>, + /// The sorted rows, last first, once the input has ended. + out: Option>>, + held: Held, + counters: Counters, +} + +/// The bytes one held row counts. +fn bytes(row: &[Value]) -> usize { + size_of::>() + row.iter().map(value_bytes).sum::() +} + +impl Sort { + pub(crate) fn new( + child: Box, + cols: Vec, + order: Vec<(usize, bool)>, + keep: Option, + meter: &Arc, + ) -> Self { + Sort { + child, + cols, + order, + keep, + rows: Vec::new(), + out: None, + held: Held::new(meter, "Sort"), + counters: Counters::new("Sort"), + } + } + + fn cmp(&self, a: &[Value], b: &[Value]) -> Ordering { + row_order(a, b, &self.order) + } + + /// Hold each row of `batch`, or, at `keep` rows, those better than the + /// worst held. + fn take(&mut self, batch: &Batch, ctx: &mut Ctx<'_>) -> Result<()> { + let null = Cell::Val(Scalar::Null); + for i in 0..batch.len { + let row = self + .cols + .iter() + .map(|&s| { + value( + batch.cols.get(s).and_then(|c| c.get(i)).unwrap_or(&null), + ctx, + ) + }) + .collect::>>()?; + if let Some(k) = self.keep + && self.rows.len() >= k + { + // Full: replace the worst row held, if this one is better. + let better = self + .rows + .first() + .is_some_and(|top| self.cmp(&row, top).is_lt()); + if better { + self.held.charge(bytes(&row))?; + if let Some(top) = self.rows.first_mut() { + let old = std::mem::replace(top, row); + self.held.uncharge(bytes(&old)); + } + self.sift_down(0); + } + continue; + } + let n = bytes(&row); + self.held.charge(n)?; + let refused = growth(self.rows.capacity(), size_of::>()); + self.held.reserve(self.rows.try_reserve(1), n, refused)?; + self.rows.push(row); + if self.keep.is_some() { + self.sift_up(self.rows.len() - 1); + } + } + Ok(()) + } + + /// Whether held row `i` comes after held row `j` (both exist). + fn after(&self, i: usize, j: usize) -> bool { + match (self.rows.get(i), self.rows.get(j)) { + (Some(a), Some(b)) => self.cmp(a, b).is_gt(), + _ => false, + } + } + + /// Restore the heap (worst on top) above `i`. + fn sift_up(&mut self, mut i: usize) { + while i > 0 { + let parent = (i - 1) / 2; + if !self.after(i, parent) { + break; + } + self.rows.swap(i, parent); + i = parent; + } + } + + /// Restore the heap (worst on top) below `i`. + fn sift_down(&mut self, mut i: usize) { + loop { + let mut worst = i; + for child in [2 * i + 1, 2 * i + 2] { + if self.after(child, worst) { + worst = child; + } + } + if worst == i { + break; + } + self.rows.swap(i, worst); + i = worst; + } + } +} + +impl Operator for Sort { + fn next_batch(&mut self, ctx: &mut Ctx<'_>) -> Result> { + if self.out.is_none() { + if self.keep == Some(0) { + self.child.close(); + } else { + while let Some(batch) = self.child.next_batch(ctx)? { + self.counters.rows_in += batch.len as u64; + self.take(&batch, ctx)?; + } + } + let mut rows = std::mem::take(&mut self.rows); + let order = &self.order; + // Rows equal under the full order are equal in every column, so + // an unstable sort gives the one order; last first, to pop. + rows.sort_unstable_by(|a, b| row_order(b, a, order)); + self.out = Some(rows); + } + let Some(rows) = self.out.as_mut().filter(|r| !r.is_empty()) else { + return Ok(None); + }; + let mut batch = Batch::new(self.cols.len()); + while batch.len < BATCH_ROWS + && let Some(row) = rows.pop() + { + self.held.uncharge(bytes(&row)); + let mut cells = row + .iter() + .map(|v| Cell::of_value(ctx.ids, ctx.keys, v).map(Some)) + .collect::>>>()?; + batch.push(&mut cells); + } + self.counters.out(&batch); + Ok(Some(batch)) + } + + fn rewind(&mut self, seed: &[Cell]) { + self.rows.clear(); + self.out = None; + self.held.release(); + self.child.rewind(seed); + } + + fn close(&mut self) { + self.rows = Vec::new(); + self.out = Some(Vec::new()); + self.held.release(); + self.child.close(); + } + + fn counters(&self, out: &mut Vec) { + out.push(Counters { + bytes_held: self.held.bytes() as u64, + ..self.counters + }); + self.child.counters(out); + } +} + +#[cfg(all(test, not(target_arch = "wasm32")))] +mod tests { + use super::*; + use crate::query::engine::Stream; + use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; + + /// The rows `rows`, two integer columns, in full batches; counts the + /// rows it returns and its closes. + struct Rows { + rows: Vec<(i64, i64)>, + next: usize, + sent: Arc, + closes: Arc, + } + + impl Operator for Rows { + fn next_batch(&mut self, _: &mut Ctx<'_>) -> Result> { + if self.next >= self.rows.len() { + return Ok(None); + } + let mut batch = Batch::new(2); + while batch.len < BATCH_ROWS && self.next < self.rows.len() { + let (a, b) = self.rows[self.next]; + batch.push(&mut [ + Some(Cell::Val(Scalar::Int(a))), + Some(Cell::Val(Scalar::Int(b))), + ]); + self.next += 1; + } + self.sent.fetch_add(batch.len, AtomicOrdering::SeqCst); + Ok(Some(batch)) + } + + fn rewind(&mut self, _: &[Cell]) { + self.next = 0; + } + + fn close(&mut self) { + self.next = self.rows.len(); + self.closes.fetch_add(1, AtomicOrdering::SeqCst); + } + + fn counters(&self, _: &mut Vec) {} + } + + /// `n` rows `(i * 7919 % n, i % 3)`: the first column a permutation of + /// `0..n` when `n` is not a multiple of 7919. + fn shuffled(n: usize) -> Vec<(i64, i64)> { + (0..n) + .map(|i| ((i * 7919 % n) as i64, (i % 3) as i64)) + .collect() + } + + struct Run { + rows: Vec, + sent: usize, + closes: usize, + peak: usize, + } + + fn run( + input: Vec<(i64, i64)>, + order: Vec<(usize, bool)>, + keep: Option, + meter: Arc, + ) -> Result { + let (sent, closes) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0))); + let child = Rows { + rows: input, + next: 0, + sent: sent.clone(), + closes: closes.clone(), + }; + let snapshot = crate::graph::storage::FactStorage::new() + .snapshot() + .unwrap(); + let sort = Sort::new(Box::new(child), vec![0, 1], order, keep, &meter); + let mut stream = Stream::for_test(snapshot, Box::new(sort)); + let mut rows = Vec::new(); + while let Some(b) = stream.next_rows()? { + for row in b { + match row.as_slice() { + // `for_test` projects the first column. + [Value::Integer(a)] => rows.push(*a), + _ => panic!("one integer column"), + } + } + } + Ok(Run { + rows, + sent: sent.load(AtomicOrdering::SeqCst), + closes: closes.load(AtomicOrdering::SeqCst), + peak: meter.peak(), + }) + } + + /// The expected first column: `input` sorted by `order`, first `keep`. + fn expected(input: &[(i64, i64)], order: &[(usize, bool)], keep: Option) -> Vec { + let mut rows: Vec> = input + .iter() + .map(|&(a, b)| vec![Value::Integer(a), Value::Integer(b)]) + .collect(); + rows.sort_by(|a, b| row_order(a, b, order)); + rows.truncate(keep.unwrap_or(usize::MAX)); + rows.iter() + .map(|r| match r.first() { + Some(Value::Integer(a)) => *a, + _ => unreachable!(), + }) + .collect() + } + + fn firsts(run: &Run) -> Vec { + run.rows.clone() + } + + #[test] + fn sorts_every_row() { + let input = shuffled(5000); + for order in [vec![(0, false), (1, false)], vec![(0, true), (1, false)]] { + let r = run(input.clone(), order.clone(), None, Meter::unlimited()).unwrap(); + assert_eq!(firsts(&r), expected(&input, &order, None)); + assert_eq!(r.rows.len(), 5000); + } + } + + #[test] + fn top_k_keeps_the_first_k_of_the_full_sort() { + let input = shuffled(5000); + for order in [vec![(0, false), (1, false)], vec![(0, true), (1, false)]] { + for k in [1, 10, 1024, 4999, 5000, 6000] { + let r = run(input.clone(), order.clone(), Some(k), Meter::unlimited()).unwrap(); + assert_eq!(firsts(&r), expected(&input, &order, Some(k)), "k {k}"); + } + } + } + + #[test] + fn ties_fall_to_the_next_column() { + // Order by the second column descending, then the first ascending. + let input = shuffled(300); + let order = vec![(1, true), (0, false)]; + for keep in [None, Some(7), Some(150)] { + let r = run(input.clone(), order.clone(), keep, Meter::unlimited()).unwrap(); + assert_eq!(firsts(&r), expected(&input, &order, keep), "{keep:?}"); + } + } + + #[test] + fn keep_zero_pulls_nothing() { + let r = run( + shuffled(5000), + vec![(0, false)], + Some(0), + Meter::unlimited(), + ) + .unwrap(); + assert!(r.rows.is_empty()); + assert_eq!(r.sent, 0, "nothing pulled"); + assert!(r.closes >= 1, "and the child closed"); + } + + #[test] + fn top_k_holds_k_rows() { + let order = vec![(0, true), (1, false)]; + let small = run(shuffled(1000), order.clone(), Some(10), Meter::unlimited()).unwrap(); + let large = run( + shuffled(100_001), + order.clone(), + Some(10), + Meter::unlimited(), + ) + .unwrap(); + assert_eq!(small.peak, large.peak, "memory follows k, not the input"); + // `k` rows, and the one replacing the worst while both are counted. + assert!(large.peak <= 11 * bytes(&[Value::Integer(0), Value::Integer(0)])); + let all = run(shuffled(100_001), order, None, Meter::unlimited()).unwrap(); + assert!(all.peak > 1000 * large.peak, "a full sort holds every row"); + } + + #[test] + fn over_the_limit_is_qry_012() { + let order = vec![(0, false)]; + let row = bytes(&[Value::Integer(0), Value::Integer(0)]); + let e = run(shuffled(5000), order.clone(), None, Meter::new(100 * row)) + .err() + .expect("a full sort over the limit"); + assert_eq!(crate::MinigrafError::from(e).code(), "QRY-012"); + let e = run( + shuffled(5000), + order.clone(), + Some(200), + Meter::new(100 * row), + ) + .err() + .expect("a top-k over the limit"); + assert_eq!(crate::MinigrafError::from(e).code(), "QRY-012"); + run(shuffled(5000), order, Some(50), Meter::new(100 * row)).unwrap(); + } + + #[test] + fn rewind_sorts_again() { + let (sent, closes) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0))); + let child = Rows { + rows: shuffled(100), + next: 50, + sent, + closes, + }; + let meter = Meter::unlimited(); + let mut sort = Sort::new( + Box::new(child), + vec![0, 1], + vec![(0, true)], + Some(3), + &meter, + ); + sort.out = Some(Vec::new()); + sort.rows.push(vec![Value::Integer(-1)]); + sort.held.charge(10).unwrap(); + sort.rewind(&[]); + assert!(sort.out.is_none() && sort.rows.is_empty(), "rewound"); + assert_eq!(meter.used(), 0, "and released"); + let snapshot = crate::graph::storage::FactStorage::new() + .snapshot() + .unwrap(); + let mut stream = Stream::for_test(snapshot, Box::new(sort)); + let mut got = Vec::new(); + while let Some(b) = stream.next_rows().unwrap() { + for row in b { + if let Some(Value::Integer(n)) = row.first() { + got.push(*n); + } + } + } + assert_eq!(got, [99, 98, 97], "the child read from its start"); + assert_eq!(meter.used(), 0, "every row released as returned"); + } +} From db1ab3645aeae6326b2daac6715668a30a5ffad0 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sun, 25 Oct 2026 03:59:21 +0530 Subject: [PATCH 06/13] fix(engine): Sort drops DICT lookups per input batch, so top-k memory follows k (#306) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01TcpoiFyFeQQQELL872LfQF --- .../plans/2026-10-24-306-limit-order-by.md | 10 ++++--- src/query/engine/mod.rs | 10 +++---- src/query/engine/sort.rs | 4 +++ src/query/engine/tests.rs | 30 +++++++++++++++++++ 4 files changed, 44 insertions(+), 10 deletions(-) diff --git a/docs/superpowers/plans/2026-10-24-306-limit-order-by.md b/docs/superpowers/plans/2026-10-24-306-limit-order-by.md index 4df69a2d7..a01db7b95 100644 --- a/docs/superpowers/plans/2026-10-24-306-limit-order-by.md +++ b/docs/superpowers/plans/2026-10-24-306-limit-order-by.md @@ -166,10 +166,12 @@ Branch `feat/306-order-by` from `v3` 59ce7f24 (PR 1 merged as #745), worktree comparator (worst row on top): `std::collections::BinaryHeap` needs `Ord` on the element, which would carry the order spec in every row. ~30 lines, unit-tested against a full sort. -12. **A `Sort` stream keeps its DICT lookups** (`Stream::hold_ids`, as for `Window`): it turns - every row's ids into values in its first pull. With top-k it holds at most `k` rows' - ids' worth of lookups beyond what the scan touched; without `keep` it holds the answer - anyway. +12. **`Sort` drops the DICT lookups after each input batch** (`IdSpace::end_pull`), and a + `Sort` stream does not set `Stream::hold_ids`. `Sort` reads its whole input in its first + pull, turning every row's ids into values; the DICT cache is per pull, so kept it would + grow with the input (10M entries at the acceptance scale), not with `keep`. The rows it + holds are values, so nothing needs the lookups again; the cache is only a cache (a DICT + id never changes), so clearing it mid-pull is safe. 13. **`:offset` alone (with `:order-by`, no `:limit`)** plans `Sort` without `keep` and `Limit { offset, limit: None }`. 14. **Error texts** (no backticks: no registry template has any, and the doc check reads diff --git a/src/query/engine/mod.rs b/src/query/engine/mod.rs index b7bd31399..adbb9a6e9 100644 --- a/src/query/engine/mod.rs +++ b/src/query/engine/mod.rs @@ -121,9 +121,9 @@ pub(crate) struct Stream { find: Vec, vars: Vec, project: Counters, - /// Keep the DICT lookups until the stream ends: a `Window` or a `Sort` - /// turns every row's ids into values in its first pull, and the rows it - /// returns later read the same ids again (#547, #306). + /// Keep the DICT lookups until the stream ends: a `Window` turns every + /// row's ids into values in its first pull, and the rows it returns + /// later read the same ids again (#547). hold_ids: bool, /// In a rule query, the relations its rules derived (#548). relations: Option, @@ -275,9 +275,7 @@ fn stream( vars, as_of: _, } = plan; - let hold_ids = steps - .iter() - .any(|s| matches!(s, plan::Step::Window { .. } | plan::Step::Sort { .. })); + let hold_ids = steps.iter().any(|s| matches!(s, plan::Step::Window { .. })); let root = build(source, steps, &meter); Stream { snapshot, diff --git a/src/query/engine/sort.rs b/src/query/engine/sort.rs index a2ae29184..d2de22392 100644 --- a/src/query/engine/sort.rs +++ b/src/query/engine/sort.rs @@ -157,6 +157,10 @@ impl Operator for Sort { while let Some(batch) = self.child.next_batch(ctx)? { self.counters.rows_in += batch.len as u64; self.take(&batch, ctx)?; + // The rows held are values, so the DICT lookups that made + // them are not needed again: kept, they would grow with + // the input, not with `keep`. + ctx.ids.end_pull(); } } let mut rows = std::mem::take(&mut self.rows); diff --git a/src/query/engine/tests.rs b/src/query/engine/tests.rs index 8a54b98e8..8139fc22a 100644 --- a/src/query/engine/tests.rs +++ b/src/query/engine/tests.rs @@ -2969,3 +2969,33 @@ fn stream_fits_the_allocator_thread_cache() { let size = std::mem::size_of::(); assert!(size <= 1032, "Stream is {size} bytes"); } + +// ── #306: Sort ────────────────────────────────────────────────────────────── + +/// Top-k holds `k` rows and one batch's DICT lookups, so it runs under a +/// limit a full sort of the same input exceeds (#306 spec §4.2). +#[test] +fn top_k_memory_follows_k_not_the_input() { + let dir = tempfile::tempdir().unwrap(); + let db = memory_db(dir.path(), 20_000, 512 * 1024); + let rows = |r: QueryResult| match r { + QueryResult::QueryResults { results, .. } => results, + _ => panic!("rows"), + }; + let top = rows( + db.execute("(query [:find ?e ?v :where [?e :a ?v] :order-by [?v :desc] :limit 10])") + .unwrap(), + ); + let firsts: Vec = top + .iter() + .map(|r| match r.get(1) { + Some(crate::Value::Integer(n)) => *n, + _ => panic!("an integer"), + }) + .collect(); + assert_eq!(firsts, (19_990..20_000).rev().collect::>()); + let e = db + .execute("(query [:find ?e ?v :where [?e :a ?v] :order-by [?v :desc]])") + .expect_err("a full sort over the limit"); + assert_eq!(crate::MinigrafError::from(e).code(), "QRY-012"); +} From 930beafa2b67d02fa8ec217453e5a22c49ada715 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sun, 25 Oct 2026 04:00:33 +0530 Subject: [PATCH 07/13] test: ordered :order-by/:offset answers on every path (#306) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01TcpoiFyFeQQQELL872LfQF --- tests/limit_order_test.rs | 330 +++++++++++++++++++++++++++++++++++++- 1 file changed, 329 insertions(+), 1 deletion(-) diff --git a/tests/limit_order_test.rs b/tests/limit_order_test.rs index fe936a2b3..ec5ebfcda 100644 --- a/tests/limit_order_test.rs +++ b/tests/limit_order_test.rs @@ -1,4 +1,4 @@ -//! `:limit` (#306, spec `docs/superpowers/specs/2026-10-24-limit-order-by-design.md`). +//! `:limit`, `:offset` and `:order-by` (#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 @@ -8,6 +8,7 @@ #![cfg(all(feature = "test-hooks", not(target_arch = "wasm32")))] use minigraf::{BindValue, Minigraf, QueryResult, Value}; +use std::cmp::Ordering; use std::collections::HashMap; /// A value as text, without `Debug` (CodeQL `rust/cleartext-logging`). @@ -209,3 +210,330 @@ fn limit_slot_errors() { assert_eq!(code(BindValue::Val(Value::String("2".into()))), "API-024"); assert_eq!(code(BindValue::TxCount(2)), "API-024"); } + +// ── :order-by and :offset ───────────────────────────────────────────────── + +/// The class of a value in the full order (spec §2). +fn class(v: &Value) -> u8 { + match v { + Value::Integer(_) | Value::Float(_) => 0, + Value::String(_) => 1, + Value::Boolean(_) => 2, + Value::Ref(_) => 3, + Value::Keyword(_) => 4, + Value::Null => 5, + _ => 6, + } +} + +/// The test's own order of values: numbers numerically (an integer before +/// an equal float, `-0.0` before `0.0`), then by class. +fn order(a: &Value, b: &Value) -> Ordering { + let num = |v: &Value| match v { + Value::Integer(i) => Some((*i as f64, 0)), + Value::Float(f) => Some((*f, 1)), + _ => None, + }; + match (a, b, num(a), num(b)) { + (_, _, Some((x, kx)), Some((y, ky))) => x + .partial_cmp(&y) + .unwrap() + .then(kx.cmp(&ky)) + .then(x.total_cmp(&y)), + (Value::String(x), Value::String(y), ..) => x.cmp(y), + (Value::Boolean(x), Value::Boolean(y), ..) => x.cmp(y), + (Value::Ref(x), Value::Ref(y), ..) => x.as_bytes().cmp(y.as_bytes()), + (Value::Keyword(x), Value::Keyword(y), ..) => x.cmp(y), + _ => class(a).cmp(&class(b)), + } +} + +/// `rows` in the full order: `keys` (column, descending), then every other +/// column ascending. +fn full_sort(mut rows: Vec>, keys: &[(usize, bool)]) -> Vec> { + let width = rows.first().map_or(0, Vec::len); + let mut all = keys.to_vec(); + for c in 0..width { + if !all.iter().any(|&(k, _)| k == c) { + all.push((c, false)); + } + } + rows.sort_by(|a, b| { + all.iter() + .map(|&(c, desc)| { + let o = order(&a[c], &b[c]); + if desc { o.reverse() } else { o } + }) + .find(|o| o.is_ne()) + .unwrap_or(Ordering::Equal) + }); + rows +} + +fn values(r: QueryResult) -> Vec> { + match r { + QueryResult::QueryResults { results, .. } => results, + _ => panic!("expected QueryResults"), + } +} + +/// Ordered shapes: a body, its `:order-by` vector, and the keys as +/// `(column, descending)`. +const ORDERED: &[(&str, &str, &[(usize, bool)])] = &[ + ( + ":find ?e ?a :where [?e :age ?a]", + "[?a :desc]", + &[(1, true)], + ), + (":find ?a :where [?e :age ?a]", "[?a]", &[(0, false)]), + ( + ":find ?n ?f :where [?e :name ?n] [?e :follows ?f]", + "[?f :desc ?n]", + &[(1, true), (0, false)], + ), + ( + ":find ?e :where [?e :age ?a] (not [?e :status :inactive])", + "[?e :desc]", + &[(0, true)], + ), + ( + ":find ?e :where (or [?e :status :inactive] [?e :age 3])", + "[?e]", + &[(0, false)], + ), + ( + ":find ?s (count ?e) :where [?e :status ?s]", + "[(count ?e) :desc]", + &[(1, true)], + ), + ( + ":find (count ?e) :where [?e :age ?a]", + "[(count ?e)]", + &[(0, false)], + ), + ( + ":find (count ?e) :where [?e :no-such ?a]", + "[(count ?e) :desc]", + &[(0, true)], + ), + ( + ":find ?e (rank :over (:order-by ?a)) :where [?e :age ?a]", + "[(rank :over (:order-by ?a)) :desc]", + &[(1, true)], + ), + (":find ?y :where (reach :p0 ?y)", "[?y :desc]", &[(0, true)]), + ( + ":find ?x ?y :where (reach ?x ?y)", + "[?y ?x :desc]", + &[(1, false), (0, true)], + ), + ( + ":find ?e :as-of 1 :where [?e :age ?a]", + "[?e]", + &[(0, false)], + ), + ( + ":find ?e ?a :any-valid-time :where [?e :age ?a]", + "[?a :desc]", + &[(1, true)], + ), + (":find ?x :where [(+ 1 2) ?x]", "[?x]", &[(0, false)]), +]; + +/// Every row of `cursor`, `n` at a time. +fn drain(mut cursor: minigraf::Cursor, n: usize) -> Vec> { + let mut rows = Vec::new(); + while let Some(batch) = cursor.next_batch(n).expect("batch") { + rows.extend(keys(batch.rows())); + } + rows +} + +fn check_ordered(db: &Minigraf) { + for (body, by, sort_keys) in ORDERED { + let mut lists: Vec>> = Vec::new(); + for reorder in [true, false] { + let _g = minigraf::__reorder(reorder); + let answer = values(db.execute(&format!("(query [{body}])")).expect(body)); + let sorted = keys(&full_sort(answer, sort_keys)); + let total = sorted.len(); + let prepared = db + .prepare(&format!( + "(query [{body} :order-by {by} :offset $o :limit $n])" + )) + .expect(body); + let mut cases: Vec<(Option, Option)> = vec![(None, None)]; + for n in [0, 1, 3, total] { + cases.push((None, Some(n))); + for o in [1, total.saturating_sub(1), total + 2] { + cases.push((Some(o), Some(n))); + } + } + cases.push((Some(2), None)); + for (offset, limit) in cases { + let mut q = format!("{body} :order-by {by}"); + if let Some(o) = offset { + q.push_str(&format!(" :offset {o}")); + } + if let Some(n) = limit { + q.push_str(&format!(" :limit {n}")); + } + let q = format!("(query [{q}])"); + let from = offset.unwrap_or(0).min(total); + let to = limit.map_or(total, |n| (from + n).min(total)); + let want = &sorted[from..to]; + let executed = keys(&values(db.execute(&q).expect(&q))); + let pulled = drain(db.query(&q).expect(&q), 1); + assert_eq!(executed, want, "{q} via execute, reorder {reorder}"); + assert_eq!(pulled, want, "{q} via cursor, reorder {reorder}"); + if let Some(n) = limit { + let bind = [ + ( + "o", + BindValue::Val(Value::Integer(offset.unwrap_or(0) as i64)), + ), + ("n", BindValue::Val(Value::Integer(n as i64))), + ]; + let got = keys(&values(prepared.execute(&bind).expect(&q))); + assert_eq!(got, want, "{q} via prepared, reorder {reorder}"); + let got = drain(prepared.query(&bind).expect(&q), 3); + assert_eq!(got, want, "{q} via prepared cursor, reorder {reorder}"); + } + } + lists.push(sorted); + } + assert_eq!( + lists[0], lists[1], + "{body}: one order with reordering on and off" + ); + } +} + +#[test] +fn order_by_in_memory() { + let db = Minigraf::in_memory().unwrap(); + people(&db, false); + check_ordered(&db); +} + +#[test] +fn order_by_on_committed_and_pending_facts() { + let dir = tempfile::tempdir().unwrap(); + let db = Minigraf::open(dir.path().join("order.graph")).unwrap(); + people(&db, true); + check_ordered(&db); +} + +/// Pages of 3 read the whole ordered answer once each. +#[test] +fn pages_cover_the_ordered_answer() { + let db = Minigraf::in_memory().unwrap(); + people(&db, false); + let body = ":find ?n ?a :where [?e :name ?n] [?e :age ?a] :order-by [?a :desc]"; + let whole = results(db.execute(&format!("(query [{body}])")).unwrap()); + let mut paged = Vec::new(); + for page in 0.. { + let q = format!("(query [{body} :offset {} :limit 3])", page * 3); + let rows = results(db.execute(&q).unwrap()); + if rows.is_empty() { + break; + } + paged.extend(rows); + } + assert_eq!(paged, whole); + assert_eq!(whole.len(), 59, "every person with an age"); +} + +/// Values of every class in one column, ascending and descending. +#[test] +fn order_by_every_value_class() { + let db = Minigraf::in_memory().unwrap(); + db.execute( + "(transact [[:v0 :v 1] [:v1 :v 1.0] [:v2 :v -0.0] [:v3 :v 0.0] [:v4 :v -5] \ + [:v5 :v 2.5] [:v6 :v \"a\"] [:v7 :v \"b\"] [:v8 :v true] [:v9 :v false] \ + [:v10 :v :k/a] [:v11 :v :k/b] [:v12 :v #{1 2}] \ + [:v13 :v #uuid \"550e8400-e29b-41d4-a716-446655440000\"]])", + ) + .unwrap(); + let answer = values(db.execute("(query [:find ?v :where [?e :v ?v]])").unwrap()); + assert_eq!(answer.len(), 14); + for (by, desc) in [("[?v]", false), ("[?v :desc]", true)] { + let got = keys(&values( + db.execute(&format!( + "(query [:find ?v :where [?e :v ?v] :order-by {by}])" + )) + .unwrap(), + )); + let want = keys(&full_sort(answer.clone(), &[(0, desc)])); + assert_eq!(got, want, "{by}"); + } + let first = keys(&values( + db.execute("(query [:find ?v :where [?e :v ?v] :order-by [?v] :limit 4])") + .unwrap(), + )); + assert_eq!( + first, + [ + ["i:-5"], + ["f:8000000000000000"], + ["f:0000000000000000"], + ["i:1"] + ] + ); +} + +/// A write transaction's staged facts are ordered with the rest. +#[test] +fn order_by_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 101]])") + .unwrap(); + let top = results( + tx.execute("(query [:find ?a :where [?e :age ?a] :order-by [?a :desc] :limit 2])") + .unwrap(), + ); + assert_eq!(top, [["i:101"], ["i:100"]]); + tx.commit().unwrap(); +} + +/// `:offset` slots: unbound is `API-010`, a bad value is rejected at execute. +#[test] +fn offset_slot_errors() { + let db = Minigraf::in_memory().unwrap(); + people(&db, false); + let q = "(query [:find ?e :where [?e :age ?a] :order-by [?e] :offset $o])"; + assert_eq!(db.execute(q).expect_err("unbound").code(), "API-010"); + let p = db.prepare(q).unwrap(); + let code = |b: BindValue| p.execute(&[("o", b)]).expect_err("rejected").code(); + assert_eq!(code(BindValue::Val(Value::Integer(-2))), "PRS-115"); + assert_eq!(code(BindValue::TxCount(2)), "API-024"); +} + +/// A full sort holds the answer; top-k under a `:limit` holds `k` rows. +#[test] +fn sort_memory_limit() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("sort.graph"); + { + let db = Minigraf::open(&path).unwrap(); + let facts: String = (0..20_000).map(|i| format!("[:p{i} :a {i}]")).collect(); + db.execute(&format!("(transact [{facts}])")).unwrap(); + db.checkpoint().unwrap(); + } + let db = minigraf::OpenOptions::new() + .query_memory_limit(512 * 1024) + .path(&path) + .open() + .unwrap(); + let e = db + .execute("(query [:find ?e ?v :where [?e :a ?v] :order-by [?v]])") + .expect_err("over the limit"); + assert_eq!(e.code(), "QRY-012"); + let top = results( + db.execute("(query [:find ?v :where [?e :a ?v] :order-by [?v :desc] :limit 2])") + .unwrap(), + ); + assert_eq!(top, [["i:19999"], ["i:19998"]]); +} From 9ec29403f6bffc5a4a1324991c19f61053a9c080 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sun, 25 Oct 2026 04:02:41 +0530 Subject: [PATCH 08/13] test(grammar): :order-by and :offset sections, verifier tokens and witnesses (#306) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01TcpoiFyFeQQQELL872LfQF --- docs/GRAMMAR.md | 7 +++- src/query/datalog/parser/verifier.rs | 32 ++++++++++++++++--- tests/grammar/grammar.pest | 8 ++++- .../semantic/query_offset_negative.edn | 4 +++ .../query_offset_without_order_by.edn | 3 ++ .../semantic/query_order_by_not_in_find.edn | 3 ++ .../semantic/query_order_by_twice_entry.edn | 3 ++ .../invalid/syntax/query_order_by_empty.edn | 3 ++ .../syntax/query_order_by_leading_desc.edn | 3 ++ .../syntax/query_order_by_not_vector.edn | 3 ++ tests/grammar/valid/query_offset_slot.edn | 5 +++ tests/grammar/valid/query_order_by.edn | 3 ++ .../valid/query_order_by_desc_aggregate.edn | 3 ++ .../valid/query_order_by_offset_limit.edn | 5 +++ 14 files changed, 78 insertions(+), 7 deletions(-) create mode 100644 tests/grammar/invalid/semantic/query_offset_negative.edn create mode 100644 tests/grammar/invalid/semantic/query_offset_without_order_by.edn create mode 100644 tests/grammar/invalid/semantic/query_order_by_not_in_find.edn create mode 100644 tests/grammar/invalid/semantic/query_order_by_twice_entry.edn create mode 100644 tests/grammar/invalid/syntax/query_order_by_empty.edn create mode 100644 tests/grammar/invalid/syntax/query_order_by_leading_desc.edn create mode 100644 tests/grammar/invalid/syntax/query_order_by_not_vector.edn create mode 100644 tests/grammar/valid/query_offset_slot.edn create mode 100644 tests/grammar/valid/query_order_by.edn create mode 100644 tests/grammar/valid/query_order_by_desc_aggregate.edn create mode 100644 tests/grammar/valid/query_order_by_offset_limit.edn diff --git a/docs/GRAMMAR.md b/docs/GRAMMAR.md index 0c5e45e8b..fc201992d 100644 --- a/docs/GRAMMAR.md +++ b/docs/GRAMMAR.md @@ -20,6 +20,7 @@ kw-any-valid-time ::= ":any-valid-time" (* not followed by word-cont *) (* t 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-offset ::= ":offset" (* 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 *) @@ -77,7 +78,8 @@ 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 | limit-section + | max-results-section | limit-section | offset-section + | order-by-section (* ── FIND SECTION ── *) find-section ::= kw-find find-spec+ @@ -124,6 +126,9 @@ 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) +offset-section ::= kw-offset (integer-lit | bind-slot) +order-by-section ::= kw-order-by "[" order-entry+ "]" +order-entry ::= find-spec (kw-asc | kw-desc)? (* ── RULE ── *) rule-cmd ::= "(" "rule" rule-vector ")" diff --git a/src/query/datalog/parser/verifier.rs b/src/query/datalog/parser/verifier.rs index c39591ece..f2ea46c66 100644 --- a/src/query/datalog/parser/verifier.rs +++ b/src/query/datalog/parser/verifier.rs @@ -66,7 +66,7 @@ fn alphabet() -> Vec { a } -const TOKENS: [&str; 34] = [ +const TOKENS: [&str; 37] = [ "(", ")", "[", @@ -89,6 +89,9 @@ const TOKENS: [&str; 34] = [ ":over", ":order-by", ":limit", + ":offset", + ":asc", + ":desc", ":valid-from", ":foo", "?x", @@ -110,7 +113,7 @@ fn bracket(t: u8) -> Option<(u8, bool)> { /// `(prefix, suffix)`: the hole sits between them. Each puts the hole where a short /// filler can complete a clause. -const TEMPLATES: [(&str, &str); 15] = [ +const TEMPLATES: [(&str, &str); 16] = [ ("(query [", "])"), ("(query [:find", ":where [?x :a ?y]])"), ("(query [:find ?x :where [?x :a ?y]", "])"), @@ -120,6 +123,7 @@ const TEMPLATES: [(&str, &str); 15] = [ ("(query [:find ?x :where [?x :a ?y] (or-join [?x]", ")])"), ("(query [:find ?x :where [?x :a ?y] (or (and", "))])"), ("(query [:find ?x (sum ?y :over (", ")) :where [?x :a ?y]])"), + ("(query [:find ?x :where [?x :a ?y] :order-by [", "]])"), ("(transact [", "])"), ("(transact [[:a :b 1", "]])"), ("(transact {", "} [[:a :b 1]])"), @@ -218,6 +222,23 @@ const SEMANTIC: &[Semantic] = &[ code: "PRS-115", witness: "(query [:find ?x :where [?x :a ?y] :limit -1])", }, + // parse_query: `:offset` without `:order-by` (#306). + Semantic { + code: "PRS-116", + witness: "(query [:find ?x :where [?x :a ?y] :offset 1])", + }, + // parse_query: an `:order-by` entry no `:find` element equals, such as a + // variable bound only in `:where`, or a list that is no aggregate (#306). + Semantic { + code: "PRS-117", + witness: "(query [:find ?x :where [?x :a ?y] :order-by [?y]])", + }, + // parse_query: an `:order-by` entry given twice (`order_entry+` allows a + // repeat, #306). + Semantic { + code: "PRS-118", + witness: "(query [:find ?x :where [?x :a ?y] :order-by [?x ?x]])", + }, // parse_rule: a rule body with no pattern or rule call (`rule_vector` allows // `where_clause*`). Semantic { @@ -510,9 +531,10 @@ fn filler_counts_match_the_spec() { } n }; - // 34 tokens: the #643 spec's 32, `#{` (#316) and `:limit` (#306). - assert_eq!(count(3), 23_020); - assert_eq!(count(4), 651_806); + // 37 tokens: the #643 spec's 32, `#{` (#316), and `:limit`, `:offset`, `:asc` + // and `:desc` (#306). + assert_eq!(count(3), 31_066); + assert_eq!(count(4), 971_903); } #[test] diff --git a/tests/grammar/grammar.pest b/tests/grammar/grammar.pest index 5e7c7ccf3..a76386141 100644 --- a/tests/grammar/grammar.pest +++ b/tests/grammar/grammar.pest @@ -20,6 +20,7 @@ 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_offset = @{ ":offset" ~ !word_cont } kw_over = @{ ":over" ~ !word_cont } kw_partition_by = @{ ":partition-by" ~ !word_cont } kw_order_by = @{ ":order-by" ~ !word_cont } @@ -129,7 +130,8 @@ 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 | limit_section + max_derived_facts_section | max_results_section | limit_section | + offset_section | order_by_section } // ── FIND SECTION ────────────────────────────────────────────────────────── @@ -208,6 +210,10 @@ 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) } +offset_section = { kw_offset ~ (integer_lit | bind_slot) } +// :order-by names :find elements, each optionally followed by a direction. +order_by_section = { kw_order_by ~ "[" ~ order_entry+ ~ "]" } +order_entry = { find_spec ~ (kw_asc | kw_desc)? } // ── RULE ────────────────────────────────────────────────────────────────── // (rule [(head ?args) body-clause ...]) diff --git a/tests/grammar/invalid/semantic/query_offset_negative.edn b/tests/grammar/invalid/semantic/query_offset_negative.edn new file mode 100644 index 000000000..6e8b370fd --- /dev/null +++ b/tests/grammar/invalid/semantic/query_offset_negative.edn @@ -0,0 +1,4 @@ +(query [:find ?n + :where [?e :person/name ?n] + :order-by [?n] + :offset -1]) diff --git a/tests/grammar/invalid/semantic/query_offset_without_order_by.edn b/tests/grammar/invalid/semantic/query_offset_without_order_by.edn new file mode 100644 index 000000000..7976ca767 --- /dev/null +++ b/tests/grammar/invalid/semantic/query_offset_without_order_by.edn @@ -0,0 +1,3 @@ +(query [:find ?n + :where [?e :person/name ?n] + :offset 10]) diff --git a/tests/grammar/invalid/semantic/query_order_by_not_in_find.edn b/tests/grammar/invalid/semantic/query_order_by_not_in_find.edn new file mode 100644 index 000000000..28cdc073d --- /dev/null +++ b/tests/grammar/invalid/semantic/query_order_by_not_in_find.edn @@ -0,0 +1,3 @@ +(query [:find ?e + :where [?e :person/name ?n] + :order-by [?n]]) diff --git a/tests/grammar/invalid/semantic/query_order_by_twice_entry.edn b/tests/grammar/invalid/semantic/query_order_by_twice_entry.edn new file mode 100644 index 000000000..b7bf82fdf --- /dev/null +++ b/tests/grammar/invalid/semantic/query_order_by_twice_entry.edn @@ -0,0 +1,3 @@ +(query [:find ?n + :where [?e :person/name ?n] + :order-by [?n :asc ?n :desc]]) diff --git a/tests/grammar/invalid/syntax/query_order_by_empty.edn b/tests/grammar/invalid/syntax/query_order_by_empty.edn new file mode 100644 index 000000000..55df56401 --- /dev/null +++ b/tests/grammar/invalid/syntax/query_order_by_empty.edn @@ -0,0 +1,3 @@ +(query [:find ?n + :where [?e :person/name ?n] + :order-by []]) diff --git a/tests/grammar/invalid/syntax/query_order_by_leading_desc.edn b/tests/grammar/invalid/syntax/query_order_by_leading_desc.edn new file mode 100644 index 000000000..6795b9287 --- /dev/null +++ b/tests/grammar/invalid/syntax/query_order_by_leading_desc.edn @@ -0,0 +1,3 @@ +(query [:find ?n + :where [?e :person/name ?n] + :order-by [:desc ?n]]) diff --git a/tests/grammar/invalid/syntax/query_order_by_not_vector.edn b/tests/grammar/invalid/syntax/query_order_by_not_vector.edn new file mode 100644 index 000000000..2d9546c9d --- /dev/null +++ b/tests/grammar/invalid/syntax/query_order_by_not_vector.edn @@ -0,0 +1,3 @@ +(query [:find ?n + :where [?e :person/name ?n] + :order-by ?n]) diff --git a/tests/grammar/valid/query_offset_slot.edn b/tests/grammar/valid/query_offset_slot.edn new file mode 100644 index 000000000..7d87de486 --- /dev/null +++ b/tests/grammar/valid/query_offset_slot.edn @@ -0,0 +1,5 @@ +(query [:find ?n + :where [?e :person/name ?n] + :order-by [?n] + :offset $page + :limit $size]) diff --git a/tests/grammar/valid/query_order_by.edn b/tests/grammar/valid/query_order_by.edn new file mode 100644 index 000000000..c42daf0a1 --- /dev/null +++ b/tests/grammar/valid/query_order_by.edn @@ -0,0 +1,3 @@ +(query [:find ?e ?n + :where [?e :person/name ?n] + :order-by [?n]]) diff --git a/tests/grammar/valid/query_order_by_desc_aggregate.edn b/tests/grammar/valid/query_order_by_desc_aggregate.edn new file mode 100644 index 000000000..8d4409d4f --- /dev/null +++ b/tests/grammar/valid/query_order_by_desc_aggregate.edn @@ -0,0 +1,3 @@ +(query [:find ?n (count ?e) + :where [?e :person/name ?n] + :order-by [(count ?e) :desc ?n :asc]]) diff --git a/tests/grammar/valid/query_order_by_offset_limit.edn b/tests/grammar/valid/query_order_by_offset_limit.edn new file mode 100644 index 000000000..408d5850e --- /dev/null +++ b/tests/grammar/valid/query_order_by_offset_limit.edn @@ -0,0 +1,5 @@ +(query [:order-by [?n :desc] + :offset 20 + :find ?n + :where [?e :person/name ?n] + :limit 10]) From 21ccfe1fa8e06a04492ebc4482e92d7080847f0b Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sun, 25 Oct 2026 04:05:04 +0530 Subject: [PATCH 09/13] test: top-k acceptance over 10M facts and envelope shape top_k (#306) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01TcpoiFyFeQQQELL872LfQF --- examples/perf_envelope.rs | 6 +++ tests/limit_acceptance_test.rs | 95 ++++++++++++++++++++++++---------- 2 files changed, 74 insertions(+), 27 deletions(-) diff --git a/examples/perf_envelope.rs b/examples/perf_envelope.rs index 983344819..ff4c4427a 100644 --- a/examples/perf_envelope.rs +++ b/examples/perf_envelope.rs @@ -666,6 +666,12 @@ fn shapes(e: usize, tx_half: u64, tx_half_time: &str, hot_tx_half: u64) -> Vec usize { @@ -40,9 +45,39 @@ fn proc_status_kb(field: &str) -> usize { .unwrap_or(0) } +/// Run `query` on `path` under `limit`: its rows, time and the peak RSS it +/// added, when measurable. +fn measure( + path: &std::path::Path, + limit: usize, + query: &str, +) -> (Vec>, std::time::Duration, Option) { + let db = OpenOptions::new() + .query_memory_limit(limit) + .path(path) + .open() + .unwrap(); + 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!( + "{query}: {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"); + }; + (results, elapsed, (reset && peak_kb > 0).then_some(added)) +} + #[test] #[ignore] -fn limit_10_over_10m_facts() { +fn limit_and_top_k_over_10m_facts() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("limit.graph"); // `LogWriter` writes the records straight into the trees in batched @@ -64,31 +99,37 @@ fn limit_10_over_10m_facts() { .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) - .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 + eprintln!("10M facts: built in {:.2?}", start.elapsed()); + let (rows, elapsed, added) = measure( + &path, + MEMORY_LIMIT, + "(query [:find ?e ?v :where [?e :a ?v] :limit 10])", ); - let QueryResult::QueryResults { results, .. } = result else { - panic!("expected query results"); - }; - assert_eq!(results.len(), 10, "ten rows"); + assert_eq!(rows.len(), 10, "ten rows"); assert!(elapsed < MAX_TIME, "stops once it has its rows"); - if reset && peak_kb > 0 { + if let Some(added) = added { assert!(added < MAX_RSS, "holds no rows"); } + + let (rows, _, added) = measure( + &path, + TOP_K_MEMORY_LIMIT, + "(query [:find ?e ?v :where [?e :a ?v] :order-by [?v :desc] :limit 10])", + ); + let top: Vec = rows + .iter() + .map(|r| match r.get(1) { + Some(Value::Integer(n)) => *n, + _ => panic!("an integer"), + }) + .collect(); + let n = FACTS as i64; + assert_eq!( + top, + ((n - 10)..n).rev().collect::>(), + "the ten largest" + ); + if let Some(added) = added { + assert!(added < MAX_RSS, "holds ten rows"); + } } From eaeaa4f89544da820046c5d96df30937b2629918 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sun, 25 Oct 2026 04:06:46 +0530 Subject: [PATCH 10/13] docs: :order-by, :offset and Sort (#306) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01TcpoiFyFeQQQELL872LfQF --- CHANGELOG.md | 3 ++- CLAUDE.md | 10 +++++----- README.md | 2 +- ROADMAP.md | 2 +- docs/TEST_COVERAGE.md | 9 ++++++--- 5 files changed, 15 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3ebe1571e..0d41131f1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -51,7 +51,8 @@ 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. +- **`:limit N` bounds the rows a query returns** (#306). `[: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` 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. +- **`:order-by` and `:offset`** (#306). `:order-by [(count ?e) :desc ?name]` orders the rows a query returns by `:find` elements (plain variables, or aggregates and windows written as in `:find`), each `:asc` (the default) or `:desc`; ties are broken by the other `:find` columns ascending, so every position of an ordered answer is decided and every plan returns the same list. Values compare as windows and `min`/`max` compare them: numbers by value (an integer before an equal float, `-0.0` before `0.0`, NaN after every number), then strings, booleans, refs, keywords, `nil` and sets. `:offset N` skips the first `N` rows of the ordered answer, before `:limit`, so `:offset 20 :limit 10` is the third page of ten. A new `Sort` operator reads its whole input; under a `:limit` it keeps only the best `offset + limit` rows, so `:order-by [?v :desc] :limit 10` over 10M facts holds ten rows, not ten million. It applies after aggregates and windows, to rule queries, cursors and prepared queries (`:offset $o`, bound like `:limit`), and counts toward `query_memory_limit`. New codes: `PRS-116` (`:offset` without `:order-by`: without an order, pages could overlap or miss rows), `PRS-117` (an entry that is not a `:find` element) and `PRS-118` (`:order-by` is not a non-empty vector, names an element twice, or has a misplaced or unknown keyword). `PRS-115` now also covers `:offset`. Inside a window's `:over`, `:order-by` is unchanged. - **Set values: `#{...}`** (#316, PR 1 of 3). A fact's value can be a finite set of scalars, `[:alice :user/scopes #{:scope/read :scope/write}]`, written in query text, returned as `Value::Set(ValueSet)`, transacted, retracted and queried by exact value (`[?e :user/scopes #{:scope/write :scope/read}]`), joined on and grouped by. Duplicates are removed and element order does not matter. Two elements are the same when `Value`'s order says so: `0.0` and `-0.0` are one element (`-0.0` is kept), every NaN is one element, a keyword never equals a ref and `1` never equals `1.0`. A set cannot hold a set (`ValueSet::new` fails with the new `API-021`; in query text, `PRS-119`); its elements are literals only (`PRS-120`), and an unclosed `#{` is `PRS-121`. In format v8 a set has value tag `0x08`, after refs; a set whose encoded elements take at most 64 bytes is stored in its index keys, a longer one once in a value page, as a long string is. On a file database a stored set may take up to 4,068 bytes, counting each keyword element as 6 bytes and each ref as 10 (`WAL-003`). The REPL prints sets as `#{...}` and `BrowserDb` returns them as JSON arrays, in `Value`'s order. - **Set-membership predicates** (#316, PR 2 of 3). `[(in? ?role #{:role/admin :role/owner})]` keeps rows whose value is in a set, and `not-in?`, `any-in?` (the two sets share an element), `all-in?` (every element of the first is in the second) and `none-in?` (they share none) complete the five. Each works as a filter or binds a Boolean, `[(any-in? ?scopes #{:scope/write}) ?can-write]`, and takes literal sets, set-valued attributes and bind slots (`BindValue::Val(Value::Set(..))`). Membership is set equality, as for set values: `0.0` is in `#{-0.0}`, NaN is in a set holding NaN, a keyword is not a ref. `in?` of an empty set is false and `not-in?` true; `all-in?` and `none-in?` of an empty first set are true and `any-in?` false. A set where a scalar is needed, or a scalar where a set is needed, is a type mismatch: the filter drops the row and the binding binds nothing, so `not-in?` with a non-set is never true. `contains?` stays string-only. The five names cannot be registered as functions. - **`collect-set`** (#316, PR 3 of 3). `[(collect-set ?e :user/scope) ?scopes]` binds, per row, the set of the values of `[?e :user/scope ?v]` under the query's `:as-of` and `:valid-at`, with retractions applied as for that pattern, and `#{}` when there are none (the row is kept). Combine it with the set predicates: `[(collect-set ?e :user/scope) ?s] [(in? :scope/admin ?s)]`. `?e` must be bound by an earlier clause (`INT-018`); a keyword names its entity, and a value that is no entity gives `#{}`. It works in `:where`, rule bodies, `not-join` bodies and `or` branches; when its output variable is already bound, it keeps the rows where the two sets are equal. It reads stored facts only, never a rule's derived facts. New codes: `PRS-122` (no output variable), `PRS-123` (an operand that is not a `?variable` then an attribute keyword, a bind slot included) and `QRY-015` (a value found is itself a set: bind a set-valued attribute with a pattern). The set it builds counts toward `query_memory_limit` (`QRY-012`). `collect-set` is read only as a whole clause (`PRS-067` inside another expression) and cannot be registered as a function. diff --git a/CLAUDE.md b/CLAUDE.md index 7ccb1770b..9a18a0c94 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -135,19 +135,19 @@ cargo run < demos/demo_negation.txt 3. **`src/query/datalog/`** — Datalog engine - `parser.rs`: EDN/Datalog parser — `transact`, `retract`, `query`, `rule`, `:as-of`, `:valid-at`, `not`, `not-join`; `resolve_pseudo_sources` names each per-fact pseudo-attribute's source (nearest earlier same-entity pattern, lexical scope) and its hidden metadata variable `?#.`, else PRS-081 (#594) - `parser/statement.rs`: `transact`/`retract` read one fact at a time (#550): `kind_of` dispatches on the first two tokens; one pass keeps today's error order (phases 1 tokens … 6 valid windows, the earliest phase's first error, phase 3 in today's check order); `read_statement` (pass 1) keeps at most `TX_CHUNK_FACTS` materialised facts and the offset after the last, and returns phase 5–6 errors from `Statement::facts`/`Statement::chunks` (one `Chunk` at a time; pass 2 from that offset for a larger statement); `set_tx_chunk_facts` for tests (`minigraf::__tx_chunk_facts` and `__statement_facts_high_water` under `test-hooks`). The `Parser` reads tokens on demand from a `Lexer` (`keep` off checks structure without holding it); `parse_transact`/`parse_retract` remain under `cfg(test)` for the differential tests - - `parser/verifier.rs`: test-only (not wasm32) bounded-exhaustive check of the parser against `tests/grammar/grammar.pest` (#643): every 1–3 character string (`parse_edn`) and every balanced token filler up to `MINIGRAF_GRAMMAR_BOUND` (default 3) in 15 command templates (`parse_datalog_command`); pest rejects ⇒ parser rejects, pest accepts ⇒ parser accepts or raises a `SEMANTIC` code; today's drift is `KNOWN_DRIFT`, one entry per cause, removed by its fix (empty: #649, #650, #651 fixed) + - `parser/verifier.rs`: test-only (not wasm32) bounded-exhaustive check of the parser against `tests/grammar/grammar.pest` (#643): every 1–3 character string (`parse_edn`) and every balanced token filler up to `MINIGRAF_GRAMMAR_BOUND` (default 3) in 16 command templates (`parse_datalog_command`); pest rejects ⇒ parser rejects, pest accepts ⇒ parser accepts or raises a `SEMANTIC` code; today's drift is `KNOWN_DRIFT`, one entry per cause, removed by its fix (empty: #649, #650, #651 fixed) - `executor.rs`: `DatalogExecutor` — `transact`/`retract`/`rule` commands, and every query on the engine (`open_query`, `open_rule_query`; planner errors after the pseudo-attribute check, then every predicate the query and its rules name, `QRY-007`); its own query path is gone (#551) - `matcher.rs`: `edn_to_value`, `edn_to_entity_id` - - `aggregate.rs`: `Agg` (one aggregate of one group) and `Running` (a window's running value), used by the engine; exact sums (`i128` integers saturated once, fixed-point floats rounded once), `value_order` (one total order of values for window sorts, peers, `rank` and `min`/`max`), `named_vars` (a window's tie-break variables), `windows` (each window's value per row: partitions, sort, peers) (#547) + - `aggregate.rs`: `Agg` (one aggregate of one group) and `Running` (a window's running value), used by the engine; exact sums (`i128` integers saturated once, fixed-point floats rounded once), `value_order` (one total order of values for window sorts, peers, `rank`, `min`/`max` and `:order-by`), `row_order` (#306), `named_vars` (a window's tie-break variables), `windows` (each window's value per row: partitions, sort, peers) (#547) - `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`, `Count` (`:limit` literal or slot, #306), `WhereClause` (incl. `Not`, `NotJoin`); `PseudoAttr` enum, `AttributeSpec` wrapper + - `types.rs`: `EdnValue`, `Pattern`, `DatalogQuery`, `AsOf`, `ValidAt`, `Count` (`:limit`/`:offset` literal or slot, #306), `DatalogQuery::order_by` (`:find` elements with `Order`, #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`; `Entity` and `Val(Ref)` fit each other's slots, bind mistakes are `API-022`/`023`/`024` (#181); a `:limit $n` slot takes `Val(Integer(n >= 0))` (`resolve_count_slot`, #306) + - `prepared.rs`: `BindValue`, `PreparedQuery` — parse-once/execute-many with named `$slot` bind slots; `prepare_query`, `substitute`; `Entity` and `Val(Ref)` fit each other's slots, bind mistakes are `API-022`/`023`/`024` (#181); a `:limit $n` or `:offset $o` slot takes `Val(Integer(n >= 0))` (`resolve_count_slot`, #306) - `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), `collect.rs` (`CollectSet`, #316: `[(collect-set ?e :a) ?s]` per row probes `[e :a ?v]` as an `IndexNestedLoopJoin` does and binds the values as one set, `#{}` when none, metered; `Expr::CollectSet`, parsed only as a whole binding clause), `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`, the set a `CollectSet` builds; 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), `collect.rs` (`CollectSet`, #316: `[(collect-set ?e :a) ?s]` per row probes `[e :a ?v]` as an `IndexNestedLoopJoin` does and binds the values as one set, `#{}` when none, metered; `Expr::CollectSet`, parsed only as a whole binding clause), `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`), `sort.rs` (`Sort`, #306: the `:find` columns as values, in the full order of `:order-by` — its keys, then every other `:find` column ascending, by `datalog::aggregate::row_order`/`value_order`; under a `:limit` a heap of the best `offset + limit` rows; reads its whole input, dropping the DICT lookups after each batch, metered; its rows turned back into cells only as returned), `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`, and after `Sort`), `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`, the set a `CollectSet` builds; 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 dd8c56277..578502efb 100644 --- a/README.md +++ b/README.md @@ -23,7 +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 +- ✅ **Bounded and ordered results** - `:limit N` returns at most N rows and stops the work once it has them; `:order-by [?score :desc] :limit 10` keeps only the top ten as it reads, and `:offset` pages through an ordered answer - ✅ **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 1e7540b07..460ee0e50 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` (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`). +- Native `:limit N`, `:order-by` and `:offset`, with a top-k `Sort` whose memory follows `N` (#306; done on `v3`, with the 10M-fact acceptance tests moved from #473; spec `docs/superpowers/specs/2026-10-24-limit-order-by-design.md`). Skipping the sort when an index already returns the rows in order is #641 (v3.next). - 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 40df051c5..ca5f2419d 100644 --- a/docs/TEST_COVERAGE.md +++ b/docs/TEST_COVERAGE.md @@ -10,7 +10,7 @@ **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. +#306 PR 2 (`:order-by`, `:offset`, `Sort`) added 18 tests and 7 more under `test-hooks`; the ignored acceptance test now also checks top-k. #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. @@ -360,11 +360,14 @@ 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) +- **`:limit`, `:offset` and `:order-by`** (#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`, `API-024`). - `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`). + - `tests/limit_order_test.rs` (7 more tests, `test-hooks`): 14 ordered shapes (each `:limit` shape with an `:order-by` over plain, aggregate and window elements, ascending and descending, with ties), in memory and over committed and pending facts, with no option, limits 0, 1, 3 and the answer size, each with offsets 1, the answer size less one and two past it, and `:offset 2` alone: through `execute` and a cursor pulled one row at a time, and the limited ones through a prepared `:offset $o :limit $n` and its cursor, with reordering on and off, each returns the test's own full sort of the unlimited answer, sliced. Pages of three cover the ordered answer; every value class in one column ascending and descending (an integer before an equal float, `-0.0` before `0.0`); a write transaction's staged facts; `:offset` slot errors; a full sort over `query_memory_limit` is `QRY-012`, top-k under the same limit is not. + - `src/query/engine/sort.rs` (7 unit tests): a full sort and top-k (k = 1 to past the input) equal a reference sort, ascending and descending, ties to the next column; `keep 0` pulls nothing; top-k's peak meter bytes are the same for 1,000 and 100,001 rows; `QRY-012` with and without `keep`; `rewind`. `src/query/engine/tests.rs` (1 test): top-k over 20,000 committed entities runs under a 512 KiB limit a full sort exceeds (the DICT lookups are dropped per batch). + - `src/query/datalog/aggregate/tests.rs` (2 tests, `row_order`), parser (4 unit tests: entries, directions, section orders, every `PRS-116`–`118`, `INT-012`/`013` and `PRS-115` case of `:offset`/`:order-by`), prepared `:offset` slots (1 unit test), `prs_116_…`–`prs_118_…` in `tests/error_codes_prs_test.rs`, the grammar corpus (4 valid, 3 syntax, 4 semantic files) and three verifier `SEMANTIC` witnesses (`PRS-116`–`118`); the verifier has 37 tokens and 16 templates. - **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. @@ -492,7 +495,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; the fixture is written with `LogWriter`. +- One `:limit` and top-k acceptance test (`tests/limit_acceptance_test.rs`, #306, the #473 `:limit` item), run by the nightly smoke workflow over one 10M-fact fixture written with `LogWriter`: `:limit 10` returns 10 rows in under 250 ms under a 64 KiB `query_memory_limit`, adding under 32 MiB of peak RSS; `:order-by [?v :desc] :limit 10` returns the ten largest values under a 1 MiB limit, adding under 32 MiB. - 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 e84e254994347540b7bbe1915f14e9d8c891649f Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sun, 25 Oct 2026 04:14:25 +0530 Subject: [PATCH 11/13] perf(size): Sort sorts with its heap, not a second sort in the binary (#306) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01TcpoiFyFeQQQELL872LfQF --- src/query/engine/sort.rs | 47 ++++++++++++++++++++++++++++------------ 1 file changed, 33 insertions(+), 14 deletions(-) diff --git a/src/query/engine/sort.rs b/src/query/engine/sort.rs index d2de22392..7c663147a 100644 --- a/src/query/engine/sort.rs +++ b/src/query/engine/sort.rs @@ -94,7 +94,7 @@ impl Sort { let old = std::mem::replace(top, row); self.held.uncharge(bytes(&old)); } - self.sift_down(0); + self.sift_down(0, self.rows.len()); } continue; } @@ -110,10 +110,11 @@ impl Sort { Ok(()) } - /// Whether held row `i` comes after held row `j` (both exist). - fn after(&self, i: usize, j: usize) -> bool { + /// Whether held row `i` comes after held row `j`, both among the first + /// `len`. + fn after(&self, i: usize, j: usize, len: usize) -> bool { match (self.rows.get(i), self.rows.get(j)) { - (Some(a), Some(b)) => self.cmp(a, b).is_gt(), + (Some(a), Some(b)) if i < len && j < len => self.cmp(a, b).is_gt(), _ => false, } } @@ -122,7 +123,7 @@ impl Sort { fn sift_up(&mut self, mut i: usize) { while i > 0 { let parent = (i - 1) / 2; - if !self.after(i, parent) { + if !self.after(i, parent, self.rows.len()) { break; } self.rows.swap(i, parent); @@ -130,12 +131,12 @@ impl Sort { } } - /// Restore the heap (worst on top) below `i`. - fn sift_down(&mut self, mut i: usize) { + /// Restore the heap (worst on top) of the first `len` rows below `i`. + fn sift_down(&mut self, mut i: usize, len: usize) { loop { let mut worst = i; for child in [2 * i + 1, 2 * i + 2] { - if self.after(child, worst) { + if self.after(child, worst, len) { worst = child; } } @@ -148,6 +149,28 @@ impl Sort { } } +impl Sort { + /// Sort the rows held, last first (to pop). A heapsort over the heap + /// top-k already keeps, rather than a second sort in the binary: rows + /// equal under the full order are equal in every column, so an unstable + /// sort gives the one order. + fn heap_sort(&mut self) { + let n = self.rows.len(); + if self.keep.is_none() { + for i in (0..n / 2).rev() { + self.sift_down(i, n); + } + } + // Move the worst row held to the end of the heap each time: best + // first. + for end in (1..n).rev() { + self.rows.swap(0, end); + self.sift_down(0, end); + } + self.rows.reverse(); + } +} + impl Operator for Sort { fn next_batch(&mut self, ctx: &mut Ctx<'_>) -> Result> { if self.out.is_none() { @@ -163,12 +186,8 @@ impl Operator for Sort { ctx.ids.end_pull(); } } - let mut rows = std::mem::take(&mut self.rows); - let order = &self.order; - // Rows equal under the full order are equal in every column, so - // an unstable sort gives the one order; last first, to pop. - rows.sort_unstable_by(|a, b| row_order(b, a, order)); - self.out = Some(rows); + self.heap_sort(); + self.out = Some(std::mem::take(&mut self.rows)); } let Some(rows) = self.out.as_mut().filter(|r| !r.is_empty()) else { return Ok(None); From f6a5066a941d5a2846675be9ae829cabbe51b897 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sun, 25 Oct 2026 09:33:16 +0530 Subject: [PATCH 12/13] ci(size): 1,230 KiB gate for #306; docs: counts, top_k numbers Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01TcpoiFyFeQQQELL872LfQF --- .github/workflows/binary-size.yml | 2 +- CLAUDE.md | 4 ++-- PHILOSOPHY.md | 4 +++- docs/BENCHMARKS.md | 2 ++ docs/TEST_COVERAGE.md | 6 +++--- 5 files changed, 11 insertions(+), 7 deletions(-) diff --git a/.github/workflows/binary-size.yml b/.github/workflows/binary-size.yml index 699f4fa70..6ab5bb25c 100644 --- a/.github/workflows/binary-size.yml +++ b/.github/workflows/binary-size.yml @@ -14,7 +14,7 @@ jobs: name: Check release binary size runs-on: ubuntu-latest env: - SIZE_LIMIT_BYTES: 1228800 # 1200 KiB — raised from 1 MiB for #277's structured error codes; see PHILOSOPHY.md §4 + SIZE_LIMIT_BYTES: 1259520 # 1230 KiB — raised from 1 MiB for #277's structured error codes, from 1200 KiB for #306's :order-by/top-k; see PHILOSOPHY.md §4 steps: - name: Checkout code uses: actions/checkout@v4 diff --git a/CLAUDE.md b/CLAUDE.md index 9a18a0c94..ea0a0ab6d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -16,7 +16,7 @@ Minigraf follows the "SQLite for bi-temporal graph databases" philosophy: - **Zero-configuration** - No setup, no config files, just works - **Embedded-first** - Library not server, in-process execution - **Single-file database** - One portable `.graph` file -- **Self-contained** - Minimal dependencies, small binary (~1.2MB budget as of #277; see PHILOSOPHY.md §4) +- **Self-contained** - Minimal dependencies, small binary (~1.2MB budget as of #277, 1,230 KiB enforced on v3 since #306; see PHILOSOPHY.md §4) - **Cross-platform** - Native, WASM, mobile, embedded - **Reliability over features** - Do less, do it perfectly - **Bi-temporal first-class** - Time travel is a core feature, not addon @@ -215,7 +215,7 @@ Auto-migrates v7 → v8 on open (spec §9, with a backup meta page). v1–v6 are ## Test Coverage -**1810 tests** (1796 passing, 14 ignored; unit + integration + doc). Engine answers once checked by the executor are frozen in `tests/answers/` (#551). +**1828 tests** (1814 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/PHILOSOPHY.md b/PHILOSOPHY.md index f847bf099..6cfd02e2c 100644 --- a/PHILOSOPHY.md +++ b/PHILOSOPHY.md @@ -113,7 +113,9 @@ SQLite's success comes from a clear philosophy: be a library, not a server. Be s #277's structured error codes — 130 documented codes across 6 categories cannot fit in 1MB even with opt-level="z", LTO, codegen-units=1, and strip=symbols all already applied; see CI's `binary-size.yml` for the - enforced limit) + enforced limit). v3.0.0 raises the enforced limit to 1,230 KiB for + `:order-by`/`:offset` and the top-k `Sort` (#306), whose ~11 KiB did not + fit the 2 KiB left after the streaming engine (#473) and set values (#316) - No runtime dependencies (no JVM, no Python, no Node.js) **Anti-pattern**: Requiring external services, libraries, or runtimes to function. diff --git a/docs/BENCHMARKS.md b/docs/BENCHMARKS.md index 5bb4aac1e..dbb7be25f 100644 --- a/docs/BENCHMARKS.md +++ b/docs/BENCHMARKS.md @@ -194,6 +194,8 @@ Measuring this found that `Aggregate` built its sorted map of groups uncharged w `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. +`top_k` is `attribute_scan` with `:order-by [?a :desc] :limit 10`: `Sort` reads the whole attribute but holds ten rows, so its time follows the scan and its memory does not (p50 22.4 ms, 241 ms and 2.54 s at 100K, 1M and 10M against `attribute_scan`'s 23.1 ms, 250 ms and 2.63 s on the same run; 6–7 MiB peak RSS at every scale). Same server, one NUMA node, 2026-10-25. The acceptance test runs `:order-by [?v :desc] :limit 10` over the same 10M facts in 28.5 s (bench profile, opt-level `z`) under a 1 MiB `query_memory_limit`, adding no measurable RSS. Skipping the read when an index returns the rows in order is #641. + ### 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 "