diff --git a/CHANGELOG.md b/CHANGELOG.md index 9181188a5..a03418f50 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **A per-fact pseudo-attribute reads the nearest pattern before it with the same entity** (#594). `:db/valid-from`, `:db/valid-to`, `:db/tx-count` and `:db/tx-id` read the metadata of one fact: the one matched by their *source*, chosen when the query, rule or prepared query is parsed, as the nearest pattern before them in the `:where` text with the same entity term and a stored attribute (a keyword, or a variable as in `[?e ?a _]`). The search starts in the pseudo-attribute's own clause list and moves outwards, before each enclosing `not` / `or`; it crosses a `not-join` / `or-join` only for an entity variable in its join variables, and stops at a rule body, which never reads its caller's patterns. Terms are the same when they are one variable, one `$slot`, or constants naming one entity (`:alice` and its UUID). Before, they read the last pattern *the planner ran* that matched the entity, so the answer depended on the plan and differed under the `wasm` feature: in `[?e :person/name ?n] [?e :person/age ?a] [?e :db/tx-count ?t]`, `?t` could be the tx-count of either fact, and is now the `:person/age` fact's. Three further changes follow: (1) with no source, the query fails at parse time with the new `PRS-081` instead of matching every visible fact of the entity (`[?e :db/tx-count ?t]` before `[?e :person/name ?n]`, a `_` entity, a pseudo-attribute in a rule body whose only candidate is in the caller); to read every fact of an entity, write `[?e ?a _]` before it; (2) inside `not`, `not-join`, `or` and `or-join`, a pseudo-attribute reads the outer fact, so `[?e :person/name ?n] (not [?e :db/tx-count 1])` excludes the entities whose *name* fact is from tx 1, where before the body matched any fact of the entity; (3) rows that differ only in a source fact's metadata are no longer merged by `or`. `:db/valid-at` is unchanged. The pseudo-attribute pattern now binds or checks a value; it never scans facts, so the planner places it after its source rather than by selectivity. - **A pseudo-attribute pattern never gives a variable its value** (#606). A variable bound by several patterns returns the value of the first real pattern in the `:where` text that binds it (#545); a `:db/valid-at` pattern that the planner ran first used to bind its entity variable itself, as a ref, so `[:find ?b ?t :where [?b :db/valid-at ?t] [?a :friend ?b] [?b :name "Bob"]]` returned the ref of `:bob` with source order and the keyword `:bob` stored by `:friend` natively. It returns the keyword now, whatever runs first. Only the keyword-or-ref kind changes; rows and counts are the same. - **Every build reorders query clauses; the `wasm` feature does nothing** (#594, #596). The `wasm` Cargo feature kept patterns, `not` clauses and `or` branches in source order and turned off the `or` short-circuits (#250), for small browser datasets and a predictable order. With every answer independent of the plan (#594, #596, #606) it only made queries slower, and no first-party build enabled it (the browser package uses `browser`). The feature is kept, so builds that name it still resolve, but is deprecated and has no effect. +- **Aggregate and window answers no longer depend on the order of their input** (#547). Before, aggregates and window functions read their rows in the order the plan produced them, and several answers followed that order. Six cases changed. (1) `sum` over floats is exact: the values are added without rounding and the total is rounded once, so `1e16 + 1.0 + -1e16 + 1.0 + 0.5` gives `2.5` in every order (before, `1.5` or `2.5`). Window `sum` and `avg` likewise. (2) `sum` over integers saturates once, at the end: `i64::MAX + 1 + -1` gives `i64::MAX` (before, `i64::MAX - 1` or `i64::MAX` by order). (3) Window `sum` over integers adds them exactly; before they were added as floats, so a running total above 2^53 lost its low digits. (4) `min` / `max` of floats follow one order: `-0.0` before `0.0`, NaN after every other number (before, the first of two such values seen won). Window `min` / `max` use the same order over any values, so the minimum of keywords is the first in text order. (5) Window functions sort each partition by `:order-by` under one total order of values (numbers by numeric value, an integer before an equal float, then strings, booleans, refs, keywords, `nil`), so keywords, refs and booleans now sort (before, any two of them compared equal and kept their input order). Rows with equal `:order-by` values are peers: `rank` gives them the same rank, and `sum`, `count`, `min`, `max`, `avg` and registered window functions give every peer the value at the last peer, as SQL does; `row-number` numbers peers in the order of the other variables `:find` names. For example, with `?s` of 1.5, 5, 10, 10, `(sum ?s :over (:order-by ?s))` gave 1.5, 6.5, 16.5, 26.5, the 16.5 going to either tied row, and now gives 1.5, 6.5, 26.5, 26.5; ranking `:x :x :y :y` by `:order-by` gave ranks depending on the order the scan met them (1 2 3 4 when interleaved) and now gives 1 1 3 3. (6) In a query with aggregates, a window's argument, `:partition-by` and `:order-by` must be plain `:find` variables, or the query fails at parse time with the new `PRS-082`; before, any other variable read `nil` on every group row. The error a failing aggregate query reports is the first error, in `:find` order, of the first failing group in output order. A registered aggregate still receives its values in input order, which can depend on the plan; one whose result depends on that order may answer equivalent queries differently. ### Added @@ -96,6 +97,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Answers that depend on the plan fail the tests (#596).** A test-only switch (`minigraf::__reorder`, `test-oracle` feature) turns off every place the native build reorders: the pattern selectivity sort, the `not`/`not-join` and `or`/`or-join` cost sorts and the `or` short-circuits (#250), and the engine's `HashJoin` build side. `tests/plan_independence_test.rs` runs a matrix of queries (joins, expressions, rules, `or`/`or-join`, `not`/`not-join`, keyword and ref, floats, long strings, aggregates, time travel), in memory and on file, through `execute` and a cursor, with the switch on and off, and requires the same rows. Under `test-oracle` the model-based (#385) and property (#386) tests also run every generated query with the switch off. - **Negation bodies in both forms (#546).** A second test-only switch (`minigraf::__subplans`, `test-oracle` feature) forces the engine's `not` / `not-join` bodies to run hashed or correlated. `tests/plan_independence_test.rs` runs every query with the body forms chosen, forced hashed and forced correlated, each with reordering on and off; its matrix gains negations over floats near zero, `0.0` and `-0.0`, keywords and refs of one entity, long strings, bodies reading an imported variable they do not bind, variable-free bodies and `:db/valid-at`. `tests/negation_equality_test.rs` checks the changed answers on every executor path, in memory and after a checkpoint. The property test's rerun also forces hashed bodies. Engine unit tests cover the form the planner picks, routing, multiplicity, failing expressions in bodies, and that a hashed body is read once and a correlated one probes once per row. - **Disjunctions (#546).** `tests/or_equality_test.rs` checks the changed `or` / `or-join` answers through `execute` and a cursor, in memory and after a checkpoint. `tests/plan_independence_test.rs` gains disjunctions (keywords meeting refs, floats near zero, long strings as keys, branches naming different outer variables or no join variable, a branch with `not`, an `or` with nothing before it, a cross product, `:db/valid-at` in a branch), run with branches chosen, forced hashed and forced correlated. The property test generates `or` and `or-join` over bound variables, with a reference evaluator that deduplicates the rows they keep. Engine unit tests run every disjunction shape with each branch form forced against the executor, and cover routing, the form the planner picks, `Distinct`, and that a hashed branch is read once. +- **Aggregates and windows (#547).** `src/query/datalog/aggregate.rs` unit tests check every accumulator over every order of small inputs (exact float sums against hand-computed values, cancellation, subnormals, signed zeros, infinities and NaN, integer saturation, errors decided at the end) and that the value order is total. `tests/aggregate_window_rules_test.rs` checks the changed answers; `tests/plan_independence_test.rs` gains aggregate, window and mixed queries over ties, keywords and refs, and floats. - `tests/snapshot_atomicity_test.rs` races a writer against a reader: a query must see each transaction whole, whether written with `execute` or a `WriteTransaction` (#554), and must read one state for all its patterns (#555), in memory and on file. It also runs the v2.x checkpoint race from #558, which never failed on v3: copy-on-write checkpoints do not overwrite pages a published reader uses. - **The `datalog_eval` fuzz guard counts partial bindings** (#597). The guard skipped inputs whose patterns join to more than 50,000 rows, counting only the rows after the last pattern. A `not-join` body still runs on the executor pattern by pattern in text order, so a 120-pattern body built hundreds of thousands of partial bindings before its last pattern matched nothing, and ran past the 60-second timeout. The guard now counts the bindings after every pattern. On the `v3` corpus (4,783 inputs) coverage drops by 7 of 9,719 edges. The input is added as a seed, as is the `btree_page` input of the cursor fix above. diff --git a/CLAUDE.md b/CLAUDE.md index 858570368..88dd8c4b1 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -131,6 +131,7 @@ cargo run < demos/demo_negation.txt - `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) - `executor.rs`: Query executor — temporal filter (tx-time → net-assert → valid-time), not/not-join post-filter - `matcher.rs`: Pattern matching with variable unification; `edn_to_value`, `edn_to_entity_id` + - `aggregate.rs`: `Agg` (one aggregate of one group) and `Running` (a window's running value), shared by the executor and 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) (#547) - `evaluator.rs`: `RecursiveEvaluator` (semi-naive), `StratifiedEvaluator` (repeats a stratum whose rules recurse through a `not`/`or` rule until it derives nothing new, #633), `evaluate_not_join` - `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 @@ -204,7 +205,7 @@ Auto-migrates v7 → v8 on open (spec §9, with a backup meta page). v1–v6 are ## Test Coverage -**1606 tests** (1596 passing, 10 ignored; unit + integration + doc). +**1616 tests** (1606 passing, 10 ignored; unit + integration + doc). 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/README.md b/README.md index df8fb8f78..e56ed5798 100644 --- a/README.md +++ b/README.md @@ -131,7 +131,7 @@ let r2 = pq.execute(&[("tx", BindValue::TxCount(2)), ("entity", BindValue::Entit ```bash cargo run # interactive Datalog REPL -cargo test # run 1606 tests +cargo test # run 1616 tests cargo run < demos/demo_recursive.txt # recursive rules demo ``` diff --git a/docs/ERROR_REFERENCE.md b/docs/ERROR_REFERENCE.md index dc0c55b51..bef8069a3 100644 --- a/docs/ERROR_REFERENCE.md +++ b/docs/ERROR_REFERENCE.md @@ -102,6 +102,7 @@ with no `CodedError` anywhere in its chain. | PRS-079 | Rule rejects unexpected trailing argument(s) | Parser | | PRS-080 | :find variable not bound in :where | Parser | | PRS-081 | Pseudo-attribute has no source pattern | Parser | +| PRS-082 | Window reads a non-grouping variable | Parser | | QRY-001 | Invalid entity | Query Execution | | QRY-002 | Attribute must be a keyword | Query Execution | | QRY-003 | Cannot transact a pseudo-attribute | Query Execution | @@ -1596,6 +1597,22 @@ $<4097-char-slot-name> ``` *Move `[?e :person/name "Alice"]` before `[?e :db/tx-count ?t]`* +### PRS-082 Window reads a non-grouping variable + +**Error text**: `{} in {} must be a plain :find variable when the query has aggregates` + +**Cause**: In a query whose `:find` holds an aggregate, window functions run over the groups the aggregation produces, after it. Each group row holds only the plain `:find` variables and the aggregate results, so a window's argument, `:partition-by` and `:order-by` must each be a plain `:find` variable. `:with` variables do not count: they do not group when `:find` holds a window. Before v3.0.0 such a variable read `nil` on every group row, so the window ordered or summed nothing. + +**Resolution**: +- Add the variable to `:find` as a plain variable, so it groups. +- Or run the window in a query without aggregates. + +**Example**: +```datalog +(query [:find ?dept (sum ?salary) (rank :over (:order-by ?salary)) :where [?e :dept ?dept] [?e :salary ?salary]]) +``` +*`?salary` is aggregated, not a group: order by `?dept`, or drop `(sum ?salary)`* + --- ## QRY — Query Execution Errors diff --git a/docs/TEST_COVERAGE.md b/docs/TEST_COVERAGE.md index 41a83578c..ea48025f0 100644 --- a/docs/TEST_COVERAGE.md +++ b/docs/TEST_COVERAGE.md @@ -2,9 +2,9 @@ ## Coverage Summary -**Verified**: 2026-10-19 with `cargo test --quiet` +**Verified**: 2026-10-20 with `cargo test --quiet` -**Result**: 1,596 passing tests and 10 ignored tests (1,606 total) +**Result**: 1,606 passing tests and 10 ignored tests (1,616 total) This covers the unreleased file-format-v8 work (#287, #371): the index-key value-collision fix, the v7→v8 migration, and the drop of formats v1–v6. It otherwise still covers the public database API, the Datalog engine, storage and recovery paths, v2.0.0's locking and structured-error-code changes, and v2.0.1's regression test for rebuilding indexes on open after several checkpoints (#370). Merged from the v2.x line: `tests/point_query_history_test.rs` and new storage unit tests check that attribute-narrowed bound-entity lookups return exactly what a full scan returns, and that `net_asserted_facts` matches its previous implementation on randomized histories (#323). Attribute-scan tests check that the AEVT range covers exactly the queried attribute, including non-ASCII names such as `:丿` and prefix siblings such as `:ab` for `:a` (#381). Incremental B+tree rebuild tests check that checkpoints which copy untouched index leaves produce exactly the entries and file checksum a full rebuild would, including over repeated checkpoints, that single-fact checkpoints do not fragment the index, and that a damaged slot directory is rejected rather than copied (#315). B+tree cursor tests check that every scan walks the tree, not the leaf `next_leaf` chain: full scans and start positions match the input, `seek` from the current position lands where a fresh descent from the root would (randomized, plus within a leaf, to a sibling, across subtrees, past the end, into the gap after a leaf, and backwards as a no-op), a seek to the sibling leaf reads exactly one page, reads are unchanged with every `next_leaf` zeroed, a child pointer into a non-B+tree page is STG-013, and a cycle in child pointers is an error, not a hang (v8 storage format PR 1, #374). Directory-sync unit tests check that creating the `.graph` file or the WAL, and deleting the WAL, fsync the parent directory after the file operation, and that reopening an existing file does not (#389). Disjunction tests check that `or-join` returns an empty result, not INT-031, when no rows reach it: on an empty database, after a pattern that matches nothing, after a `not` that removes every row, and in a rule body (#405). A stratification unit test builds the same two-rule negative cycle 64 times and checks that the INT-054 message names the same predicate pair every time (#410). Magic-sets tests check that a recursive rule with a literal start that recurses through another rule returns the same results as without the literal (#297). API tests check that `execute()` and `WriteTransaction::execute()` reject `$slot` bind tokens in every position with API-010 (#407). The REPL `run()` test now runs in a child process with null stdin, so it no longer hangs when the test runner's stdin is an open non-TTY stream (#415). v8 page-format tests (PR 2, #374, #388) check the 24-byte page header (a flipped bit, a wrong page id and a future generation each give their own STG code, legacy page types fail before the CRC, and a failed page is never cached), meta-page encoding, the free-list chain, WAL v2's base generation, and every row of the meta-selection table: both slots valid, a torn newest slot with the WAL's base generation (opens at the older one), a WAL newer than the valid meta, pages of the next generation on the free list or past `page_count` with no WAL (all STG-033, with the file unmodified), an unknown feature bit (STG-034), and a pre-release v8 single header (STG-032). Fault-injection tests crash `save()` at every page write (cleanly and with torn pages, including the meta write) and every sync, and crash a v7 migration at every write; each reopen finds exactly the old or the new checkpoint, and every reachable page plus the free list covers the file exactly. Twenty-five consecutive saves never write a page the previous meta references. Covering-key tests (PR 3, #433) check that byte order equals logical order for every key component and value type on random tuples (negative, zero and large integers, NaN and ±0.0, strings with 0x00, 64- and 65-byte strings, FOREVER, newest-transaction-first) and that every component round-trips; that prefix-compressed leaves and shortest-separator internal nodes round-trip, agree with a linear scan, keep `max(left) < s ≤ min(right)`, and reject an overflow; that the byte-key B+tree's builds, incremental rebuilds, cursor and `seek` match their expected sets; that dictionary ids are assigned once, never reused, shared between a keyword attribute and value, and deterministic; that a re-asserted or retracted long value writes no new value page and a forced hash collision still finds the right value; that one transaction with two timestamps is STG-037 and a missing dictionary entry STG-036; that committed reads by entity, entity and attribute, attribute, and in full equal an in-memory model across eight checkpoints and a reopen; that short-valued reads touch no value page; that a v7 file with every value type, long strings, refs and retractions migrates to exactly the same facts; that a pre-covering-key v8 file is STG-032; and that string values over 4,068 bytes and attributes or keywords over 1,024 bytes are WAL-003. A WAL test checks that reopening next to a fully checkpointed WAL does not reuse transaction numbers (#447). Copy-on-write checkpoint tests (PR 4, #434) check that lazy free-list pops and pushes conserve every page across page boundaries, partial head pages and chain exhaustion, never hand out a page freed in the same generation, and share unread chain pages unchanged; that a copy-on-write insert equals a sorted merge for random batches (including keys below the first and above the last, many into one leaf, root growth to depth 3, and 1 KB separators that split internal nodes), leaves the old tree readable and unchanged, frees exactly the replaced nodes, writes only new pages, writes exactly one leaf and two internal nodes for one entry in a depth-3 tree, and does not fragment leaves under single inserts; that a checkpoint after 1 or 100 right-edge facts writes at most the tree-depth difference more pages at 100,000 facts than at 10,000, and at most one path per new entry per tree for random keys; and that four reader threads querying throughout 40 checkpoints never fail and never see the fact count go backwards. Hardening tests (PR 5) damage an index leaf, an internal node, a dictionary leaf, a value page and a free-list page in turn. Every committed read or checkpoint that reaches the damaged page fails with STG-029, never with wrong or fewer rows. A query through `Minigraf` on a file with damaged leaves fails with STG-029, and the file still opens at its last checkpoint. A size test measures at most 200 bytes per fact (142 measured) on 144k facts in #433's shape; its 1.15M-fact variant is ignored locally and runs in the nightly smoke workflow. A shared dictionary-cache test checks that cached ident and timestamp lookups return the same facts and that the cache stays bounded. Merged from the v2.x line: storage and `edge_cases_test.rs` tests check that more than 65,535 uncheckpointed facts each resolve to their own fact through entity lookups and attribute scans, in memory and file-backed (#445). Merged from the v2.x line: `tests/golden_corpus_test.rs` opens a copy of each committed golden file in `tests/golden/` (v7 files written by minigraf 2.0.3: one checkpoint, several checkpoints, indexes rebuilt on open, a crash-left WAL, an already-checkpointed WAL, same-transaction multi-values) and checks its manifest's CRC32, `tx_count` and query results after open, after a checkpoint and reopen, and after one more write; a second test checks that every golden file has a manifest (#391). On `v3` the same test checks that each v7 file has a v8 meta page after its first open and gives the same results after reopening the migrated file, and adds seven v8 golden files (one checkpoint, seven copy-on-write checkpoints, a crash-left v2 WAL, an already-checkpointed v2 WAL, same-transaction multi-values, a v7 file migrated in place, and long values at the size limits); the #371 queries, skipped on v2.x, run and pass. `tests/model_based_test.rs` runs random sequences of transacts, retracts, write transactions, checkpoints, reopens and simulated crashes against a file and, after every step, compares the transaction counter, `verify`, full rows with windows and tx counts, and six query shapes under `:as-of N` and `:valid-at T` with a reference model (#385). `tests/property_test.rs` writes random histories (batched transacts, retracts, write transactions, checkpoints) in memory and to a file, and compares random queries (one to three joined patterns, `not` or `not-join`, a comparison predicate, `:as-of`, `:valid-at` or `:any-valid-time`) with a reference evaluator, on the open file and after a reopen; a query error fails the test (#386). `tests/crash_kill_test.rs` kills a child process running a seeded workload (multi-valued batches, batched retracts, five checkpoint modes) at a random point and checks that the reopened file holds exactly the transactions the child logged as committed, or those plus the one in flight, through full-scan, AEVT, AVET, EAVT and `:as-of` queries and `verify`, again after a second reopen and after a checkpoint plus reopen (#384). Integrity tests (#373) check that intact files verify with no problems and an exact fact and page count, and that `verify` reports each kind of damage with its code: an EAVT, AEVT, AVET or VAET leaf missing an entry (STG-038), keys out of order and a page reached from two trees (STG-039), a rotted index leaf (STG-029), a free page still in use and a leaked page (STG-035), a missing transaction entry (STG-036), and an entity id whose UUID maps to another id (STG-040). After each index or free-list damage, `rebuild_indexes` leaves a file that verifies clean, reads back every fact and accounts for every page. The source is an index that agrees with another one, or the only intact one; three intact indexes that all disagree, or a damaged dictionary, give STG-041 with the file unchanged. A rebuild commits pending facts, including new ids and long values. A crash at every write (clean and torn) and every sync of a rebuild reopens either at the old generation with only its old damage or at the new one with none, never losing a fact. Through `Minigraf` (`tests/integrity_test.rs`), a rebuild commits WAL entries and deletes the WAL, and flipping a byte in each leaf of a file in turn is always reported, then either repaired (index leaves) or refused with STG-041 and the file unchanged (dictionary leaves). Page-count tests (#497) check that a CRC-valid meta whose `page_count` is past the end of the file (one page past, 2^50, `u64::MAX`; with the other slot valid or damaged; with or without a WAL) and a file cut short by one page, three pages or down to three pages are `STG-044` at open, with the file unchanged and not grown, while a `page_count` exactly at the end opens; that a checkpoint whose pages did not all reach the backend fails with `INT-049` before writing its meta, and the file reopens at the previous generation; and that `verify` at a meta counting pages past the end reports `STG-044` without scanning to that count. Through `Minigraf` (`tests/integrity_test.rs`), a file truncated by one or three pages and a file whose meta was rewritten with `page_count` 2^50 and a valid CRC are `STG-044` read-write and read-only, with no WAL created and the file length unchanged. The `file_header` fuzz input from #497 is a corpus seed. A sub-page test (#506, `tests/integrity_test.rs`) checks that a file of 1 to 4,095 bytes is `STG-045` read-write and read-only, unchanged and with no WAL, when it is a database cut below one page or foreign bytes, and opens as a new database that takes writes when it is a prefix of the first meta page or all zeros; a file-backend unit test checks that `partial_page` returns the bytes past the last whole page. The `corrupt_file` fuzzer no longer skips files under one page: they must fail to open or open empty, and the #506 input is a corpus seed. Damaged v7 fact page tests (#496) check that a page in the counted fact range whose type byte is anything but the fact-page type (zero, `0x01`, the v2.x B+tree types, `0xFF`, a zeroed page; first and last page) is `STG-014` rather than skipped, while a range derived from the first index root skips only v2.x B+tree pages; that opening such a v7 file read-write or read-only fails with `STG-014` and writes nothing; and, in `tests/migration_matrix_test.rs`, that the golden `v7_basic.graph` with page 1's type byte changed or page 1 overwritten with noise fails both opens with the file unchanged and no WAL created. Both inputs are `fact_page` corpus seeds. Merged from the v2.x line: `tests/edge_cases_test.rs` checks the same refusal and acceptance through `Minigraf::open` on a database checkpointed twice (#506). Torn-WAL-tail tests (#513) check that opening a WAL writer cuts off a partial last entry (cut in the checksum, the header or the payload) or an entry with a bad checksum and everything after it, then appends where replay reaches, and leaves an intact WAL byte for byte; that an append that fails part-way makes every later append `WAL-007` while a rejected oversize fact (`WAL-003`, nothing written) does not; through `Minigraf`, that the failed transaction is not applied, later `transact` and `commit` are `WAL-007`, and a `checkpoint()` makes writes work again; and, in `tests/wal_test.rs`, that a commit made after reopening a crash-left WAL with each kind of torn tail survives a second crash. The fault matrix (#390, `src/fault_matrix.rs`) drives a test-only fault plan through the real file backend, WAL and directory fsync: an EIO, a short or torn write (1, 40 and 4,095 bytes), a full disk that keeps failing every later write, a failed fsync, and a failed fsync that also loses the writes since the last good one ("fsyncgate"), each at every write or sync call of a transact, a transact that creates the WAL, an automatic checkpoint, `checkpoint()`, the close-time checkpoint and `rebuild_indexes()`, on a database committed over two checkpoints with multi-valued, retracted and long values and entries pending in the WAL. After each fault, the handle holds the old facts or those plus the whole transaction, never part of one; after a failed checkpoint another is `STG-046`; a later write succeeds or is `WAL-007`/`STG-046`; a kill and reopen gives exactly the handle's facts, plus the failed transaction whole or not at all, and `verify` finds nothing; and a commit after recovery survives another kill. Two more tests fill the disk halfway through a checkpoint (`STG-046`, reads unchanged, reopen intact), and fail the fsync after a checkpoint's meta write and then retry the checkpoint, crashing at each of its writes: the retry is refused and every reopen is intact (without the refusal, a reopen hits `STG-029`). Unit tests check the fault plan's counting, and that a `rebuild_indexes` refused with `STG-041` leaves the handle usable. Live-read tests (#379) check that `triple_len` ends exactly where `tx↓` begins for every value type and index; that on random histories (several valid-time windows, retractions, re-assertions, several values per attribute, long values, 12 transactions per checkpoint) the committed live read for every scan kind and `:as-of` point is exactly the net-asserted set, that it stays equivalent with pending facts on top, and after reopen; and that a point read on a triple asserted and retracted 2,000 times reads under a third of the pages a history read does, with or without `:as-of`. `tests/live_reads_test.rs` runs the same history in a checkpointed file database and in memory and checks that point, attribute, full-scan, join, valid-time, `:as-of` and rule queries return the same rows, before and after reopen. One-window tests (#435) check that `net_asserted_facts` keeps each triple's newest transaction only (a later window replaces an earlier one, closing an open window, a retraction in the newest transaction hides it, records with two windows in one transaction keep both, once each) against a per-triple model on random histories; that the on-disk key walk agrees at every `:as-of` before and after reopen; and that `API-011` rejects two windows of one fact in one transaction without taking a `tx_count`. `tests/valid_time_window_test.rs` checks replacing, closing, extending, reopening and retract-then-reassert through entity-bound and attribute queries in memory, on a checkpointed file and after reopen, `API-011` from `transact` and from `WriteTransaction::execute`, and a later statement's window committed when two statements of a `WriteTransaction` give one fact different windows. Window-validation tests (#436) check that `check_valid_windows` rejects an empty window, an inverted one, and a `:valid-to` at or before the transaction time when `:valid-from` is left out, but not retractions; that `execute` rejects each such `transact` (both bounds, equal bounds, `:valid-to` only, a per-fact bound inverting the transaction's window) with `API-019` without writing or taking a `tx_count`, in memory, checkpointed and after reopen; that `tx.execute()` rejects them and the transaction still commits its other statements; that `commit()` rejects a default `valid_from` that has passed the `:valid-to` by commit time; and that `FactStorage::transact_batch` takes no `tx_count` on rejection. Statement-order tests (#477, `tests/statement_order_test.rs`) run every mix of two statements on one fact inside a `WriteTransaction` (retract then assert, assert then retract, two retracts, two windows, after a committed assertion and on an empty database) and check that reads inside the transaction and the committed rows agree, in memory, checkpointed and after reopen; that other facts of an overridden statement survive; that the commit is one transaction without the overridden records; and that two windows in one statement are `API-011` from `tx.execute()` while the rest commits. A unit test checks the commit collapse keeps each fact's newest statement in order. A log-writer test checks that an inverted window is `API-019`, a second window of a fact in one transaction `API-011`, and an assertion and a retraction of one fact in one transaction `API-020` in either order, each leaving the writer usable and the output exact, while duplicate retractions, identical repeats and the same fact in the next transaction are accepted. Cursor tests (#432, `tests/cursor_test.rs`) check that `query()` batches partition the `execute()` answer for several batch sizes (never empty, never over `max_rows`, `max_rows == 0` still progresses, `None` stays `None`), that the iterator and aggregates, rules, `:as-of` and `:any-valid-time` match `execute()`, that a cursor opened before a `transact`, a `WriteTransaction` commit, a retract or a checkpoint returns the pre-write answer to the end (in memory and file-backed), that `transact`/`retract`/`rule` are `API-012` and bind slots `API-010`, that a `WriteTransaction` thread gets `INT-001`, that rule limits give `INT-020` as from `execute()`, that `PreparedQuery::query` matches `PreparedQuery::execute`, and that a cursor is `Send` and outlives its database handle. Fact-log tests (#430, `tests/fact_log_test.rs`) check that `fact_log()` returns exactly the records written (assertions and retractions, with `tx_count` and `tx_id`) in memory, with everything pending, half checkpointed, fully checkpointed and after reopen, in both orders, with windows of 1 to 1,000 records and batch sizes of 1 to 1,000, in non-decreasing `tx_count` in Tx order; that every layout gives identical records, including valid-time windows and long string values; that a transaction larger than the window is split with no loss or duplicate; that attribute, attribute-prefix, entity and tx-range filters and their combinations match a model, and unknown names match nothing; that writes after open are not in the log; that `checkpoint()` and `rebuild_indexes()` are `API-013` while a log is open (a checkpoint with nothing to write is allowed), that automatic checkpoints from `execute()` and `WriteTransaction::commit` wait for the log, and that closing, dropping or finishing the log releases the database; that a `WriteTransaction` thread gets `INT-001`; and that a log is `Send`, outlives its database handle, and the deferred writes replay on the next open. Read-only open tests (#429, `tests/read_only_open_test.rs`) check that a read-only handle on a file with checkpointed facts and a pending WAL returns the same query results, `tx_count` and fact log as a read-write open; that queries, a cursor, a fact log, `verify`, `prepare`, rules and a UDF leave the `.graph` and WAL bytes and the `.graph` mtime unchanged, and no WAL is created; that `transact`, `retract`, `begin_write`, `checkpoint` and `rebuild_indexes` are `API-014` without taking a `tx_count`, also on a read-only in-memory database and through the builder path; that a missing file is `STG-042` and nothing is created; that an empty file is an empty database and stays 0 bytes; that read-only handles share a file while a read-write one is refused with `STG-025`, both ways; that a file without write permission opens read-only; and that small and whole-file page caches work. File-backend unit tests check that a shared-mode backend refuses page writes and syncs with `INT-056`, that the in-process registry counts readers and admits readers or one writer (also without kernel locks), and that a reader in another process shares the file with a reader here and refuses a writer here with `STG-026`. Storage unit tests check that a read-only open of a v7 file at every point of an interrupted migration (clean and torn writes) sees every fact and writes nothing, and that an empty backend stays empty. `tests/golden_corpus_test.rs` opens a copy of every golden file read-only first and checks its `tx_count` and queries, and that neither the file nor its WAL changes. Log-writer tests (#431, `tests/log_writer_test.rs`) copy a source's fact log (assertions, retractions, an explicit valid-time window, long strings, refs, keywords, checkpointed and pending facts) through `LogWriter` and check that the new file has no `.partial` or WAL, verifies clean, and gives the source's `tx_count`, fact log and query results (current, `:valid-at`, `:any-valid-time`, every `:as-of`, and entity-, attribute- and value-bound lookups) read-write, read-only and after reopen (#370); that it takes new writes at the source's next `tx_count`; that 90,000 facts over two batch commits verify clean and count correctly at `:as-of`; that a purged middle transaction is a hole whose `:as-of` equals the one before, and trailing holes keep the counter through `advance_tx_count`; that an empty log is an empty database with the counter; that `tx_count` 0, a lower or closed transaction and a backwards advance are `API-015`, a second `tx_id` `API-016`, a second window `API-011`, a retraction of an asserted fact in one transaction `API-020`, an empty window `API-019` and oversize values or idents `WAL-003`, each leaving the writer usable and the output exact, with repeats written once; that an existing target or WAL is `STG-043`, a second writer `STG-025` and `read_only` `API-014`; that a stale `.partial` is reused; that dropping an unfinished writer leaves nothing; that `finish` refuses a target created during the build and leaves it untouched; and that `LogWriter` is `Send`. Unit tests check that a batch commits only when a transaction opens past the batch size, never splitting one, and that `FileBackend::truncate` empties a file. A unit test in `error.rs` checks that `MinigrafError::invalid_argument` and `MinigrafError::closed`, the constructors for language bindings, give API-017 and API-018 with their text. Query-validation tests (#406, #414) check that a plain `:find` variable or a window function's argument, `:partition-by` or `:order-by` variable that no `:where` clause binds (including `?_` wildcards) is `PRS-080` from both `execute()` and `prepare()`, naming the variable, while variables bound by a pattern, an attribute position, a rule call, an expression binding, an `or-join`, every branch of an `or`, a pseudo-attribute or a window spec are accepted, and bind slots are not mistaken for `:find` variables; and that an unknown aggregate is `QRY-010` and an unknown window function `QRY-011` on an empty database, when no row matches, when rows match, through a rule query and through a prepared query, while a registered UDF in either position returns an empty result on an empty database. Coverage-gate tests (#396) check that replay stops at a WAL entry whose first fact is longer than 10 MB or that claims more than 1M facts, that a WAL writer cannot be opened in a missing directory, and that a failed cut of a torn WAL tail fails the open and leaves the file as it was; that a short v7 page and a slot past the record count are `INT-024`; that a migration backup meta of another generation or file length is not used (`STG-032`, file unchanged), and a torn first meta with data in page 1 is `STG-032`, not a new file; that `save` with nothing pending writes nothing, `into_backend` saves pending facts, and `rebuild_indexes` with nothing checkpointed commits the pending facts; and that unsorted or repeated B+tree input and a stored key inserted with another value are `INT-049`, a tree pointer to a free-list page is `STG-013` from `get`, a leaf walk and a copy-on-write insert, and the read-only `MutexStorageBackend` refuses writes, syncs and close. A page-cache test holds two threads in the backend read so both miss the same page, and checks that both get the one cached copy; before it, that path ran only when the scheduler happened to race the threads, and the gate flapped. Three `BrowserDb` tests (wasm32, run with `wasm-bindgen-test` in a browser, not counted above) check that `query()` batches of 1, 7 and 1,000 give the rows of `execute()`, that a cursor keeps the answer from when it opened and returns the end after `close()`, and that a `transact` is API-012. The wasm32 targets run in CI (`.github/workflows/wasm.yml`). In a browser (`wasm-pack test --headless`, Firefox and Chrome): 32 `wasm_bindgen_test` tests: nineteen for `BrowserDb` (including IndexedDB persistence, the cursor, and the flush returning every dirty page and failing on one it cannot read, from #470), eleven for `BrowserBufferBackend` (which used `#[test]` and never ran before), and two checking that IndexedDB page keys accept only whole numbers from 0 to 2^53. Five of the `BrowserDb` tests (#411, #440) check that opening a v7 store written straight into IndexedDB stores a v8 meta page and only the live pages, with nothing written on the next open; that after each of 60 transacts memory, and at the end IndexedDB, hold exactly the pages not on the free list while the file grows by under one page per transact; that a commit applied in memory but never flushed reopens at the previous one and keeps working; that an export after churn is as long as the meta says and imports back with its free pages dropped; and that `importGraph()` deletes the replaced database's IndexedDB keys. Two buffer tests check that releasing a page drops its bytes and any unflushed write, and that export writes a released page as zeros. A native storage test checks that a file whose free pages are all zeros (as the browser exports them), both just migrated from v7 and after 20 saves, opens, verifies clean and keeps committing, and the free-list test checks that a checkpoint reports exactly the ids it newly lists. On `wasm32-wasip1` under Wasmtime: the unit and integration tests that do not use a file run, 956 passing; ten that spawn threads or a child process are ignored on WASI. These are not counted above. A WAL test checks that a checkpoint with no facts to flush still removes a WAL whose entries are all already in the file, and the golden-corpus test now requires the WAL to be gone after a checkpoint for the `stale_wal` fixtures too (#457). A WAL test checks that closing a handle opened with `wal_checkpoint_threshold(usize::MAX)`, in both sync modes, leaves the `.graph` file byte for byte as it was and keeps the WAL, whose entries replay on the next open, and a unit test closes an in-memory handle under the same setting (#476). A recursive-rule test checks that a rule query on a database with more live facts than `max_results` returns its answer, with a bound and with a free first argument, and that a query deriving more facts than its `:max-results` is still `INT-020` (#528). A log-writer test runs the online backup from `docs/DURABILITY.md`: it copies a read-write handle with an uncheckpointed transaction through `fact_log` and `LogWriter` while another thread writes past the automatic checkpoint threshold, and checks that the copy verifies clean, holds exactly the records the log returned and the counter read after the log opened, and that every concurrent write reaches the source's file once the log is dropped (#398). Rule-body tests in `tests/recursive_rules_test.rs` check that a `:db/valid-at` pattern in a rule definition, in its `not` and `not-join` bodies and in its `or` branches binds the query's valid time, on the semi-naive, stratified and magic-sets paths (#617). Tests in `tests/not_join_test.rs` check that a rule calling a rule with `not-join`, `not` or `or` returns that rule's rows, with a free and a literal argument and through a forwarding rule, and that a rule with its own `not-join` calling one gets the same rows whether or not the query asks for the callee first; stratification unit tests check that such a caller, a rule forwarding to it and a caller of a recursive one sit above the callee, and that recursion through a rule with a `not` still stratifies (#628). Tests in `tests/not_join_test.rs` check that recursion through a rule with `not`, `not-join` or `or`, directly or through another rule, returns every row along a chain, and that the `not` still removes the rows it should; evaluator unit tests check that such recursion stops at its fixpoint that `max_iterations` and `max_results` end one that does not with INT-020, and that `max_results` counts the facts every pass derives, not only the mixed rule's (#633). @@ -22,6 +22,7 @@ This covers the unreleased file-format-v8 work (#287, #371): the index-key value - **Pseudo-attribute sources** (`tests/pseudo_source_test.rs`, 12 tests; #594): a per-fact pseudo-attribute reads the nearest earlier pattern in the text with the same entity term (#594's example in both text orders), two pseudo-attributes share a source, `[:alice ?a _]` gives one row per fact, constant and bound value terms check, `:alice` and its UUID are one term while `?e` and `:alice` are not, PRS-081 for a pattern after it, alone, `_`/`?_x` entities, another entity's pattern and a rule call; a `not` body reads the outer fact (and a source inside the body), `not-join` crosses for a listed variable only, `or` branches read the outer fact and never a sibling, `or-join`, closed rule bodies, `$slot`s by name at prepare time, and `:db/valid-at` once per fact. Parser unit tests check the hidden variables (`?#.`), the join variable a crossed `not-join` gains, and the PRS-081 text; an optimizer test places a pseudo-attribute pattern after its source and the pattern binding its value; matcher tests read the hidden variable, a substituted integer, and nothing without a source; the engine tests run the sourced shapes and check PRS-081 for the old fallback shapes. The plan-independence matrix gains eleven pseudo-attribute shapes. - **Negation equality** (`tests/negation_equality_test.rs`, 5 tests; #546): a value a `not` / `not-join` body takes from the outer row compares by join equality on every path (exclusion set, per-row body with an expression, a query that invokes a rule, a rule body): `1e-20` and `2e-20` differ, `0.0` equals `-0.0`, a keyword imported into an entity position equals the ref, and a float literal in the body still matches within `f64::EPSILON`; through `execute` and a cursor, in memory and after a checkpoint. Engine unit tests (`src/query/engine/tests.rs`, 6 more) run every negation shape (hashed and correlated bodies, keys on entity, value and attribute positions, floats, keywords and refs, long strings, bodies reading an imported variable, variable-free bodies, per-fact pseudo-attributes and `:db/valid-at` in bodies, deferred and repeated negations) at six temporal settings with each body form forced and reordering on and off, in memory and on file before and after checkpoints, against the executor; and check the form the planner picks, routing, multiplicity, failing expressions and unknown predicates in bodies, and that a hashed body is read once while a correlated one probes once per outer row. Optimizer unit tests cover `plan_seeded` and the `SUBPLANS` switch. - **Disjunction equality** (`tests/or_equality_test.rs`, 5 tests; #546): an `or` / `or-join` branch meets the outer row by join equality on every executor path (a keyword in the outer row meets a ref in a branch; `0.0` and `-0.0`; floats `1e-20` apart do not), branches naming different outer variables each meet the outer row on their own, and an `or-join` branch that does not bind its join variable keeps every outer row when it has a row; through `execute` and a cursor, in memory and after a checkpoint. Engine unit tests run every disjunction shape with each branch form forced and reordering on and off against the executor, and cover routing (rule invocations, nested `or`, registry predicates, branch-local metadata stay on the executor), the branch form the planner picks, `Distinct` (equal rows collapse; a keyword and a ref, `-0.0` and `0.0`), and that a hashed branch is read once and a correlated one runs only for rows earlier branches did not match. +- **Aggregates and windows** (`tests/aggregate_window_rules_test.rs`, 8 tests; #547): exact float `sum` and window `avg`, integer `sum` saturating once, exact window `sum` of integers above 2^53, `min`/`max` of `-0.0` and `0.0`, `rank` by keywords (peers 1 1 3 3), a running `sum` giving peers the value at the last peer, `row-number` breaking ties by the other `:find` variables, and `INT-041` from a failing group. `src/query/datalog/aggregate.rs` (15 unit tests) checks every accumulator over every order of small inputs and that `value_order` is a total order; `tests/window_functions_test.rs` checks `PRS-082` and a window over the groups of an aggregate query, and `tests/plan_independence_test.rs` runs aggregate, window and mixed queries with reordering on and off. - **Plan independence** (`tests/plan_independence_test.rs`, 15 tests, `test-oracle` feature only; #596): a matrix of queries (joins in both text orders, cross products, keyword and ref, floats, long strings, filter and binding expressions, recursive rules and rules with `not`, `or`/`or-join` including the #250 short-circuit shapes, placed and deferred `not`/`not-join`, aggregates, `:as-of`, `:valid-at`, `:any-valid-time`), in memory and on file after a checkpoint and a reopen, through `execute` and a cursor, gives the same bag of rows with clause reordering on and off (`minigraf::__reorder`) and with the engine's negation bodies and `or` branches chosen, forced hashed and forced correlated (`minigraf::__subplans`, #546; the matrix gains #546's negation and disjunction shapes); #596's example and the unification cases (an expression whose output is its input, two expressions on one variable, integer vs float, `-0.0` vs `0.0`); and `or` / `or-join` branches whose expressions read or bind from a variable only the outer row binds, nested and in rule bodies (#620); and expressions, chains of them and a `not` after them, reading a variable only an `or` binds, in queries, with rules and in rule bodies (#625; `tests/disjunction_test.rs` repeats them without `test-oracle`). Under `test-oracle` the model-based and property tests also rerun every generated query with reordering off (the property test also with negation bodies and `or` branches forced hashed; it generates `or` / `or-join` over bound variables, #546). Unit tests cover `branch_outer_reads` (#620), the planner deferring an expression only an `or` can bind for, after the expressions placed before it and ahead of the deferred `not`s (#625), `apply_expr` (equal, different, integer vs float, NaN, `-0.0`/`0.0` with and without replacing, keyword vs ref both ways, a failing expression), `Binders::expression`, the switch (source order kept, per thread, restored on drop), and the engine planner's `Bind` modes and build side with the switch off; the engine's join matrix gains six unification shapes. - **Public diagnostics**: structured parser, query, storage, WAL, API, and internal error codes; the error-code registry is checked against [`ERROR_REFERENCE.md`](ERROR_REFERENCE.md). - **Compatibility and quality**: golden-file corpus of released file formats, cross-platform format compatibility, property-based tests, long-haul smoke coverage, a weekly soak at 10M facts with a 5-hour churn (#392), XTDB/Datomic semantic compatibility, grammar conformance, and rustdoc examples. diff --git a/docs/superpowers/plans/2026-10-20-547-aggregate-window.md b/docs/superpowers/plans/2026-10-20-547-aggregate-window.md new file mode 100644 index 000000000..0dc7d4224 --- /dev/null +++ b/docs/superpowers/plans/2026-10-20-547-aggregate-window.md @@ -0,0 +1,148 @@ +# Aggregation and window functions on the streaming engine (#547) — Plan + +Spec: `docs/superpowers/specs/2026-10-20-aggregate-window-engine-design.md`. PRs target +`v3`. None uses a closing keyword for #473; #547 closes by hand after PR 3. + +| PR | Steps | Branch | +|---|---|---| +| 1 | 1–8: the rules on the executor (`value_order`, `aggregate.rs`, peers, `PRS-082`) | `feat/547-rules` | +| 2 | 9–12: `Aggregate` on the engine | `feat/547-aggregate` | +| 3 | 13–16: `Window` on the engine | `feat/547-window` | + +Each PR leaves `v3` with no unused code, updates the test counts and `CHANGELOG.md` for +what it ships, and has a docs-site PR (`../minigraf-docs`) when it changes documented +behaviour (PR 1 does; PRs 2 and 3 change no answer). + +## What PR 1 replaces + +| Today (`v3`) | Used by | After PR 1 | +|---|---|---| +| `functions::AggState`, `WindowOps`, `AggImpl::Builtin(WindowOps)` | registry built-ins, `apply_window_functions` | `AggImpl::Builtin(Builtin)`, `Builtin` an enum of the seven built-ins; the arithmetic lives in `aggregate.rs` | +| `functions::apply_builtin_aggregate` | `compute_aggregation` | `aggregate::Agg` | +| `functions::value_lt` | built-in window `min`/`max` | `aggregate::value_order` | +| `functions::value_cmp` | window sort (`executor.rs:1799`) | `aggregate::value_order` | +| `FunctionRegistry::is_window_compatible`, `is_known` (`#[allow(dead_code)]`) | tests only | removed with their tests | + +The parser already rejects `count-distinct`/`sum-distinct` in a window (`PRS-031`) and +`avg`/`rank`/`row-number` as aggregates (`PRS-027`). Registered aggregates (`AggImpl::Udf`) +are unchanged. + +## Design decisions not fixed by the spec + +1. **Window `sum` and `avg` skip values that are not numbers**, as the built-in + `WindowOps` do today; only the aggregate `sum` fails with `INT-041` (spec §3.2 names + the aggregate). Window `sum` is an `Integer` while every number so far was an + integer, else a `Float`, as today, with the exact sum of spec §3.2. No numbers so + far: `Integer(0)` for `sum`, `nil` for `avg`. +2. **Exact float sums use a fixed-point accumulator**, not Shewchuk partials: a signed + integer of 36 `u64` limbs counting units of 2^-1074, which holds the sum of 2^64 + values of any finite `f64` without overflow. It is boxed and allocated on the first + finite float, so integer-only and empty accumulators stay small. Integers are summed + in an `i128` beside it and added in at the end. NaN, `+inf`, `-inf` and "every float + was `-0.0`" are flags. Rounding to `f64` is round-half-even on the limbs. Reason: no + intermediate overflow to handle, and the same code removes a value exactly (#183). +3. **`value_order` is a free function in `aggregate.rs`**, `pub(crate)`, used by the + window sort, peers, `rank`, window and aggregate `min`/`max`. An `i64` and an `f64` + compare exactly: a float outside `[-2^63, 2^63)` is beyond every integer; otherwise + compare the integer with `f.floor()` as `i128`, then the fraction. +4. **`aggregate::Agg`** (one per aggregate per group): `Agg::new(func, registry)`, + `step(&Value)`, `finish(self) -> Result>`, where `Ok(None)` is + spec §3.2's dropped group (`INT-042`, today's string match on "no non-null values in + group" goes). Variants: `Count(i64)`, `Sum(Sum, first non-number's type)`, + `Min/Max(Extreme)`, `CountDistinct/SumDistinct(BTreeSet)` with canonical + floats, `Udf`. No `Avg`: `avg` is a window function only (`PRS-027`), so + `Agg::new("avg")` is `INT-029`. `Extreme` records a bitset of kinds and the best + value. +5. **`aggregate::Running`** (one per window per partition): `Running::new(func, + registry)`, `step(&Value)`, `value(&self) -> Value` (a registered window's + `finalise(acc, rows_stepped)`). `apply_window_functions` steps a whole peer group, + then reads `value()` once for all its rows. +6. **Partitions by `Value`'s `Ord`**: a `BTreeMap>` (a missing + partition variable reads `nil`), replacing today's `HashMap, _>`, so + NaN keys share one partition (spec §4.1). +7. **Named variables** (spec §2) are computed once per query by + `aggregate::named_vars(find)`; the sort reads them from each binding (`nil` when + absent, which only happens for a missing grouping value). +8. **`PRS-082`** text: `"{} in {} must be a plain :find variable when the query has aggregates"` + (variable, the window's display name). Checked in `parse_query` after `PRS-080`. + +## PR 1 — the rules on the executor + +1. **`aggregate.rs` order and sums.** `value_order`; the fixed-point `Exact` + accumulator (add `f64`, add `i64`, round). Unit tests: `value_order` is a total + order consistent with itself on a grid of values (every class, `±0.0`, NaNs with + both signs, `i64::MIN/MAX` against nearby floats, `2^53 + 1` against `2^53` as float); + `Exact` against hand-computed results (cancellation `1e308 + 1e308 - 1e308`, + `0.1` ten times, subnormals, `-0.0` only, overflow to `inf`, `i64::MAX` plus `0.5`) + and against every permutation of 6-element inputs. +2. **`Agg` and `Running`** (decisions 1, 4, 5). Unit tests per function: empty input, + `nil` only, saturation, errors decided at `finish` whatever the order (`min` over + `[kw, kw, 1]` and `[1, kw, kw]` both `INT-043`), canonical distinct floats, a + registered aggregate's `n`. +3. **Registry.** `AggImpl::Builtin(Builtin)`; drop `AggState`, `WindowOps`, + `apply_builtin_aggregate`, `value_lt`, `value_cmp`, `is_window_compatible`, + `is_known` and their tests (`functions.rs:520–800`), moving each still-meaningful + case to step 1 or 2. Update the `eval_binop` comment that cites `value_lt`/`value_cmp`. +4. **`compute_aggregation`** on `Agg`: groups stay a `BTreeMap, _>`, now of + accumulators instead of bindings (one pass); finish groups in key order with spec + §3.3's error rule. The zero row for `count`-only queries is unchanged. +5. **`apply_window_functions`** on decisions 5–7: partition, sort by (order key with + direction, named variables, input index), walk peer groups. Output stays in input + order. +6. **`PRS-082`** (decision 8): `ErrorCode::Prs082` in the four places `Prs081` + appears in `error.rs`, `docs/ERROR_REFERENCE.md` (table row and section with cause, + resolution, example), parser check, parser test. +7. **Integration tests** (`tests/aggregation_test.rs`, `tests/window_functions_test.rs`): + every case of spec §3.4 and the §3.1 / §4.2 examples, each run with + `minigraf::__reorder` on and off under `test-oracle` (as `plan_independence_test.rs` + does), plus ungated copies of the plain assertions so the coverage job (which runs + without `test-oracle`) counts them. Fix existing tests that asserted input-order + ties; list each in the PR body. +8. **Docs and counts.** `CHANGELOG.md` (v3.0.0 breaking: spec §3.4 items 1–6), + `register_aggregate` doc comment (order dependence, spec §3.2), CLAUDE.md module list + (`aggregate.rs`) and test count, `docs/TEST_COVERAGE.md`, `ROADMAP.md` if it lists + #547 status. Docs-site PR: `content/datalog-reference.md` window section (peers, + value order, frame sentence), aggregate section (exact sums, `min`/`max` of floats), + `@since v3.0.0` / `@until v3.0.0` around the changed text. File the follow-up issue + (index-order sort skip and sort-grouped `Aggregate`, with #306) and link it from + the PR. + +## PR 2 — `Aggregate` on the engine + +9. **Routing.** `plan::plan` accepts `FindSpec::Aggregate` and `:with` (windows still + return `None`). It calls `check_find_functions` first (made `pub(crate)`). `Plan` + gets `find: Vec` describing each output column (a slot, or an aggregate + result), built by a new `Step::Aggregate { groups: Vec, aggs: Vec<(String, + usize)> }` at the end of `steps`. +10. **`engine/aggregate.rs`.** Ungrouped: one `Agg` per aggregate; `count` steps only on + whether the cell is `Val(Null)`; `count-distinct` keys on the cell (`JoinKey::of` + with keyword-to-ref off, long strings by content, floats canonical) and converts + to `Value` only at finish; other functions convert the argument cell to a `Value` + (`project::value`). Grouped: `HashMap, Vec>` where `GroupKey` + compares as `Value`'s `Ord` (as `JoinKey` but NaN equal to NaN, keyword never a + ref); at the end, convert keys, sort by `Value`'s `Ord`, finish (spec §3.3), emit + batches whose cells are the group cells and the results as `Cell::Val`. A result + that is a `Ref` or `Keyword` (from `min`/`max` this cannot happen; from a + registered aggregate it can) needs a cell: add `Cell::Value(Box)`, projected + as is, if a registered aggregate can return one; otherwise restrict to scalars and + check in `project`. +11. **Tests.** Engine unit tests: counters show `Aggregate` held no rows for an + ungrouped `count`; empty input; zero row; error in a later group; `:with`. Routing + test (`RUNS`) that aggregate queries run on the engine. The oracle covers the + integration suites. +12. **Measurement.** Envelope `count` and `count_grouped` at 100K, 1M, 10M on the bench + machine (idle check first), peak RSS for `count` flat across scales; numbers in the + PR body and `docs/BENCHMARKS.md`. + +## PR 3 — `Window` on the engine + +13. **Routing.** `plan::plan` accepts windows. `Step::Window(Vec, slots)` + after `Step::Aggregate` (mixed) or after the `:where` steps. +14. **`engine/window.rs`.** Holds every input row's cells of the named variables and + the window arguments, converted to `Value`s on arrival (the sort compares values), + plus input position; at end of input computes each window with the PR 1 code + (factor `apply_window_functions`' per-partition core into `aggregate.rs` so both + engines call it over `&[Vec]`), emits rows in input order. +15. **Tests.** Routing; mixed aggregate + window; oracle over `window_functions_test.rs`. +16. **Close.** CLAUDE.md engine paragraph, counts; close #547 by hand; docs-site PR only + if the architecture page lists engine operators. diff --git a/src/db.rs b/src/db.rs index bc13a0b29..88eb10f99 100644 --- a/src/db.rs +++ b/src/db.rs @@ -1288,6 +1288,16 @@ impl Minigraf { /// It is type-erased internally. The function is usable in both `:find` /// grouping position and `:over` (window) position. /// + /// As an aggregate, `step` receives the group's non-`nil` values in the + /// order the query produces them, and `finalise` receives their count; + /// with no values the result is `nil`. In a window, `step` receives + /// every value, `nil` included, in the window's order, and `finalise` + /// is called once per set of rows with equal `:order-by` values, with + /// the number of rows stepped so far. The order of an aggregate's + /// values can differ between equivalent queries, so a function whose + /// result depends on that order (anything but a commutative, + /// associative combination) may answer them differently. + /// /// # Errors /// /// Returns an error if `name` is already registered (built-in or UDF) @@ -1349,7 +1359,6 @@ impl Minigraf { step: step_boxed, finalise: finalise_boxed, }), - is_builtin: false, }; self.inner .functions diff --git a/src/error.rs b/src/error.rs index 98ace72cf..aa532820d 100644 --- a/src/error.rs +++ b/src/error.rs @@ -128,6 +128,7 @@ pub(crate) enum ErrorCode { Prs079, Prs080, Prs081, + Prs082, Qry001, Qry002, Qry003, @@ -772,6 +773,12 @@ pub(crate) const REGISTRY: &[(ErrorCode, &str, &str, ErrorCategory)] = &[ "{} on {} has no pattern before it, with the same entity, that reads a stored attribute; add one (e.g. [{} ?a _]) before it", ErrorCategory::Parser, ), + ( + ErrorCode::Prs082, + "PRS-082", + "{} in {} must be a plain :find variable when the query has aggregates", + ErrorCategory::Parser, + ), ( ErrorCode::Qry001, "QRY-001", @@ -1992,6 +1999,7 @@ mod tests { | ErrorCode::Prs079 | ErrorCode::Prs080 | ErrorCode::Prs081 + | ErrorCode::Prs082 | ErrorCode::Qry001 | ErrorCode::Qry002 | ErrorCode::Qry003 @@ -2219,6 +2227,7 @@ mod tests { ErrorCode::Prs079, ErrorCode::Prs080, ErrorCode::Prs081, + ErrorCode::Prs082, ErrorCode::Qry001, ErrorCode::Qry002, ErrorCode::Qry003, diff --git a/src/query/datalog/aggregate.rs b/src/query/datalog/aggregate.rs new file mode 100644 index 000000000..fbeaadeb9 --- /dev/null +++ b/src/query/datalog/aggregate.rs @@ -0,0 +1,592 @@ +//! Aggregates and the order of values (#547 spec §3): one accumulator per +//! built-in function, used by the executor and the engine, whose result +//! does not depend on the order of its input. + +use super::functions::{AggImpl, Builtin, FunctionRegistry, UdfFinaliseFn, UdfStepFn}; +use super::types::{FindSpec, WindowSpec}; +use crate::error::{ErrorCode, err_coded}; +use crate::graph::types::Value; +use anyhow::Result; +use std::any::Any; +use std::cmp::Ordering; +use std::collections::BTreeSet; + +/// The class of a value in [`value_order`]: numbers, strings, booleans, +/// refs, keywords, `nil`. +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, + } +} + +/// The total order of values (spec §3.1): numbers by exact numeric value +/// (an `Integer` before an equal `Float`, floats then by `f64::total_cmp`, +/// NaN after every other number), then strings, booleans, refs, keywords +/// and `nil`. `Equal` only for identical values. +pub(crate) fn value_order(a: &Value, b: &Value) -> Ordering { + match (a, b) { + (Value::Integer(x), Value::Integer(y)) => x.cmp(y), + (Value::Float(x), Value::Float(y)) => match (x.is_nan(), y.is_nan()) { + (false, false) => x + .partial_cmp(y) + .unwrap_or(Ordering::Equal) + .then_with(|| x.total_cmp(y)), + (true, true) => x.total_cmp(y), + (true, false) => Ordering::Greater, + (false, true) => Ordering::Less, + }, + (Value::Integer(i), Value::Float(f)) => int_float(*i, *f).then(Ordering::Less), + (Value::Float(f), Value::Integer(i)) => int_float(*i, *f).reverse().then(Ordering::Greater), + (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), + (Value::Null, Value::Null) => Ordering::Equal, + _ => class(a).cmp(&class(b)), + } +} + +/// `i` against `f` by exact numeric value; a NaN `f` is greater. +fn int_float(i: i64, f: f64) -> Ordering { + if f.is_nan() { + return Ordering::Less; + } + // 2^63: every float at or above it exceeds every i64, every float + // below -2^63 is less than every i64. + const TWO_63: f64 = 9_223_372_036_854_775_808.0; + if f >= TWO_63 { + return Ordering::Less; + } + if f < -TWO_63 { + return Ordering::Greater; + } + let floor = f.floor(); + // In [-2^63, 2^63), so the floor is an exact integer in i128 range. + #[allow(clippy::cast_possible_truncation)] + let whole = floor as i128; + i128::from(i).cmp(&whole).then(if f > floor { + Ordering::Less + } else { + Ordering::Equal + }) +} + +/// Limbs of [`Exact`]: 36 × 64 bits hold the sum of 2^64 values of any +/// finite `f64` and the `i128` integer total, in units of 2^-1074. +const LIMBS: usize = 36; + +/// The bit of 2^0 in [`Exact`]'s units of 2^-1074. +const ONE: u32 = 1074; + +/// An exact sum of finite floats and integers: a two's complement integer +/// counting units of 2^-1074, the smallest positive `f64`. +#[derive(Clone)] +struct Exact { + limbs: [u64; LIMBS], +} + +impl Exact { + fn new() -> Self { + Exact { limbs: [0; LIMBS] } + } + + /// Add or subtract `mag << shift`. + fn add_shifted(&mut self, mag: u64, shift: u32, neg: bool) { + let idx = (shift / 64) as usize; + let wide = u128::from(mag) << (shift % 64); + let parts = [low(wide), low(wide >> 64)]; + let mut carry = false; + for (i, limb) in self.limbs.iter_mut().enumerate().skip(idx) { + let part = parts.get(i - idx).copied().unwrap_or(0); + if part == 0 && !carry && i > idx + 1 { + break; + } + if neg { + let (d, b1) = limb.overflowing_sub(part); + let (d, b2) = d.overflowing_sub(u64::from(carry)); + *limb = d; + carry = b1 || b2; + } else { + let (s, c1) = limb.overflowing_add(part); + let (s, c2) = s.overflowing_add(u64::from(carry)); + *limb = s; + carry = c1 || c2; + } + } + } + + /// Add a finite float. + fn add_float(&mut self, f: f64) { + let bits = f.to_bits(); + let neg = bits >> 63 == 1; + let exp = u32::try_from((bits >> 52) & 0x7ff).unwrap_or(0); + let frac = bits & ((1 << 52) - 1); + if exp == 0 { + self.add_shifted(frac, 0, neg); + } else { + self.add_shifted(frac | (1 << 52), exp - 1, neg); + } + } + + /// Add an integer. + fn add_int(&mut self, i: i128) { + let mag = i.unsigned_abs(); + let neg = i < 0; + self.add_shifted(low(mag), ONE, neg); + self.add_shifted(low(mag >> 64), ONE + 64, neg); + } + + fn bit(mag: &[u64; LIMBS], p: u32) -> bool { + mag.get((p / 64) as usize) + .is_some_and(|l| (l >> (p % 64)) & 1 == 1) + } + + /// Whether any bit of `mag` below `p` is set. + fn any_below(mag: &[u64; LIMBS], p: u32) -> bool { + let full = (p / 64) as usize; + mag.iter().take(full).any(|&l| l != 0) + || mag + .get(full) + .is_some_and(|l| !p.is_multiple_of(64) && l & ((1u64 << (p % 64)) - 1) != 0) + } + + /// The sum rounded once to the nearest `f64`, ties to even; `None` + /// when it is zero. + fn round(&self) -> Option { + let neg = self.limbs.last().is_some_and(|l| l >> 63 == 1); + let mut mag = self.limbs; + if neg { + let mut carry = true; + for limb in &mut mag { + let (s, c) = (!*limb).overflowing_add(u64::from(carry)); + *limb = s; + carry = c; + } + } + let top = mag.iter().rposition(|&l| l != 0)?; + let top_limb = mag.get(top).copied().unwrap_or(0); + let h = u32::try_from(top).unwrap_or(0) * 64 + (63 - top_limb.leading_zeros()); + let value = if h < 53 { + // Below 2^53 units: a multiple of 2^-1074 under 2^-1021, exact. + let m = mag.first().copied().unwrap_or(0); + #[allow(clippy::cast_precision_loss)] + let m = m as f64; + m * f64::from_bits(1) + } else { + let s = h - 52; + let mut m: u64 = 0; + for p in (s..=h).rev() { + m = (m << 1) | u64::from(Self::bit(&mag, p)); + } + let half = Self::bit(&mag, s - 1); + let sticky = Self::any_below(&mag, s - 1); + let mut s = s; + if half && (sticky || m & 1 == 1) { + m += 1; + if m == 1 << 53 { + m >>= 1; + s += 1; + } + } + // The value is m · 2^(s-1074) with m in [2^52, 2^53): biased + // exponent s + 1. + let biased = u64::from(s) + 1; + if biased >= 0x7ff { + f64::INFINITY + } else { + f64::from_bits((biased << 52) | (m & ((1 << 52) - 1))) + } + }; + Some(if neg { -value } else { value }) + } +} + +/// The low 64 bits of `v`. +#[allow(clippy::cast_possible_truncation)] +fn low(v: u128) -> u64 { + v as u64 +} + +/// An exact sum of numbers (spec §3.2). +#[derive(Clone, Default)] +pub(crate) struct Sum { + ints: i128, + /// The exact sum of the finite floats, allocated on the first one. + floats: Option>, + /// Numbers taken. + count: u64, + any_float: bool, + nan: bool, + pos_inf: bool, + neg_inf: bool, + /// Whether some number taken is not `-0.0`. + not_neg_zero: bool, +} + +impl Sum { + /// Take `v`; `false` when it is not a number. + pub(crate) fn add(&mut self, v: &Value) -> bool { + match v { + Value::Integer(i) => { + self.ints += i128::from(*i); + self.not_neg_zero = true; + } + Value::Float(f) => { + self.any_float = true; + if f.is_nan() { + self.nan = true; + } else if f.is_infinite() { + if *f > 0.0 { + self.pos_inf = true; + } else { + self.neg_inf = true; + } + } else { + if f.to_bits() != (-0.0f64).to_bits() { + self.not_neg_zero = true; + } + self.floats + .get_or_insert_with(|| Box::new(Exact::new())) + .add_float(*f); + } + } + _ => return false, + } + self.count += 1; + true + } + + /// The sum as a float, rounded once. + fn float(&self) -> f64 { + if self.nan || (self.pos_inf && self.neg_inf) { + return f64::NAN; + } + if self.pos_inf { + return f64::INFINITY; + } + if self.neg_inf { + return f64::NEG_INFINITY; + } + let mut exact = self.floats.as_deref().cloned().unwrap_or_else(Exact::new); + exact.add_int(self.ints); + match exact.round() { + Some(f) => f, + None if self.count > 0 && !self.not_neg_zero => -0.0, + None => 0.0, + } + } + + /// `Integer` while every number was an integer, saturated; else + /// `Float`. No numbers: `Integer(0)`. + pub(crate) fn value(&self) -> Value { + if self.any_float { + Value::Float(self.float()) + } else { + let clamped = self.ints.clamp(i128::from(i64::MIN), i128::from(i64::MAX)); + Value::Integer(i64::try_from(clamped).unwrap_or(i64::MAX)) + } + } + + /// The mean of the numbers, or `nil` without any. + pub(crate) fn mean(&self) -> Value { + if self.count == 0 { + return Value::Null; + } + #[allow(clippy::cast_precision_loss)] + let n = self.count as f64; + Value::Float(self.float() / n) + } +} + +/// `v` with `-0.0` as `0.0` and every NaN as one NaN, so equal values under +/// `Value`'s `Ord` are stored alike. +fn canonical(v: &Value) -> Value { + match v { + Value::Float(f) if f.is_nan() => Value::Float(f64::NAN), + Value::Float(f) if *f == 0.0 => Value::Float(0.0), + other => other.clone(), + } +} + +/// The type name used in error messages. +fn type_name(v: &Value) -> &'static str { + match v { + Value::String(_) => "String", + Value::Integer(_) => "Integer", + Value::Float(_) => "Float", + Value::Boolean(_) => "Boolean", + Value::Ref(_) => "Ref", + Value::Keyword(_) => "Keyword", + Value::Null => "Null", + } +} + +/// Variant kinds, in `Value`'s declaration order, for `INT-043`. +const KINDS: [&str; 7] = [ + "String", "Integer", "Float", "Boolean", "Ref", "Keyword", "Null", +]; + +fn kind(v: &Value) -> u8 { + match v { + Value::String(_) => 0, + Value::Integer(_) => 1, + Value::Float(_) => 2, + Value::Boolean(_) => 3, + Value::Ref(_) => 4, + Value::Keyword(_) => 5, + Value::Null => 6, + } +} + +/// The least or greatest value by [`value_order`], and the kinds met. +#[derive(Clone, Default)] +pub(crate) struct Extreme { + best: Option, + kinds: u8, +} + +impl Extreme { + fn add(&mut self, v: &Value, max: bool) { + if matches!(v, Value::Null) { + return; + } + self.kinds |= 1 << kind(v); + let better = match &self.best { + None => true, + Some(b) => { + let o = value_order(v, b); + if max { + o == Ordering::Greater + } else { + o == Ordering::Less + } + } + }; + if better { + self.best = Some(v.clone()); + } + } +} + +/// A registered aggregate's state. +pub(crate) struct Udf { + acc: Box, + step: UdfStepFn, + finalise: UdfFinaliseFn, + /// Values stepped. + n: usize, +} + +impl Udf { + fn of(registry: &FunctionRegistry, name: &str) -> Option { + match ®istry.get(name)?.impl_ { + AggImpl::Udf(ops) => Some(Udf { + acc: (ops.init)(), + step: ops.step.clone(), + finalise: ops.finalise.clone(), + n: 0, + }), + AggImpl::Builtin(_) => None, + } + } + + fn add(&mut self, v: &Value) { + (self.step)(&mut self.acc, v); + self.n += 1; + } + + fn value(&self) -> Value { + (self.finalise)(&self.acc, self.n) + } +} + +/// One aggregate of one group (spec §3.2). Takes values one at a time and +/// never fails while taking them; [`Agg::finish`] decides the result. +pub(crate) enum Agg { + Count(i64), + /// The sum, and the type of the first value that is not a number. + Sum(Sum, Option<&'static str>), + Min(Extreme), + Max(Extreme), + CountDistinct(BTreeSet), + SumDistinct(BTreeSet), + Udf(Udf), +} + +impl Agg { + /// The accumulator of the aggregate `func`, or `INT-029` when it is + /// neither built in nor registered. + pub(crate) fn new(func: &str, registry: &FunctionRegistry) -> Result { + let desc = registry + .get(func) + .ok_or_else(|| err_coded!(ErrorCode::Int029, func))?; + Ok(match &desc.impl_ { + AggImpl::Builtin(b) => match b { + Builtin::Count => Agg::Count(0), + Builtin::Sum => Agg::Sum(Sum::default(), None), + // `avg` is a window function only (`PRS-027`). + Builtin::Avg => return Err(err_coded!(ErrorCode::Int029, func)), + Builtin::Min => Agg::Min(Extreme::default()), + Builtin::Max => Agg::Max(Extreme::default()), + Builtin::CountDistinct => Agg::CountDistinct(BTreeSet::new()), + Builtin::SumDistinct => Agg::SumDistinct(BTreeSet::new()), + }, + AggImpl::Udf(_) => Agg::Udf( + Udf::of(registry, func).ok_or_else(|| err_coded!(ErrorCode::Int029, func))?, + ), + }) + } + + /// Take one value; `nil` is skipped by every aggregate. + pub(crate) fn step(&mut self, v: &Value) { + if matches!(v, Value::Null) { + return; + } + match self { + Agg::Count(n) => *n = n.saturating_add(1), + Agg::Sum(s, bad) => { + if !s.add(v) && bad.is_none() { + *bad = Some(type_name(v)); + } + } + Agg::Min(e) => e.add(v, false), + Agg::Max(e) => e.add(v, true), + Agg::CountDistinct(set) | Agg::SumDistinct(set) => { + set.insert(canonical(v)); + } + Agg::Udf(u) => u.add(v), + } + } + + /// The result: `Ok(None)` when the group is dropped (`min`/`max` with no + /// value, `INT-042`), or the error the values make (spec §3.3). + pub(crate) fn finish(self, func: &str) -> Result> { + Ok(Some(match self { + Agg::Count(n) => Value::Integer(n), + Agg::Sum(_, Some(bad)) => return Err(err_coded!(ErrorCode::Int041, bad)), + Agg::Sum(s, None) => s.value(), + Agg::Min(e) | Agg::Max(e) => { + let Some(best) = e.best else { + return Ok(None); + }; + if e.kinds.count_ones() > 1 { + let mut names = KINDS + .iter() + .enumerate() + .filter(|(i, _)| e.kinds & (1 << i) != 0) + .map(|(_, n)| *n); + let first = names.next().unwrap_or("?"); + let second = names.next().unwrap_or("?"); + return Err(err_coded!(ErrorCode::Int043, func, first, second)); + } + if !matches!(best, Value::Integer(_) | Value::Float(_) | Value::String(_)) { + return Err(err_coded!(ErrorCode::Int044, func, type_name(&best))); + } + best + } + Agg::CountDistinct(set) => Value::Integer(i64::try_from(set.len()).unwrap_or(i64::MAX)), + Agg::SumDistinct(set) => { + let mut s = Sum::default(); + for v in &set { + if !s.add(v) { + return Err(err_coded!(ErrorCode::Int041, type_name(v))); + } + } + s.value() + } + Agg::Udf(u) if u.n == 0 => Value::Null, + Agg::Udf(u) => u.value(), + })) + } +} + +/// A window function's running value over a sorted partition (spec §4.2). +pub(crate) enum Running { + Count(i64), + /// Numbers only; other values are skipped. + Sum(Sum), + Avg(Sum), + Min(Extreme), + Max(Extreme), + /// Steps every value, `nil` included. + Udf(Udf), +} + +impl Running { + /// The accumulator of `ws`, or `INT-030` when its function is unknown. + /// Not for `rank` or `row-number`. + pub(crate) fn new(ws: &WindowSpec, registry: &FunctionRegistry) -> Result { + let name = ws.func_name(); + let unknown = || err_coded!(ErrorCode::Int030, name.clone()); + let desc = registry.get(&name).ok_or_else(unknown)?; + Ok(match &desc.impl_ { + AggImpl::Builtin(Builtin::Count) => Running::Count(0), + AggImpl::Builtin(Builtin::Sum) => Running::Sum(Sum::default()), + AggImpl::Builtin(Builtin::Avg) => Running::Avg(Sum::default()), + AggImpl::Builtin(Builtin::Min) => Running::Min(Extreme::default()), + AggImpl::Builtin(Builtin::Max) => Running::Max(Extreme::default()), + AggImpl::Builtin(Builtin::CountDistinct | Builtin::SumDistinct) => { + return Err(unknown()); + } + AggImpl::Udf(_) => Running::Udf(Udf::of(registry, &name).ok_or_else(unknown)?), + }) + } + + pub(crate) fn step(&mut self, v: &Value) { + match self { + Running::Count(n) => { + if !matches!(v, Value::Null) { + *n = n.saturating_add(1); + } + } + Running::Sum(s) | Running::Avg(s) => { + s.add(v); + } + Running::Min(e) => e.add(v, false), + Running::Max(e) => e.add(v, true), + Running::Udf(u) => u.add(v), + } + } + + pub(crate) fn value(&self) -> Value { + match self { + Running::Count(n) => Value::Integer(*n), + Running::Sum(s) => s.value(), + Running::Avg(s) => s.mean(), + Running::Min(e) | Running::Max(e) => e.best.clone().unwrap_or(Value::Null), + Running::Udf(u) => u.value(), + } + } +} + +/// The variables a window's tie-break reads (spec §2, "named variables"), +/// in `:find` order, each once. +pub(crate) fn named_vars(find: &[FindSpec]) -> Vec { + let aggregate = find.iter().any(|s| matches!(s, FindSpec::Aggregate { .. })); + let mut out: Vec = Vec::new(); + let mut push = |v: &String| { + if !out.contains(v) { + out.push(v.clone()); + } + }; + for spec in find { + match spec { + FindSpec::Variable(v) => push(v), + FindSpec::Window(ws) if !aggregate => { + ws.var.iter().for_each(&mut push); + ws.partition_by.iter().for_each(&mut push); + push(&ws.order_by); + } + FindSpec::Window(_) | FindSpec::Aggregate { .. } => {} + } + } + out +} + +#[cfg(test)] +mod tests; diff --git a/src/query/datalog/aggregate/tests.rs b/src/query/datalog/aggregate/tests.rs new file mode 100644 index 000000000..81b94bbc5 --- /dev/null +++ b/src/query/datalog/aggregate/tests.rs @@ -0,0 +1,407 @@ +use super::*; +use crate::query::datalog::functions::{AggregateDesc, UdfOps}; +use crate::query::datalog::parser::parse_datalog_command; +use crate::query::datalog::types::DatalogCommand; +use std::sync::Arc; + +fn f(x: f64) -> Value { + Value::Float(x) +} +fn i(x: i64) -> Value { + Value::Integer(x) +} +fn kw(k: &str) -> Value { + Value::Keyword(k.to_string()) +} + +/// The exact float sum of `vals`, as `Sum` gives it. +fn fsum(vals: &[Value]) -> f64 { + let mut s = Sum::default(); + for v in vals { + assert!(s.add(v), "a number"); + } + match s.value() { + Value::Float(x) => x, + _ => f64::NAN, + } +} + +/// Every permutation of `vals`. +fn permutations(vals: &[Value]) -> Vec> { + if vals.len() <= 1 { + return vec![vals.to_vec()]; + } + let mut out = Vec::new(); + for k in 0..vals.len() { + let mut rest = vals.to_vec(); + let first = rest.remove(k); + for mut p in permutations(&rest) { + p.insert(0, first.clone()); + out.push(p); + } + } + out +} + +fn finish(func: &str, vals: &[Value]) -> Result> { + let reg = FunctionRegistry::with_builtins(); + let mut agg = Agg::new(func, ®)?; + for v in vals { + agg.step(v); + } + agg.finish(func) +} + +fn code(r: Result>) -> &'static str { + match r { + Err(e) => crate::MinigrafError::from(e).code(), + Ok(_) => "", + } +} + +#[test] +fn value_order_is_a_total_order() { + let grid = vec![ + f(f64::NEG_INFINITY), + i(i64::MIN), + f(-9_223_372_036_854_775_808.0), + i(-1), + f(-0.5), + f(-0.0), + i(0), + f(0.0), + f(5e-324), + i(1), + f(1.0), + f(1.5), + i(9_007_199_254_740_993), + f(9_007_199_254_740_992.0), + i(i64::MAX), + f(9_223_372_036_854_775_808.0), + f(f64::INFINITY), + f(f64::NAN), + f(-f64::NAN), + Value::String("a".into()), + Value::String("b".into()), + Value::Boolean(false), + Value::Boolean(true), + Value::Ref(uuid::Uuid::from_u128(1)), + Value::Ref(uuid::Uuid::from_u128(2)), + kw(":a"), + kw(":b"), + Value::Null, + ]; + for a in &grid { + for b in &grid { + let ab = value_order(a, b); + assert_eq!(ab, value_order(b, a).reverse(), "antisymmetric"); + let same = match (a, b) { + (Value::Float(x), Value::Float(y)) => x.to_bits() == y.to_bits(), + _ => a == b, + }; + assert_eq!(ab == Ordering::Equal, same, "equal only when identical"); + for c in &grid { + if ab != Ordering::Greater && value_order(b, c) != Ordering::Greater { + assert_ne!(value_order(a, c), Ordering::Greater, "transitive"); + } + } + } + } +} + +#[test] +fn value_order_compares_integers_and_floats_exactly() { + // 2^53 + 1 is not a float; as an integer it exceeds the float 2^53. + assert_eq!( + value_order(&i(9_007_199_254_740_993), &f(9_007_199_254_740_992.0)), + Ordering::Greater + ); + // i64::MAX is below 2^63, the float it rounds to. + assert_eq!( + value_order(&i(i64::MAX), &f(9_223_372_036_854_775_808.0)), + Ordering::Less + ); + assert_eq!(value_order(&i(1), &f(1.0)), Ordering::Less); + assert_eq!(value_order(&f(-0.0), &f(0.0)), Ordering::Less); + assert_eq!( + value_order(&f(f64::NAN), &f(f64::INFINITY)), + Ordering::Greater + ); + assert_eq!(value_order(&f(f64::NAN), &i(i64::MAX)), Ordering::Greater); + assert_eq!( + value_order(&Value::String("z".into()), &f(f64::NAN)), + Ordering::Greater + ); +} + +#[test] +fn exact_sum_rounds_once() { + assert_eq!(fsum(&vec![f(0.1); 10]), 1.0); + assert_eq!(fsum(&[f(1e308), f(1e308), f(-1e308)]), 1e308); + assert_eq!(fsum(&[f(1e100), f(1.0), f(-1e100)]), 1.0); + assert_eq!(fsum(&[f(5e-324), f(5e-324)]), 1e-323); + assert_eq!(fsum(&[f(f64::MAX), f(f64::MAX)]), f64::INFINITY); + assert_eq!(fsum(&[f(-f64::MAX), f(-f64::MAX)]), f64::NEG_INFINITY); + assert_eq!(fsum(&[f(f64::MAX), f(f64::MAX), f(-f64::MAX)]), f64::MAX); + // i64::MAX + 0.5 rounds to 2^63. + assert_eq!(fsum(&[i(i64::MAX), f(0.5)]), 9_223_372_036_854_775_808.0); + // Ties to even: 2^53 + 1 → 2^53, 2^53 + 3 → 2^53 + 4. + assert_eq!( + fsum(&[f(9_007_199_254_740_992.0), i(1)]), + 9_007_199_254_740_992.0 + ); + assert_eq!( + fsum(&[f(9_007_199_254_740_992.0), i(3)]), + 9_007_199_254_740_996.0 + ); + assert_eq!(fsum(&[f(0.5), i(-1)]), -0.5); +} + +#[test] +fn exact_sum_signs_of_zero_and_specials() { + assert_eq!(fsum(&[f(-0.0), f(-0.0)]).to_bits(), (-0.0f64).to_bits()); + assert_eq!(fsum(&[f(-0.0), f(0.0)]).to_bits(), 0.0f64.to_bits()); + assert_eq!(fsum(&[f(-0.0), i(0)]).to_bits(), 0.0f64.to_bits()); + assert_eq!(fsum(&[f(1.5), f(-1.5)]).to_bits(), 0.0f64.to_bits()); + assert!(fsum(&[f(f64::NAN), f(1.0)]).is_nan()); + assert!(fsum(&[f(f64::INFINITY), f(f64::NEG_INFINITY)]).is_nan()); + assert_eq!(fsum(&[f(f64::INFINITY), f(-f64::MAX)]), f64::INFINITY); +} + +#[test] +fn exact_sum_is_the_same_in_every_order() { + let vals = [f(1e16), f(1.0), f(-1e16), f(3.5), f(1e-300), f(-2.0)]; + for p in permutations(&vals) { + assert_eq!(fsum(&p).to_bits(), 2.5f64.to_bits()); + } + let mixed = [ + i(i64::MAX), + f(0.25), + i(-7), + f(1e-17), + f(-0.125), + i(i64::MIN), + ]; + let first = fsum(&mixed); + for p in permutations(&mixed) { + assert_eq!(fsum(&p).to_bits(), first.to_bits()); + } +} + +#[test] +fn integer_sum_saturates_once() { + for p in permutations(&[i(i64::MAX), i(1), i(-1)]) { + assert_eq!(finish("sum", &p).ok().flatten(), Some(i(i64::MAX))); + } + assert_eq!( + finish("sum", &[i(i64::MAX), i(i64::MAX)]).ok().flatten(), + Some(i(i64::MAX)) + ); + assert_eq!( + finish("sum", &[i(i64::MIN), i(-1)]).ok().flatten(), + Some(i(i64::MIN)) + ); + assert_eq!(finish("sum", &[]).ok().flatten(), Some(i(0))); + assert_eq!(finish("sum", &[Value::Null]).ok().flatten(), Some(i(0))); +} + +#[test] +fn sum_error_is_decided_at_finish() { + for p in permutations(&[i(1), Value::String("a".into()), i(2)]) { + assert!(code(finish("sum", &p)) == "INT-041"); + } +} + +#[test] +fn min_max_errors_do_not_depend_on_order() { + for p in permutations(&[kw(":a"), kw(":b"), i(1)]) { + assert!(code(finish("min", &p)) == "INT-043"); + } + for p in permutations(&[i(1), f(2.0)]) { + assert!(code(finish("max", &p)) == "INT-043"); + } + assert!(code(finish("min", &[kw(":a")])) == "INT-044"); + assert!(code(finish("max", &[Value::Boolean(true)])) == "INT-044"); + assert!(matches!(finish("min", &[]), Ok(None))); + assert!(matches!(finish("max", &[Value::Null]), Ok(None))); +} + +#[test] +fn min_max_of_floats_use_the_total_order() { + for p in permutations(&[f(0.0), f(-0.0), f(1.0)]) { + let Ok(Some(Value::Float(m))) = finish("min", &p) else { + unreachable!("min of floats") + }; + assert_eq!(m.to_bits(), (-0.0f64).to_bits()); + } + for p in permutations(&[f(0.0), f(-0.0)]) { + let Ok(Some(Value::Float(m))) = finish("max", &p) else { + unreachable!("max of floats") + }; + assert_eq!(m.to_bits(), 0.0f64.to_bits()); + } + for p in permutations(&[f(f64::NAN), f(1.0)]) { + assert!(matches!(finish("max", &p), Ok(Some(Value::Float(m))) if m.is_nan())); + assert_eq!(finish("min", &p).ok().flatten(), Some(f(1.0))); + } + assert_eq!( + finish( + "min", + &[Value::String("b".into()), Value::String("a".into())] + ) + .ok() + .flatten(), + Some(Value::String("a".into())) + ); +} + +#[test] +fn distinct_sets_store_canonical_floats() { + assert_eq!( + finish( + "count-distinct", + &[f(0.0), f(-0.0), f(f64::NAN), f(-f64::NAN), i(0)] + ) + .ok() + .flatten(), + Some(i(3)) + ); + let Ok(Some(Value::Float(s))) = finish("sum-distinct", &[f(-0.0), f(-0.0)]) else { + unreachable!("sum-distinct of floats") + }; + assert_eq!(s.to_bits(), 0.0f64.to_bits()); + assert_eq!( + finish("sum-distinct", &[i(2), i(2), i(3)]).ok().flatten(), + Some(i(5)) + ); + assert!(code(finish("sum-distinct", &[i(2), kw(":k")])) == "INT-041"); + assert_eq!(finish("count-distinct", &[]).ok().flatten(), Some(i(0))); +} + +#[test] +fn count_skips_nil() { + assert_eq!( + finish("count", &[i(1), Value::Null, kw(":k")]) + .ok() + .flatten(), + Some(i(2)) + ); + assert!( + code(finish("avg", &[i(1)])) == "INT-029", + "avg is a window function" + ); +} + +#[test] +fn unknown_aggregate_is_int_029() { + let reg = FunctionRegistry::with_builtins(); + assert!( + Agg::new("nope", ®) + .err() + .is_some_and(|e| crate::MinigrafError::from(e).code() == "INT-029") + ); +} + +fn registry_with_udf() -> FunctionRegistry { + let mut reg = FunctionRegistry::with_builtins(); + reg.register_aggregate_desc( + "total".to_string(), + AggregateDesc { + impl_: AggImpl::Udf(UdfOps { + init: Arc::new(|| Box::new(0i64) as Box), + step: Arc::new(|acc, v| { + if let (Some(n), Value::Integer(x)) = (acc.downcast_mut::(), v) { + *n += x; + } + }), + finalise: Arc::new(|acc, n| { + let total = acc.downcast_ref::().copied().unwrap_or(0); + Value::Integer(total * 100 + i64::try_from(n).unwrap_or(0)) + }), + }), + }, + ) + .expect("registers"); + reg +} + +#[test] +fn registered_aggregate_steps_non_nil_values() { + let reg = registry_with_udf(); + let mut agg = Agg::new("total", ®).expect("registered"); + for v in [i(2), Value::Null, i(3)] { + agg.step(&v); + } + assert_eq!(agg.finish("total").ok().flatten(), Some(i(502))); + let empty = Agg::new("total", ®).expect("registered"); + assert_eq!(empty.finish("total").ok().flatten(), Some(Value::Null)); +} + +fn window(src: &str) -> WindowSpec { + let q = format!("(query [:find {src} :where [?e :x ?v] [?e :y ?o]])"); + match parse_datalog_command(&q) { + Ok(DatalogCommand::Query(q)) => match q.find.into_iter().next() { + Some(FindSpec::Window(ws)) => ws, + _ => unreachable!("a window"), + }, + _ => unreachable!("parses"), + } +} + +#[test] +fn running_values() { + let reg = registry_with_udf(); + let mut sum = Running::new(&window("(sum ?v :over (:order-by ?o))"), ®).expect("sum"); + assert_eq!(sum.value(), i(0)); + sum.step(&i(2)); + assert_eq!(sum.value(), i(2)); + sum.step(&Value::String("skipped".into())); + sum.step(&f(0.5)); + assert_eq!(sum.value(), f(2.5)); + + let mut min = Running::new(&window("(min ?v :over (:order-by ?o))"), ®).expect("min"); + assert_eq!(min.value(), Value::Null); + min.step(&kw(":b")); + min.step(&kw(":a")); + assert_eq!(min.value(), kw(":a")); + + let mut count = Running::new(&window("(count ?v :over (:order-by ?o))"), ®).expect("count"); + count.step(&Value::Null); + count.step(&i(1)); + assert_eq!(count.value(), i(1)); + + let mut avg = Running::new(&window("(avg ?v :over (:order-by ?o))"), ®).expect("avg"); + assert_eq!(avg.value(), Value::Null); + avg.step(&i(1)); + avg.step(&i(2)); + assert_eq!(avg.value(), f(1.5)); + + let mut udf = Running::new(&window("(total ?v :over (:order-by ?o))"), ®).expect("udf"); + udf.step(&Value::Null); + udf.step(&i(4)); + assert_eq!(udf.value(), i(402)); + + assert!( + Running::new(&window("(nope ?v :over (:order-by ?o))"), ®) + .err() + .is_some_and(|e| crate::MinigrafError::from(e).code() == "INT-030") + ); +} + +#[test] +fn named_vars_in_find_order() { + let parse = |src: &str| match parse_datalog_command(src) { + Ok(DatalogCommand::Query(q)) => q.find, + _ => unreachable!("parses"), + }; + let find = parse( + "(query [:find ?e (sum ?v :over (:partition-by ?p :order-by ?o)) ?v \ + :where [?e :v ?v] [?e :p ?p] [?e :o ?o]])", + ); + assert_eq!(named_vars(&find), vec!["?e", "?v", "?p", "?o"]); + let find = parse( + "(query [:find ?d (count ?e) (rank :over (:order-by ?d)) \ + :where [?e :d ?d]])", + ); + assert_eq!(named_vars(&find), vec!["?d"]); +} diff --git a/src/query/datalog/executor.rs b/src/query/datalog/executor.rs index 63ad0554d..5eb6ea218 100644 --- a/src/query/datalog/executor.rs +++ b/src/query/datalog/executor.rs @@ -1,5 +1,6 @@ +use super::aggregate::{Agg, Running, named_vars, value_order}; use super::evaluator::{StratifiedEvaluator, evaluate_not_join}; -use super::functions::{AggImpl, FunctionRegistry, apply_builtin_aggregate, value_cmp}; +use super::functions::FunctionRegistry; use super::matcher::{PatternMatcher, edn_to_entity_id, edn_to_value}; use super::optimizer; use super::rules::RuleRegistry; @@ -1610,6 +1611,11 @@ fn apply_post_processing( /// Group bindings by non-aggregate find vars + with_vars, apply aggregate functions, /// return one binding map per group. Aggregate results stored under `"__agg_{i}"`. +/// +/// Each group holds one accumulator per aggregate (#547 spec §3.2), so the +/// result does not depend on the order of the bindings. Groups finish in +/// key order; the first aggregate in `:find` order whose result is an error +/// fails the query, except `INT-042`, which drops the group (spec §3.3). fn compute_aggregation( bindings: Vec, find_specs: &[FindSpec], @@ -1638,11 +1644,9 @@ fn compute_aggregation( } // In a mixed aggregate+window query, :with vars must NOT be added to the - // grouping key. The window phase runs after aggregation, so :with vars that - // are used only by window specs (var, order_by) would otherwise inflate the - // number of groups. Even :with vars used by aggregate specs (e.g. ?e in - // count(?e)) should not split groups — the aggregate operates over all rows - // in the base group determined by the Variable find specs. + // grouping key. The window phase runs after aggregation and reads only + // the grouping variables (PRS-082), so :with vars would only inflate the + // number of groups. let has_windows = find_specs.iter().any(|s| matches!(s, FindSpec::Window(_))); // Grouping key = Variable find specs (in find order). @@ -1656,238 +1660,186 @@ fn compute_aggregation( }) .collect(); if !has_windows { - // Pure aggregate: with_vars add to grouping key. group_var_names.extend(with_vars.iter().map(|s| s.as_str())); } - // Group using BTreeMap keyed by group key (O(log g) instead of O(g) per binding). + // The aggregates of :find: their position, function and argument. + let aggs: Vec<(usize, &str, &str)> = find_specs + .iter() + .enumerate() + .filter_map(|(i, s)| match s { + FindSpec::Aggregate { func, var } => Some((i, func.as_str(), var.as_str())), + _ => None, + }) + .collect(); + + // Groups in key order, as `Value`'s `Ord` compares keys. use std::collections::BTreeMap; - let mut groups: BTreeMap, Vec> = BTreeMap::new(); + let mut groups: BTreeMap, Vec> = BTreeMap::new(); for b in bindings { let key: Vec = group_var_names .iter() .map(|v| b.get(*v).cloned().unwrap_or(Value::Null)) .collect(); - groups.entry(key).or_default().push(b); - } - - // Build a position map for Variable specs only (indices 0..n_vars in the key vector). - // with_vars occupy key positions n_vars..end and are used only for grouping, not for output. - // Map of Variable spec name → its index in the group key Vec. - let mut group_key_idx: std::collections::HashMap<&str, usize> = - std::collections::HashMap::new(); - { - let mut var_pos = 0usize; - for spec in find_specs { - if let FindSpec::Variable(v) = spec { - group_key_idx.insert(v.as_str(), var_pos); - var_pos += 1; + let accs = match groups.entry(key) { + std::collections::btree_map::Entry::Occupied(e) => e.into_mut(), + std::collections::btree_map::Entry::Vacant(e) => e.insert( + aggs.iter() + .map(|(_, func, _)| Agg::new(func, registry)) + .collect::>>()?, + ), + }; + for (acc, (_, _, var)) in accs.iter_mut().zip(&aggs) { + if let Some(v) = b.get(*var) { + acc.step(v); } } } + // The plain :find variables occupy the first key positions; with_vars + // follow and only group. + let plain: Vec<&str> = find_specs + .iter() + .filter_map(|s| match s { + FindSpec::Variable(v) => Some(v.as_str()), + _ => None, + }) + .collect(); + let mut results: Vec = Vec::new(); - for (key, group_bindings) in groups.iter() { + 'group: for (key, accs) in groups { let mut binding = Binding::new(); - let mut skip = false; - - // Plain variable values from group key. - for (v, &idx) in &group_key_idx { - if let Some(val) = key.get(idx) { - binding.insert((*v).to_string(), val.clone()); - } + for (v, val) in plain.iter().zip(key) { + binding.insert((*v).to_string(), val); } - - // Aggregate values stored under "__agg_{i}". - for (i, spec) in find_specs.iter().enumerate() { - if let FindSpec::Aggregate { func, var } = spec { - let non_null: Vec<&Value> = group_bindings - .iter() - .filter_map(|b| b.get(var.as_str())) - .filter(|v| !matches!(v, Value::Null)) - .collect(); - let agg_val: anyhow::Result = match registry.get(func.as_str()) { - Some(desc) if desc.is_builtin => { - // Built-in: use batch path which enforces strict type-error semantics. - apply_builtin_aggregate(func, &non_null) - } - Some(desc) => { - if let AggImpl::Udf(ops) = &desc.impl_ { - if non_null.is_empty() { - Ok(Value::Null) - } else { - let mut acc = (ops.init)(); - for v in &non_null { - (ops.step)(&mut acc, v); - } - Ok((ops.finalise)(&acc, non_null.len())) - } - } else { - // AggImpl::Builtin with is_builtin=false shouldn't happen - apply_builtin_aggregate(func, &non_null) - } - } - None => Err(err_coded!(ErrorCode::Int029, func)), - }; - match agg_val { - Ok(v) => { - binding.insert(format!("__agg_{}", i), v); - } - Err(e) => { - let msg = e.to_string(); - if msg.contains("no non-null values in group") { - skip = true; - break; - } - return Err(e); - } + for (acc, (i, func, _)) in accs.into_iter().zip(&aggs) { + match acc.finish(func)? { + Some(v) => { + binding.insert(format!("__agg_{}", i), v); } + None => continue 'group, } } - - if !skip { - results.push(binding); - } + results.push(binding); } Ok(results) } /// Compute window function values for each row and store under `"__win_{i}"`. -/// Modifies `bindings` in place. +/// Modifies `bindings` in place; rows keep their order. +/// +/// Each partition (by `Value`'s `Ord`) is sorted by its `:order-by` value +/// under [`value_order`], then by the named variables, then by input +/// position (#547 spec §4.1). Rows with order-equal `:order-by` values are +/// peers: `rank` is equal for them, and a running aggregate gives every +/// peer its value at the last peer (spec §4.2). fn apply_window_functions( bindings: &mut [Binding], find_specs: &[FindSpec], registry: &FunctionRegistry, ) -> Result<()> { + let named = named_vars(find_specs); + let read = |b: &Binding, var: &str| b.get(var).cloned().unwrap_or(Value::Null); + // Each row's named-variable values, for the tie-break. + let tiebreak: Vec> = bindings + .iter() + .map(|b| named.iter().map(|v| read(b, v)).collect()) + .collect(); + for (i, spec) in find_specs.iter().enumerate() { let FindSpec::Window(ws) = spec else { continue; }; let key = format!("__win_{}", i); - // Build partitions: (partition_key, sorted row indices). - let mut partitions: HashMap, Vec> = HashMap::new(); - for (row_idx, binding) in bindings.iter().enumerate() { - let part_key = ws + let mut partitions: std::collections::BTreeMap> = + std::collections::BTreeMap::new(); + for (row, binding) in bindings.iter().enumerate() { + let part = ws .partition_by .as_ref() - .and_then(|pv| binding.get(pv)) - .cloned(); - partitions.entry(part_key).or_default().push(row_idx); + .map_or(Value::Null, |pv| read(binding, pv)); + partitions.entry(part).or_default().push(row); } - // For each partition: sort, compute window values, write back. - for row_indices in partitions.values_mut() { - // Pre-extract order_by values into a contiguous Vec so the sort - // comparator never touches the HashMap — O(n) lookups here instead - // of O(n log n) random HashMap accesses inside sort_by. - // Safety: row_indices are populated from 0..bindings.len() enumeration above, - // so all indices are valid. - #[allow(clippy::indexing_slicing)] - let mut keyed: Vec<(Value, usize)> = row_indices - .iter() - .map(|&i| { - let k = bindings[i] - .get(&ws.order_by) - .cloned() - .unwrap_or(Value::Null); - (k, i) - }) + let mut out: Vec<(usize, Value)> = Vec::with_capacity(bindings.len()); + for rows in partitions.into_values() { + let mut keyed: Vec<(Value, usize)> = rows + .into_iter() + .filter_map(|r| bindings.get(r).map(|b| (read(b, &ws.order_by), r))) .collect(); - keyed.sort_by(|(a, _), (b, _)| { - let cmp = value_cmp(a, b); - match ws.order { - Order::Asc => cmp, - Order::Desc => cmp.reverse(), - } + keyed.sort_by(|(a, ra), (b, rb)| { + let by_key = match ws.order { + Order::Asc => value_order(a, b), + Order::Desc => value_order(b, a), + }; + by_key + .then_with(|| { + let (ta, tb) = (tiebreak.get(*ra), tiebreak.get(*rb)); + ta.into_iter() + .flatten() + .zip(tb.into_iter().flatten()) + .map(|(x, y)| value_order(x, y)) + .find(|o| o.is_ne()) + .unwrap_or(std::cmp::Ordering::Equal) + }) + .then(ra.cmp(rb)) }); - // Rewrite row_indices in sorted order for the write-back step. - for (dest, (_, src)) in row_indices.iter_mut().zip(keyed.iter()) { - *dest = *src; - } - // Compute one window value per row in sorted order. - let window_values: Vec = match ws.func { - WindowFunc::RowNumber => { - let mut values = Vec::with_capacity(keyed.len()); - for pos in 1..=keyed.len() { - values.push(Value::Integer(i64::try_from(pos).map_err(|_| { - err_coded!(ErrorCode::Int032, "row number overflow") - })?)); - } - values - } - - WindowFunc::Rank => { - // Reuse pre-extracted keys for tie-detection — no extra HashMap lookups. - let mut values = Vec::with_capacity(keyed.len()); - let mut rank = 1i64; - let mut prev: Option<&Value> = None; - for (row_num, (key, _)) in (1i64..).zip(keyed.iter()) { - if prev != Some(key) { - rank = row_num; - prev = Some(key); - } - values.push(Value::Integer(rank)); + let mut running = match ws.func { + WindowFunc::Rank | WindowFunc::RowNumber => None, + _ => Some(Running::new(ws, registry)?), + }; + let mut start = 0; + while start < keyed.len() { + let peers_end = keyed + .iter() + .skip(start + 1) + .position(|(k, _)| { + keyed + .get(start) + .is_none_or(|(s, _)| value_order(k, s).is_ne()) + }) + .map_or(keyed.len(), |n| start + 1 + n); + let peers = keyed.get(start..peers_end).unwrap_or(&[]); + if let Some(acc) = running.as_mut() { + for (_, r) in peers { + let v = ws + .var + .as_ref() + .and_then(|var| bindings.get(*r).and_then(|b| b.get(var))) + .unwrap_or(&Value::Null); + acc.step(v); } - values } - - _ => { - // Accumulator-based: built-ins (sum, count, min, max, avg) and UDF aggregates. - // UDF aggregates (WindowFunc::Udf) also route here via registry lookup. - let func_name = ws.func_name(); - let desc = registry - .get(&func_name) - .ok_or_else(|| err_coded!(ErrorCode::Int030, func_name.clone()))?; - - let mut values = Vec::with_capacity(keyed.len()); - // Safety: row_idx values come from enumerate() over bindings, so indices are valid. - #[allow(clippy::indexing_slicing)] - match &desc.impl_ { - AggImpl::Builtin(ops) => { - let mut acc = (ops.init)(); - for (_, row_idx) in keyed.iter() { - let val = ws - .var - .as_ref() - .and_then(|v| bindings[*row_idx].get(v)) - .unwrap_or(&Value::Null); - (ops.step)(&mut acc, val); - values.push((ops.finalise)(&acc)); - } - } - AggImpl::Udf(ops) => { - let mut acc = (ops.init)(); - let mut row_count = 0usize; - for (_, row_idx) in keyed.iter() { - let val = ws - .var - .as_ref() - .and_then(|v| bindings[*row_idx].get(v)) - .unwrap_or(&Value::Null); - (ops.step)(&mut acc, val); - row_count += 1; - values.push((ops.finalise)(&acc, row_count)); - } - } - } - values + for (pos, (_, r)) in (start..peers_end).zip(peers) { + let value = match (&running, &ws.func) { + (Some(acc), _) => acc.value(), + (None, WindowFunc::RowNumber) => position(pos)?, + (None, _) => position(start)?, + }; + out.push((*r, value)); } - }; - - // Write window values back to rows. - // Safety: row_idx values come from enumerate() over bindings, so indices are valid. - for (&row_idx, window_val) in row_indices.iter().zip(window_values) { - #[allow(clippy::indexing_slicing)] - bindings[row_idx].insert(key.clone(), window_val); + start = peers_end; + } + } + for (r, value) in out { + if let Some(b) = bindings.get_mut(r) { + b.insert(key.clone(), value); } } } Ok(()) } +/// The 1-based position `pos + 1` as an integer. +fn position(pos: usize) -> Result { + i64::try_from(pos + 1) + .map(Value::Integer) + .map_err(|_| err_coded!(ErrorCode::Int032, "row number overflow")) +} + /// Project binding maps to output rows in find-spec order. fn project_find_specs(bindings: &[Binding], find_specs: &[FindSpec]) -> Vec> { let mut results = Vec::new(); @@ -2513,9 +2465,9 @@ fn eval_binop(op: &BinOp, l: Value, r: Value) -> Result { } match op { - // Ordering comparisons. Strings order lexicographically, matching value_lt / - // value_cmp — which min/max and :order-by already use — so the same two strings - // order the same way whether compared by a predicate or by an aggregate. + // Ordering comparisons. Strings order lexicographically, matching + // aggregate::value_order — which min/max and :order-by use — so the same two + // strings order the same way whether compared by a predicate or by an aggregate. // Numbers order numerically with int/float promotion via to_float_pair. // Mixed types (string vs number) remain a type error. BinOp::Lt | BinOp::Gt | BinOp::Lte | BinOp::Gte => { diff --git a/src/query/datalog/functions.rs b/src/query/datalog/functions.rs index 7f98c6983..c62fb9f2a 100644 --- a/src/query/datalog/functions.rs +++ b/src/query/datalog/functions.rs @@ -1,32 +1,19 @@ -use crate::error::{ErrorCode, bail_coded, err_coded}; +use crate::error::{ErrorCode, bail_coded}; use crate::graph::types::Value; use std::any::Any; use std::collections::HashMap; use std::sync::Arc; -/// Accumulator state for window aggregate functions (incremental computation). -#[derive(Clone, Debug)] -pub enum AggState { - Sum { - total: f64, - is_float: bool, - }, - Count(i64), - /// Used by both min and max; semantics determined by the registered step fn. - MinMax { - current: Option, - }, - Avg { - sum: f64, - count: usize, - }, -} - -/// Incremental operations for window-compatible aggregate functions. -pub struct WindowOps { - pub init: fn() -> AggState, - pub step: fn(&mut AggState, &Value), - pub finalise: fn(&AggState) -> Value, +/// A built-in aggregate; `aggregate.rs` holds its arithmetic (#547). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Builtin { + Count, + Sum, + Min, + Max, + Avg, + CountDistinct, + SumDistinct, } /// Type alias for the type-erased UDF accumulator step closure. @@ -45,8 +32,8 @@ pub struct UdfOps { /// Implementation discriminator for aggregate functions. pub enum AggImpl { - /// Built-in: uses `fn()` function pointers and `AggState`. - Builtin(WindowOps), + /// Built-in: computed by `aggregate::Agg` and `aggregate::Running`. + Builtin(Builtin), /// User-defined: uses `Arc` closures and `Box`. Udf(UdfOps), } @@ -60,8 +47,6 @@ pub struct PredicateDesc { /// Descriptor for one registered aggregate function. pub struct AggregateDesc { pub impl_: AggImpl, - /// True for built-in functions handled by `apply_builtin_aggregate`. - pub is_builtin: bool, } /// Registry of aggregate function descriptors, keyed by hyphenated name. @@ -81,197 +66,22 @@ impl FunctionRegistry { predicates: HashMap::new(), }; - // count (window-compatible) - reg.aggregates.insert( - "count".into(), - AggregateDesc { - is_builtin: true, - impl_: AggImpl::Builtin(WindowOps { - init: || AggState::Count(0), - step: |state, v| { - if !matches!(v, Value::Null) { - let AggState::Count(n) = state else { return }; - *n = n.saturating_add(1); - } - }, - finalise: |state| { - if let AggState::Count(n) = state { - Value::Integer(*n) - } else { - Value::Null - } - }, - }), - }, - ); - - // sum (window-compatible) - reg.aggregates.insert( - "sum".into(), - AggregateDesc { - is_builtin: true, - impl_: AggImpl::Builtin(WindowOps { - init: || AggState::Sum { - total: 0.0, - is_float: false, - }, - step: |state, v| { - if let AggState::Sum { total, is_float } = state { - match v { - Value::Integer(i) => *total += *i as f64, - Value::Float(f) => { - *total += f; - *is_float = true; - } - _ => {} - } - } - }, - finalise: |state| { - if let AggState::Sum { total, is_float } = state { - if *is_float { - Value::Float(*total) - } else { - #[allow(clippy::cast_possible_truncation)] - // value was accumulated from integers, range checked by is_float guard - Value::Integer(*total as i64) - } - } else { - Value::Null - } - }, - }), - }, - ); - - // min (window-compatible) - reg.aggregates.insert( - "min".into(), - AggregateDesc { - is_builtin: true, - impl_: AggImpl::Builtin(WindowOps { - init: || AggState::MinMax { current: None }, - step: |state, v| { - if matches!(v, Value::Null) { - return; - } - if let AggState::MinMax { current } = state { - match current { - None => *current = Some(v.clone()), - Some(cur) => { - if value_lt(v, cur) { - *current = Some(v.clone()); - } - } - } - } - }, - finalise: |state| { - if let AggState::MinMax { current } = state { - current.clone().unwrap_or(Value::Null) - } else { - Value::Null - } - }, - }), - }, - ); - - // max (window-compatible) - reg.aggregates.insert( - "max".into(), - AggregateDesc { - is_builtin: true, - impl_: AggImpl::Builtin(WindowOps { - init: || AggState::MinMax { current: None }, - step: |state, v| { - if matches!(v, Value::Null) { - return; - } - if let AggState::MinMax { current } = state { - match current { - None => *current = Some(v.clone()), - Some(cur) => { - if value_lt(cur, v) { - *current = Some(v.clone()); - } - } - } - } - }, - finalise: |state| { - if let AggState::MinMax { current } = state { - current.clone().unwrap_or(Value::Null) - } else { - Value::Null - } - }, - }), - }, - ); - - // avg (window-compatible) - reg.aggregates.insert( - "avg".into(), - AggregateDesc { - is_builtin: true, - impl_: AggImpl::Builtin(WindowOps { - init: || AggState::Avg { sum: 0.0, count: 0 }, - step: |state, v| { - if let AggState::Avg { sum, count } = state { - match v { - Value::Integer(i) => { - *sum += *i as f64; - *count = count.saturating_add(1); - } - Value::Float(f) => { - *sum += f; - *count = count.saturating_add(1); - } - _ => {} - } - } - }, - finalise: |state| { - if let AggState::Avg { sum, count } = state { - if *count == 0 { - Value::Null - } else { - Value::Float(*sum / *count as f64) - } - } else { - Value::Null - } - }, - }), - }, - ); - - // count-distinct (NOT window-compatible) — stub WindowOps; never called in window position - reg.aggregates.insert( - "count-distinct".into(), - AggregateDesc { - is_builtin: true, - impl_: AggImpl::Builtin(WindowOps { - init: || AggState::Count(0), - step: |_state, _v| {}, - finalise: |_state| Value::Null, - }), - }, - ); - - // sum-distinct (NOT window-compatible) — stub WindowOps; never called in window position - reg.aggregates.insert( - "sum-distinct".into(), - AggregateDesc { - is_builtin: true, - impl_: AggImpl::Builtin(WindowOps { - init: || AggState::Count(0), - step: |_state, _v| {}, - finalise: |_state| Value::Null, - }), - }, - ); + for (name, b) in [ + ("count", Builtin::Count), + ("sum", Builtin::Sum), + ("min", Builtin::Min), + ("max", Builtin::Max), + ("avg", Builtin::Avg), + ("count-distinct", Builtin::CountDistinct), + ("sum-distinct", Builtin::SumDistinct), + ] { + reg.aggregates.insert( + name.into(), + AggregateDesc { + impl_: AggImpl::Builtin(b), + }, + ); + } // Built-in predicate name sentinels — block user registration of these names. for name in [ @@ -301,21 +111,6 @@ impl FunctionRegistry { self.aggregates.get(name) } - #[allow(dead_code)] - pub fn is_known(&self, name: &str) -> bool { - self.aggregates.contains_key(name) - } - - /// count-distinct and sum-distinct are not window-compatible. - /// All other registered aggregates (built-in or UDF) are window-compatible. - #[allow(dead_code)] - pub fn is_window_compatible(&self, name: &str) -> bool { - match name { - "count-distinct" | "sum-distinct" => false, - _ => self.aggregates.contains_key(name), - } - } - /// Register a UDF aggregate descriptor. Returns Err if the name is already taken. pub fn register_aggregate_desc( &mut self, @@ -349,174 +144,6 @@ impl FunctionRegistry { } } -/// Returns true if a < b by value ordering. Mixed Integer/Float promotes to f64. -/// Returns false for incomparable types (no panic). -pub(crate) fn value_lt(a: &Value, b: &Value) -> bool { - match (a, b) { - (Value::Integer(x), Value::Integer(y)) => x < y, - (Value::Float(x), Value::Float(y)) => x < y, - (Value::Integer(x), Value::Float(y)) => (*x as f64) < *y, - (Value::Float(x), Value::Integer(y)) => *x < (*y as f64), - (Value::String(x), Value::String(y)) => x < y, - _ => false, - } -} - -/// Return the ordering between two `Value`s in a single pass. -/// Matches `value_lt`'s type-crossing rules; incomparable types are Equal. -pub(crate) fn value_cmp(a: &Value, b: &Value) -> std::cmp::Ordering { - match (a, b) { - (Value::Integer(x), Value::Integer(y)) => x.cmp(y), - (Value::Float(x), Value::Float(y)) => x.partial_cmp(y).unwrap_or(std::cmp::Ordering::Equal), - (Value::Integer(x), Value::Float(y)) => (*x as f64) - .partial_cmp(y) - .unwrap_or(std::cmp::Ordering::Equal), - (Value::Float(x), Value::Integer(y)) => x - .partial_cmp(&(*y as f64)) - .unwrap_or(std::cmp::Ordering::Equal), - (Value::String(x), Value::String(y)) => x.cmp(y), - _ => std::cmp::Ordering::Equal, - } -} - -/// Human-readable type name for error messages. -pub(crate) fn value_type_name(v: &Value) -> &'static str { - match v { - Value::String(_) => "String", - Value::Integer(_) => "Integer", - Value::Float(_) => "Float", - Value::Boolean(_) => "Boolean", - Value::Ref(_) => "Ref", - Value::Keyword(_) => "Keyword", - Value::Null => "Null", - } -} - -/// Apply a built-in aggregate function to a slice of non-null values (batch mode). -pub fn apply_builtin_aggregate(name: &str, values: &[&Value]) -> anyhow::Result { - match name { - "count" => Ok(Value::Integer( - i64::try_from(values.len()).unwrap_or(i64::MAX), - )), - - "count-distinct" => { - // O(n log n) via BTreeSet — Value::Ord is a stable discriminant-based - // total order so BTreeSet membership is correct. - let seen: std::collections::BTreeSet<&Value> = values.iter().copied().collect(); - Ok(Value::Integer( - i64::try_from(seen.len()).unwrap_or(i64::MAX), - )) - } - - "sum" | "sum-distinct" => { - let deduped: Vec<&Value> = if name == "sum-distinct" { - // O(n log n) via BTreeSet (same rationale as count-distinct). - let seen: std::collections::BTreeSet<&Value> = values.iter().copied().collect(); - seen.into_iter().collect() - } else { - values.to_vec() - }; - - if deduped.is_empty() { - return Ok(Value::Integer(0)); - } - - let has_float = deduped.iter().any(|v| matches!(v, Value::Float(_))); - if has_float { - let mut sum = 0.0_f64; - for v in &deduped { - match v { - Value::Float(f) => sum += f, - Value::Integer(i) => sum += *i as f64, - other => { - return Err(err_coded!(ErrorCode::Int041, value_type_name(other))); - } - } - } - Ok(Value::Float(sum)) - } else { - let mut sum = 0_i64; - for v in &deduped { - match v { - Value::Integer(i) => sum = sum.saturating_add(*i), - other => { - return Err(err_coded!(ErrorCode::Int041, value_type_name(other))); - } - } - } - Ok(Value::Integer(sum)) - } - } - - "min" | "max" => { - if values.is_empty() { - return Err(err_coded!(ErrorCode::Int042)); - } - let first = values - .first() - .copied() - .ok_or_else(|| err_coded!(ErrorCode::Int042))?; - for v in values.get(1..).unwrap_or(&[]) { - if std::mem::discriminant(*v) != std::mem::discriminant(first) { - return Err(err_coded!( - ErrorCode::Int043, - name, - value_type_name(first), - value_type_name(v) - )); - } - } - let result = values.iter().try_fold((*first).clone(), |acc, v| { - let ordering = match (&acc, v) { - (Value::Integer(a), Value::Integer(b)) => a.cmp(b), - (Value::Float(a), Value::Float(b)) => { - a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal) - } - (Value::String(a), Value::String(b)) => a.cmp(b), - (_, other) => { - return Err(err_coded!(ErrorCode::Int044, name, value_type_name(other))); - } - }; - let replace = if name == "min" { - ordering == std::cmp::Ordering::Greater - } else { - ordering == std::cmp::Ordering::Less - }; - Ok::(if replace { (*v).clone() } else { acc }) - })?; - Ok(result) - } - - "avg" => { - if values.is_empty() { - return Ok(Value::Null); - } - let mut sum = 0.0_f64; - let mut count = 0usize; - for v in values { - match v { - Value::Integer(i) => { - sum += *i as f64; - count = count.saturating_add(1); - } - Value::Float(f) => { - sum += f; - count = count.saturating_add(1); - } - _ => {} - } - } - if count == 0 { - Ok(Value::Null) - } else { - Ok(Value::Float(sum / count as f64)) - } - } - - other => Err(err_coded!(ErrorCode::Int029, other)), - } -} - #[cfg(test)] mod tests { use super::*; @@ -534,24 +161,9 @@ mod tests { "max", "avg", ] { - assert!(reg.is_known(name), "expected '{}' to be registered", name); - } - } - - #[test] - fn window_compatible_flags() { - let reg = FunctionRegistry::with_builtins(); - for name in ["count", "sum", "min", "max", "avg"] { assert!( - reg.is_window_compatible(name), - "'{}' should be window-compatible", - name - ); - } - for name in ["count-distinct", "sum-distinct"] { - assert!( - !reg.is_window_compatible(name), - "'{}' should NOT be window-compatible", + reg.get(name).is_some(), + "expected '{}' to be registered", name ); } @@ -561,134 +173,6 @@ mod tests { fn unknown_name_returns_none() { let reg = FunctionRegistry::with_builtins(); assert!(reg.get("nonexistent").is_none()); - assert!(!reg.is_known("nonexistent")); - } - - #[test] - fn apply_builtin_count() { - let vals = [Value::Integer(1), Value::Integer(2), Value::Integer(3)]; - let refs: Vec<&Value> = vals.iter().collect(); - let result = apply_builtin_aggregate("count", &refs).unwrap(); - assert_eq!(result, Value::Integer(3)); - } - - #[test] - fn apply_builtin_sum_integers() { - let vals = [Value::Integer(10), Value::Integer(20), Value::Integer(30)]; - let refs: Vec<&Value> = vals.iter().collect(); - let result = apply_builtin_aggregate("sum", &refs).unwrap(); - assert_eq!(result, Value::Integer(60)); - } - - #[test] - fn apply_builtin_sum_floats() { - let vals = [Value::Float(1.5), Value::Float(2.5)]; - let refs: Vec<&Value> = vals.iter().collect(); - let result = apply_builtin_aggregate("sum", &refs).unwrap(); - assert_eq!(result, Value::Float(4.0)); - } - - #[test] - fn apply_builtin_min() { - let vals = [Value::Integer(30), Value::Integer(10), Value::Integer(20)]; - let refs: Vec<&Value> = vals.iter().collect(); - let result = apply_builtin_aggregate("min", &refs).unwrap(); - assert_eq!(result, Value::Integer(10)); - } - - #[test] - fn apply_builtin_max() { - let vals = [Value::Integer(30), Value::Integer(10), Value::Integer(20)]; - let refs: Vec<&Value> = vals.iter().collect(); - let result = apply_builtin_aggregate("max", &refs).unwrap(); - assert_eq!(result, Value::Integer(30)); - } - - #[test] - fn apply_builtin_avg() { - let vals = [Value::Integer(10), Value::Integer(20), Value::Integer(30)]; - let refs: Vec<&Value> = vals.iter().collect(); - let result = apply_builtin_aggregate("avg", &refs).unwrap(); - assert_eq!(result, Value::Float(20.0)); - } - - #[test] - fn apply_builtin_count_distinct() { - let vals = [Value::Integer(1), Value::Integer(2), Value::Integer(1)]; - let refs: Vec<&Value> = vals.iter().collect(); - let result = apply_builtin_aggregate("count-distinct", &refs).unwrap(); - assert_eq!(result, Value::Integer(2)); - } - - #[test] - fn apply_builtin_min_empty_errors() { - let result = apply_builtin_aggregate("min", &[]); - assert!(result.is_err()); - assert!( - result - .unwrap_err() - .to_string() - .contains("no non-null values") - ); - } - - #[test] - fn value_lt_integers() { - assert!(value_lt(&Value::Integer(1), &Value::Integer(2))); - assert!(!value_lt(&Value::Integer(2), &Value::Integer(1))); - assert!(!value_lt(&Value::Integer(1), &Value::Integer(1))); - } - - #[test] - fn value_lt_strings() { - assert!(value_lt( - &Value::String("a".into()), - &Value::String("b".into()) - )); - assert!(!value_lt( - &Value::String("b".into()), - &Value::String("a".into()) - )); - } - - #[test] - fn window_ops_sum_accumulator() { - let reg = FunctionRegistry::with_builtins(); - let desc = reg.get("sum").unwrap(); - let AggImpl::Builtin(ops) = &desc.impl_ else { - panic!("sum should be Builtin") - }; - let mut state = (ops.init)(); - (ops.step)(&mut state, &Value::Integer(10)); - assert_eq!((ops.finalise)(&state), Value::Integer(10)); - (ops.step)(&mut state, &Value::Integer(20)); - assert_eq!((ops.finalise)(&state), Value::Integer(30)); - } - - #[test] - fn window_ops_count_accumulator() { - let reg = FunctionRegistry::with_builtins(); - let desc = reg.get("count").unwrap(); - let AggImpl::Builtin(ops) = &desc.impl_ else { - panic!("count should be Builtin") - }; - let mut state = (ops.init)(); - (ops.step)(&mut state, &Value::Integer(1)); - (ops.step)(&mut state, &Value::Integer(2)); - assert_eq!((ops.finalise)(&state), Value::Integer(2)); - } - - #[test] - fn window_ops_avg_accumulator() { - let reg = FunctionRegistry::with_builtins(); - let desc = reg.get("avg").unwrap(); - let AggImpl::Builtin(ops) = &desc.impl_ else { - panic!("avg should be Builtin") - }; - let mut state = (ops.init)(); - (ops.step)(&mut state, &Value::Integer(10)); - (ops.step)(&mut state, &Value::Integer(20)); - assert_eq!((ops.finalise)(&state), Value::Float(15.0)); } #[test] @@ -710,12 +194,10 @@ mod tests { .unwrap_or(Value::Null) }), }), - is_builtin: false, }, ) .expect("register should succeed"); - assert!(reg.is_known("myfn")); - assert!(reg.is_window_compatible("myfn")); + assert!(reg.get("myfn").is_some()); } #[test] @@ -727,7 +209,6 @@ mod tests { step: Arc::new(|_acc, _v| {}), finalise: Arc::new(|_acc, _n| Value::Null), }), - is_builtin: false, }; reg.register_aggregate_desc("myfn".to_string(), make_desc()) .expect("first ok"); @@ -748,7 +229,6 @@ mod tests { step: Arc::new(|_acc, _v| {}), finalise: Arc::new(|_acc, _n| Value::Null), }), - is_builtin: false, }, ); assert!(result.is_err()); @@ -788,18 +268,4 @@ mod tests { ); assert!(result.is_err()); } - - #[test] - fn builtin_sum_accumulator_regression_guard() { - let reg = FunctionRegistry::with_builtins(); - // Regression guard: existing window_ops path still works. - let desc = reg.get("sum").unwrap(); - if let AggImpl::Builtin(ops) = &desc.impl_ { - let mut acc = (ops.init)(); - (ops.step)(&mut acc, &Value::Integer(5)); - assert_eq!((ops.finalise)(&acc), Value::Integer(5)); - } else { - panic!("sum should be Builtin"); - } - } } diff --git a/src/query/datalog/mod.rs b/src/query/datalog/mod.rs index 642f49c13..fa3d49f6d 100644 --- a/src/query/datalog/mod.rs +++ b/src/query/datalog/mod.rs @@ -1,3 +1,4 @@ +pub(crate) mod aggregate; pub mod evaluator; pub mod executor; pub mod functions; diff --git a/src/query/datalog/parser.rs b/src/query/datalog/parser.rs index 833faf20e..bc49c1874 100644 --- a/src/query/datalog/parser.rs +++ b/src/query/datalog/parser.rs @@ -950,6 +950,34 @@ fn parse_query(elements: &[EdnValue]) -> Result { } } } + // In a query with aggregates, windows run over the groups, which hold + // only the plain :find variables (#547 spec §4.4). + if find_specs + .iter() + .any(|s| matches!(s, FindSpec::Aggregate { .. })) + { + let plain: Vec<&String> = find_specs + .iter() + .filter_map(|s| match s { + FindSpec::Variable(v) => Some(v), + _ => None, + }) + .collect(); + for spec in &find_specs { + if let FindSpec::Window(ws) = spec { + let vars = ws + .var + .iter() + .chain(ws.partition_by.iter()) + .chain(std::iter::once(&ws.order_by)); + for var in vars { + if !plain.contains(&var) { + bail_coded!(ErrorCode::Prs082, var, spec.display_name()); + } + } + } + } + } for var in &with_vars { if !outer_bound.contains(var) { bail_coded!(ErrorCode::Prs022, var); @@ -3934,19 +3962,39 @@ mod window_parse_tests { #[test] fn parse_mixed_aggregate_and_window_allowed() { let cmd = parse_datalog_command( - r#"(query [:find (count ?e) (sum ?v :over (:order-by ?v)) + r#"(query [:find ?v (count ?e) (sum ?v :over (:order-by ?v)) :where [?e :x ?v]])"#, ); assert!(cmd.is_ok(), "parse failed"); if let Ok(DatalogCommand::Query(q)) = cmd { - assert_eq!(q.find.len(), 2); - assert!(matches!(&q.find[0], FindSpec::Aggregate { func, .. } if func == "count")); - assert!(matches!(&q.find[1], FindSpec::Window(ws) if ws.func == WindowFunc::Sum)); + assert_eq!(q.find.len(), 3); + assert!(matches!(&q.find[1], FindSpec::Aggregate { func, .. } if func == "count")); + assert!(matches!(&q.find[2], FindSpec::Window(ws) if ws.func == WindowFunc::Sum)); } else { panic!("expected Query"); } } + #[test] + fn window_in_aggregate_query_reads_only_plain_find_variables() { + // PRS-082 (#547 spec §4.4): the argument, :partition-by and :order-by + // of a window in an aggregate query must be plain :find variables; + // :with variables do not count. + for q in [ + r#"(query [:find (count ?e) (sum ?v :over (:order-by ?v)) :where [?e :x ?v]])"#, + r#"(query [:find ?d (count ?e) (rank :over (:order-by ?v)) :where [?e :x ?v] [?e :d ?d]])"#, + r#"(query [:find ?d (count ?e) (sum ?d :over (:partition-by ?v :order-by ?d)) :where [?e :x ?v] [?e :d ?d]])"#, + r#"(query [:find ?d (count ?e) (sum ?v :over (:order-by ?d)) :with ?v :where [?e :x ?v] [?e :d ?d]])"#, + ] { + let err = parse_datalog_command(q).err().expect("rejected"); + assert_eq!(crate::MinigrafError::from(err).code(), "PRS-082"); + } + parse_datalog_command( + r#"(query [:find ?d (count ?e) (rank :over (:partition-by ?d :order-by ?d)) :where [?e :d ?d]])"#, + ) + .expect("grouping variables are allowed"); + } + #[test] fn parse_udf_window_function() { // A name not in the built-in list should produce WindowFunc::Udf diff --git a/tests/aggregate_window_rules_test.rs b/tests/aggregate_window_rules_test.rs new file mode 100644 index 000000000..a60d06754 --- /dev/null +++ b/tests/aggregate_window_rules_test.rs @@ -0,0 +1,158 @@ +//! Aggregate and window answers that do not depend on the plan (#547, spec +//! `docs/superpowers/specs/2026-10-20-aggregate-window-engine-design.md` §3.4). +//! `plan_independence_test.rs` runs the same shapes with reordering on and +//! off; these check the values themselves, without the `test-oracle` +//! feature. + +use minigraf::{Minigraf, QueryResult, Value}; + +fn rows(db: &Minigraf, q: &str) -> Vec> { + match db.execute(q).expect("query") { + QueryResult::QueryResults { results, .. } => results, + _ => panic!("expected QueryResults"), + } +} + +fn one(db: &Minigraf, q: &str) -> Vec { + let mut r = rows(db, q); + assert_eq!(r.len(), 1, "one row"); + r.remove(0) +} + +fn float(v: &Value) -> f64 { + match v { + Value::Float(f) => *f, + _ => panic!("expected a float"), + } +} + +#[test] +fn float_sum_is_exact() { + let db = Minigraf::in_memory().unwrap(); + db.execute( + "(transact [[:a :v 10000000000000000.0] [:b :v 1.0] [:c :v -10000000000000000.0] \ + [:d :v 1.0] [:e :v 0.5]])", + ) + .unwrap(); + // Added left to right in f64, 1e16 + 1.0 loses the 1.0. + let r = one(&db, "(query [:find (sum ?v) :where [?e :v ?v]])"); + assert_eq!(float(&r[0]), 2.5); + let r = rows( + &db, + "(query [:find ?v (avg ?v :over (:order-by ?v)) :where [?e :v ?v]])", + ); + let last = r + .iter() + .find(|row| row[0] == Value::Float(1e16)) + .expect("row"); + assert_eq!(float(&last[1]), 0.5, "exact 2.5 over 5 rows"); +} + +#[test] +fn integer_sum_saturates_once() { + let db = Minigraf::in_memory().unwrap(); + db.execute("(transact [[:a :v 9223372036854775807] [:b :v 1] [:c :v -1]])") + .unwrap(); + let r = one(&db, "(query [:find (sum ?v) :where [?e :v ?v]])"); + assert_eq!(r[0], Value::Integer(i64::MAX)); +} + +#[test] +fn window_sum_of_integers_is_exact() { + let db = Minigraf::in_memory().unwrap(); + // 2^53 + 1 is not a float: summed as f64 it became 2^53. + db.execute("(transact [[:a :v 9007199254740992] [:b :v 1]])") + .unwrap(); + let r = rows( + &db, + "(query [:find ?v (sum ?v :over (:order-by ?v)) :where [?e :v ?v]])", + ); + let top = r + .iter() + .find(|row| row[0] == Value::Integer(9_007_199_254_740_992)) + .expect("row"); + assert_eq!(top[1], Value::Integer(9_007_199_254_740_993)); +} + +#[test] +fn min_max_of_signed_zeros_and_nan() { + let db = Minigraf::in_memory().unwrap(); + db.execute("(transact [[:a :v 0.0] [:b :v -0.0] [:c :v 1.5]])") + .unwrap(); + let r = one(&db, "(query [:find (min ?v) (max ?v) :where [?e :v ?v]])"); + assert_eq!(float(&r[0]).to_bits(), (-0.0f64).to_bits()); + assert_eq!(float(&r[1]), 1.5); +} + +#[test] +fn rank_by_keyword_ties_peers() { + let db = Minigraf::in_memory().unwrap(); + db.execute("(transact [[:a :d :x] [:b :d :x] [:c :d :y] [:d :d :y]])") + .unwrap(); + let mut ranks: Vec = rows( + &db, + "(query [:find ?e (rank :over (:order-by ?d)) :where [?e :d ?d]])", + ) + .iter() + .map(|r| match r[1] { + Value::Integer(n) => n, + _ => panic!("rank is an integer"), + }) + .collect(); + ranks.sort_unstable(); + assert_eq!(ranks, vec![1, 1, 3, 3]); +} + +#[test] +fn running_sum_gives_peers_the_value_at_the_last_peer() { + let db = Minigraf::in_memory().unwrap(); + db.execute("(transact [[:a :s 10] [:b :s 10] [:c :s 5] [:d :s 1.5]])") + .unwrap(); + let mut sums: Vec = rows( + &db, + "(query [:find ?e (sum ?s :over (:order-by ?s)) :where [?e :s ?s]])", + ) + .iter() + .map(|r| float(&r[1])) + .collect(); + sums.sort_by(f64::total_cmp); + assert_eq!(sums, vec![1.5, 6.5, 26.5, 26.5]); +} + +#[test] +fn row_number_breaks_ties_by_the_other_find_variables() { + let db = Minigraf::in_memory().unwrap(); + db.execute( + r#"(transact [[:a :name "b"] [:a :s 1] [:b :name "a"] [:b :s 1] [:c :name "c"] [:c :s 0]])"#, + ) + .unwrap(); + let mut r = rows( + &db, + "(query [:find ?n (row-number :over (:order-by ?s)) :where [?e :s ?s] [?e :name ?n]])", + ); + r.sort_by_key(|row| match &row[1] { + Value::Integer(n) => *n, + _ => 0, + }); + let names: Vec<&Value> = r.iter().map(|row| &row[0]).collect(); + assert_eq!( + names, + vec![ + &Value::String("c".into()), + &Value::String("a".into()), + &Value::String("b".into()) + ] + ); +} + +#[test] +fn aggregate_error_is_decided_per_group() { + let db = Minigraf::in_memory().unwrap(); + db.execute(r#"(transact [[:a :v 1] [:b :v "x"] [:c :v 2]])"#) + .unwrap(); + let err = db + .execute("(query [:find (sum ?v) :where [?e :v ?v]])") + .err() + .expect("sum of a string fails"); + assert_eq!(err.code(), "INT-041"); +} diff --git a/tests/plan_independence_test.rs b/tests/plan_independence_test.rs index bc81a68e7..66cf40177 100644 --- a/tests/plan_independence_test.rs +++ b/tests/plan_independence_test.rs @@ -117,6 +117,15 @@ fn people(db: &Minigraf) { /// Queries whose answers must not depend on the plan. const MATRIX: &[&str] = &[ + // Aggregates and windows (#547): exact sums, peers, the total value order. + r#"[:find ?a (count ?e) :where [?e :age ?a]]"#, + r#"[:find (sum ?s) (count-distinct ?s) :where [?p :score ?s]]"#, + r#"[:find (sum ?v) (min ?v) (max ?v) (count-distinct ?v) :where [?e ?attr ?v] [(float? ?v)]]"#, + r#"[:find ?n (row-number :over (:order-by ?a)) :where [?e :age ?a] [?e :name ?n]]"#, + r#"[:find ?n (sum ?a :over (:order-by ?a)) (rank :over (:order-by ?a :desc)) :where [?e :name ?n] [?e :age ?a]]"#, + r#"[:find ?e (rank :over (:order-by ?b)) (row-number :over (:order-by ?b)) :where [?e :friend ?b]]"#, + r#"[:find ?a (count ?e) (row-number :over (:order-by ?a :desc)) :where [?e :age ?a]]"#, + r#"[:find ?a (count ?b) :where (knows ?a ?b)]"#, // Patterns, joins, cross products. r#"[:find ?n ?a :where [?e :name ?n] [?e :age ?a]]"#, r#"[:find ?n ?a :where [?e :age ?a] [?e :name ?n]]"#, diff --git a/tests/window_functions_test.rs b/tests/window_functions_test.rs index da80d024b..3550ab1fa 100644 --- a/tests/window_functions_test.rs +++ b/tests/window_functions_test.rs @@ -253,23 +253,40 @@ fn row_number_desc_ordering() { #[test] fn mixed_aggregate_and_window_in_same_find() { let db = setup_employees(); - // count(e) collapses by dept, then sum runs over collapsed rows + // count(e) collapses by dept, then the window runs over the dept rows, + // reading only the grouping variable ?dept (#547). let result = db .execute( - r#"(query [:find ?dept (count ?e) (sum ?salary :over (:order-by ?salary)) - :with ?e ?salary + r#"(query [:find ?dept (count ?e) (rank :over (:order-by ?dept :desc)) :where [?e :employee/dept ?dept] [?e :employee/salary ?salary]])"#, ) .expect("query"); let rows = get_results(result); assert_eq!(rows.len(), 2, "expected one row per dept"); - // Each dept row has [dept, count, cumulative-sum] let eng = rows .iter() .find(|r| r[0] == Value::String("Engineering".into())) .unwrap(); assert_eq!(eng[1], Value::Integer(2), "Engineering has 2 employees"); + assert_eq!(eng[2], Value::Integer(2), "Product ranks first under :desc"); +} + +#[test] +fn window_in_aggregate_query_reading_a_non_grouping_variable_is_rejected() { + // Before #547 ?salary read nil on every dept row, so the window summed + // nothing; now it is PRS-082. + let db = setup_employees(); + let err = db + .execute( + r#"(query [:find ?dept (count ?e) (sum ?salary :over (:order-by ?salary)) + :with ?e ?salary + :where [?e :employee/dept ?dept] + [?e :employee/salary ?salary]])"#, + ) + .err() + .expect("rejected"); + assert_eq!(err.code(), "PRS-082"); } // ── single-row partition edge case ─────────────────────────────────────────