From dddc2b128d9a30f4cef04e5cc43531eedf4269b5 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Tue, 6 Oct 2026 20:30:44 +0530 Subject: [PATCH 1/2] perf: decide net-assert on v8 index keys before building facts (#379) Committed reads for a query are reduced to the records that survive net-assert while walking the index keys: the history of a triple is contiguous and newest first, so a transaction holding a retraction hides everything older (skipped, with a seek after a run of 8), and the first assertion per valid-time window wins. Only survivors are decoded and translated through the dictionary. :as-of N counter queries now take the same selective index path, passing over entries newer than N, instead of reading every fact in the file. The executor still applies net_asserted_facts to committed and pending facts together; prefiltering is exact because every pending fact is newer than every committed one and a record can only be hidden by a newer one. Randomized and differential (file vs in-memory) tests check this. point_query_chain_depth at depth 2000: churned attribute 4.61 -> 1.03 ms, attribute scan 5.87 -> 1.24 ms; :as-of point query 5.59 ms -> 23 us (depth 1). Design: docs/superpowers/specs/2026-10-06-live-key-scans-design.md Refs #379, #383 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_015GgLEs8YozJjYhwPEYgTUn --- CHANGELOG.md | 1 + CLAUDE.md | 4 +- README.md | 2 +- ROADMAP.md | 2 +- benches/minigraf_bench.rs | 5 + docs/BENCHMARKS.md | 20 ++ docs/TEST_COVERAGE.md | 4 +- .../specs/2026-10-06-live-key-scans-design.md | 77 +++++++ src/graph/storage.rs | 76 +++++++ src/query/datalog/executor.rs | 47 +++-- src/storage/btree.rs | 2 - src/storage/keys.rs | 106 ++++++++++ src/storage/mod.rs | 41 ++++ src/storage/persistent_facts.rs | 186 +++++++++++++++++ src/storage/reader.rs | 194 +++++++++++++++++- tests/live_reads_test.rs | 132 ++++++++++++ 16 files changed, 871 insertions(+), 28 deletions(-) create mode 100644 docs/superpowers/specs/2026-10-06-live-key-scans-design.md create mode 100644 tests/live_reads_test.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index d236784da..0777089ca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Internal +- **Committed reads hide superseded history on index keys** (#379). A query's committed facts are now reduced to the ones that survive net-assert (newest assertion per valid-time window, newer than every retraction) while walking the index keys, before any fact is built. Only surviving entries are decoded and translated through the dictionary. A transaction that retracted a triple hides everything older in it, and a long run of hidden entries is skipped with one `seek`. `:as-of N` point and attribute queries now take the same index path instead of reading every fact in the file. On a checkpointed file with 2,000 superseded values, a point query takes 1.03 ms instead of 4.61 ms, an attribute scan 1.24 ms instead of 5.87 ms, and an `:as-of` point query 23 µs instead of 5.6 ms (at depth 1). Results are unchanged: the executor still applies net-assert to committed and pending facts together, and a pending fact is always newer than a committed one. - **B+tree reads no longer follow leaf sibling links** (v8 storage format, PR 1 of 5; #374). Every index scan uses a cursor that walks down from the root and keeps the path, with a forward `seek` that climbs only as far as needed. This prepares for copy-on-write index pages, which cannot keep sibling links. The file format and query results are unchanged. Concurrent 100k-fact scans take 25–37 % less time and checkpointing after one fact at 10k facts 9 % less; single point lookups and attribute scans are 2–6 % slower (about 0.4 µs per lookup). A corrupted child pointer now fails with STG-013 (wrong page type) or INT-049 (cycle) instead of following a bad link. ### Tests diff --git a/CLAUDE.md b/CLAUDE.md index bed4d7914..b47811664 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -114,7 +114,7 @@ cargo run < demos/demo_negation.txt - `btree.rs`: On-disk B+tree over byte keys (`build_btree`, `cow_insert`, `LeafCursor` with `seek`, `prefix_scan`, `get`, `MutexStorageBackend`) - `dict.rs`: `DictReader` (id ↔ UUID/ident, tx timestamps, long values) and `Encoder` (checkpoint-time id assignment and key building) - `value_pages.rs`: append-only value pages for strings over 64 bytes - - `reader.rs`: `OnDiskReader` — covering reads of committed facts (`CommittedReader`) + - `reader.rs`: `OnDiskReader` — covering reads of committed facts (`CommittedReader`); `live_facts` decides net-assert on index keys (#379) - `meta.rs` / `page.rs` / `freelist.rs`: meta pages A/B, the common page header and allocator, free-list chain - `cache.rs`: LRU page cache (`PageCache`, default 256 pages) - `dir_sync.rs`: `sync_parent_dir` — fsyncs the parent directory after creating the `.graph`/WAL or deleting the WAL (no-op off Unix) @@ -193,7 +193,7 @@ Auto-migrates v7 → v8 on open (spec §9, with a backup meta page). v1–v6 are ## Test Coverage -**1238 tests** (1229 passing, 9 ignored; unit + integration + doc). +**1242 tests** (1233 passing, 9 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 fef7e6c2d..eb7db781a 100644 --- a/README.md +++ b/README.md @@ -105,7 +105,7 @@ let r2 = pq.execute(&[("tx", BindValue::TxCount(2)), ("entity", BindValue::Entit ```bash cargo run # interactive Datalog REPL -cargo test # run 1238 tests +cargo test # run 1242 tests cargo run < demos/demo_recursive.txt # recursive rules demo ``` diff --git a/ROADMAP.md b/ROADMAP.md index df8880166..6cefd5c56 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -61,7 +61,7 @@ Scope — format and storage integrity: The format is frozen for the golden-file corpus (#391). - Index `verify` and public `rebuild_indexes()` (#373) -- Net-assert on v8 keys before resolving facts, for O(live) point queries (#379) +- Net-assert on v8 keys before resolving facts, for O(live) point queries (#379; done on `v3`) - `btree_page` fuzz target that reaches node decoding (#375) Scope — release-gate testing and documentation: diff --git a/benches/minigraf_bench.rs b/benches/minigraf_bench.rs index d9c52bf63..4220bfb77 100644 --- a/benches/minigraf_bench.rs +++ b/benches/minigraf_bench.rs @@ -1673,6 +1673,11 @@ fn bench_point_query_chain_depth(c: &mut Criterion) { "(query [:find ?v :where [:e/hot :other ?v]])", ), ("attr_scan", "(query [:find ?v :where [?e :hash ?v]])"), + // tx 2002 asserted h0: an :as-of point query on committed history (#379). + ( + "churned_attr_as_of", + "(query [:find ?v :as-of 2002 :valid-at :any-valid-time :where [:e/hot :hash ?v]])", + ), ]; for &(name, q) in QUERIES { let mut group = c.benchmark_group(format!("point_query_chain_depth/{name}")); diff --git a/docs/BENCHMARKS.md b/docs/BENCHMARKS.md index d67494cba..ca094c867 100644 --- a/docs/BENCHMARKS.md +++ b/docs/BENCHMARKS.md @@ -50,6 +50,26 @@ One entity's `:hash` is retracted and reasserted `depth` times (exactly one live The churned attribute still scales with its own history on v2.x; see #379 for the v3.0.0 fix. +### Net-Assert on Index Keys (#379, v3.0.0) + +**Date**: 2026-10-06 · **Command**: `cargo bench --bench minigraf_bench -- point_query_chain_depth --warm-up-time 1 --measurement-time 3` · same fixture as above, on format v8. "Before" is `v3` at `74310ee`; "after" decides net-assert on index keys and translates only surviving entries. The `:as-of` row is new: `[:find ?v :as-of 2002 :valid-at :any-valid-time :where [:e/hot :hash ?v]]`, which used to scan every fact in the file. + +| Query | Depth | Before | After | +|---|---:|---:|---:| +| `[:e/hot :hash ?v]` (churned attribute) | 1 | 19.9 µs | 21.1 µs | +| | 500 | 1.10 ms | 262 µs | +| | 2000 | 4.61 ms | 1.03 ms | +| `[:e/hot :other ?v]` (sibling attribute) | 1 | 21.6 µs | 22.6 µs | +| | 2000 | 28.8 µs | 30.1 µs | +| `[?e :hash ?v]` (attribute scan) | 1 | 22.7 µs | 23.5 µs | +| | 500 | 1.20 ms | 294 µs | +| | 2000 | 5.87 ms | 1.24 ms | +| `:as-of 2002` on the churned attribute | 1 | 5.59 ms | 23.4 µs | +| | 500 | 6.30 ms | 216 µs | +| | 2000 | 8.55 ms | 845 µs | + +Depth-1 differences are within run-to-run noise: two back-to-back A/B runs measured 19.0–19.3 µs before and 19.8–19.9 µs after for the churned attribute, and 22.6 µs before and 21.2–21.5 µs after for the sibling attribute. In this fixture every retraction and re-assertion uses a new value, so each history entry is its own two-entry triple. The remaining cost is the walk over those keys. A triple with a long history of its own is skipped with one seek. + ### Checkpoint After One Dirty Fact (#315) **Date**: 2026-09-26 · **Command**: `cargo bench --bench minigraf_bench -- checkpoint/after_1_fact` · same host as above; file on tmpfs, so fsync cost is excluded. diff --git a/docs/TEST_COVERAGE.md b/docs/TEST_COVERAGE.md index 76a0116e9..d2d6bc229 100644 --- a/docs/TEST_COVERAGE.md +++ b/docs/TEST_COVERAGE.md @@ -4,9 +4,9 @@ **Verified**: 2026-10-06 with `cargo test --quiet` -**Result**: 1,229 passing tests and 9 ignored tests (1,238 total) +**Result**: 1,233 passing tests and 9 ignored tests (1,242 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. +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. 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. ### Covered Areas diff --git a/docs/superpowers/specs/2026-10-06-live-key-scans-design.md b/docs/superpowers/specs/2026-10-06-live-key-scans-design.md new file mode 100644 index 000000000..fbe0948db --- /dev/null +++ b/docs/superpowers/specs/2026-10-06-live-key-scans-design.md @@ -0,0 +1,77 @@ +# Net-assert on index keys (O(live) committed reads) — Design + +**Issue:** #379 (follow-up to #323), tracker #383 +**Milestone:** v3.0.0, branch `v3`, file format v8 (no format change) +**Builds on:** `2026-10-05-v8-storage-format-design.md` §5.3, §7 + +## 1. Problem + +Every non-`:as-of` query fetches facts, then applies `net_asserted_facts` (keep an +assertion only if it is the newest in its `(e, a, v, vf, vt)` window and newer than every +retraction of `(e, a, v)`). On v8, a committed read decodes every index entry of the +range and translates it into a `Fact` through the dictionary (UUID, attribute name, +timestamp, maybe a long-value page), including every superseded and retracted record. +The cost is O(history), and the history is thrown away a moment later. + +`:as-of N` queries are worse: they never take the selective path and always scan and +translate every fact in the file. + +## 2. Idea + +v8 keys are ordered `e a v tx↓ vf vt op` (EAVT; AEVT is `a e v …`). The history of one +triple is contiguous and newest first. Net-assert can therefore be decided on the key +bytes while walking, and only surviving entries are decoded and translated: + +- Walk the triple's entries in `tx` descending order, one transaction group at a time. +- If a group holds a retraction, its assertions and every older entry of the triple are + dead. They are passed over one by one; after 8 in a row the walk calls + `seek(triple ‖ 0xFF)` (`0xFF` sorts after every `tx↓` byte). Short histories, the + common case, pay no seek; long ones pay O(log) page visits. +- Otherwise each assertion survives if its `(vf, vt)` bytes have not been seen in this + triple; the first one seen is the newest of its window. +- `:as-of N`: entries newer than `N` are passed over the same way; the seek target is + `triple ‖ tx↓(N)`, the triple's newest entry at or before `N`. + +## 3. Why prefiltering committed facts is exact + +The executor still applies `net_asserted_facts` to committed and pending facts together. +Prefiltering is exact because every pending fact has a larger `tx_count` than every +committed one, and net-assert only ever lets a newer record hide an older one: + +- A committed assertion hidden by a committed record stays hidden in the union. +- A dropped committed retraction can hide only committed assertions older than it, and + those are already gone. It cannot hide a pending assertion, which is newer. +- `:as-of N` applies the same argument to the records with `tx_count ≤ N`. + +A randomized test checks `net(prefilter(committed) ∪ pending) = net(committed ∪ pending)` +for random histories, with and without `:as-of`. + +## 4. Changes + +- `keys::triple_len(index, key)`: bytes of the leading `(e, a, v)` (or index-order) + triple, without decoding the value. +- `CommittedReader::live_facts(scan, as_of)` with `scan` one of all, entity, + entity + attribute, attribute. The default implementation (history, then `as_of` + filter, then `net_asserted_facts`) is the reference; `OnDiskReader` overrides it with + the key walk. +- `FactStorage::get_live_facts(scan, as_of)`: pending facts (filtered by `as_of`) plus + committed live facts. +- Executor `filter_facts_for_query`: when there is no transaction overlay, the + `None` and `:as-of N` (counter) cases use live reads, selective when patterns bind + entities or attributes and a full live scan otherwise. `:as-of` timestamps, which + have no key form, and the transaction overlay keep their current path. + +## 5. Testing + +- `triple_len` agrees with a full decode for every value type and index. +- Randomized histories (asserts, retracts, re-asserts, valid-time windows, several + values per attribute, as-of points): `live_facts` from disk equals the reference for + every scan kind, before and after reopen. +- Query results with and without the prefilter are identical on the existing suites; + new point-query tests for retract/re-assert and `:as-of` on committed data. +- A counting backend shows that a point query on an entity with long history reads + fewer leaves than a history read. + +## 6. Philosophy + +Aligned: no format change, no API change, no dependency; less work per query. diff --git a/src/graph/storage.rs b/src/graph/storage.rs index 0dfa72f83..c832d86c8 100644 --- a/src/graph/storage.rs +++ b/src/graph/storage.rs @@ -504,6 +504,78 @@ fn pending_fact(d: &FactData, pos: usize) -> Result { /// Production helpers on FactStorage: index-driven entity/attribute lookups used by the query executor. impl FactStorage { + /// Facts for a query's net-assert step: the pending facts of `scan` with + /// `tx_count <= as_of` (all when `None`), plus the committed facts of `scan` + /// that survive net-assert at `as_of` (#379). + /// + /// `net_asserted_facts` over the result equals `net_asserted_facts` over + /// every record of `scan` up to `as_of` (see + /// [`CommittedReader::live_facts`](crate::storage::CommittedReader::live_facts)), + /// but committed history is not translated into facts. + pub(crate) fn get_live_facts( + &self, + scan: crate::storage::Scan<'_>, + as_of: Option, + ) -> Result> { + use crate::storage::Scan; + use crate::storage::index::{AevtKey, EavtKey}; + let d = self.data.read().unwrap_or_else(|e| e.into_inner()); + let mut facts = Vec::new(); + let mut push = |pos: usize| -> Result<()> { + let f = pending_fact(&d, pos)?; + if as_of.is_none_or(|n| f.tx_count <= n) { + facts.push(f); + } + Ok(()) + }; + match scan { + Scan::All => { + for pos in 0..d.facts.len() { + push(pos)?; + } + } + Scan::Entity(entity) => { + for (key, &pos) in d + .pending_indexes + .eavt + .range(EavtKey::entity_start(*entity)..) + { + if key.entity != *entity { + break; + } + push(pos)?; + } + } + Scan::EntityAttribute(entity, attribute) => { + let start = EavtKey::entity_attribute_start(*entity, attribute); + for (key, &pos) in d.pending_indexes.eavt.range(start..) { + if key.entity != *entity || key.attribute != attribute { + break; + } + push(pos)?; + } + } + Scan::Attribute(attribute) => { + for (key, &pos) in d + .pending_indexes + .aevt + .range(AevtKey::attribute_start(attribute)..) + { + if key.attribute != attribute { + break; + } + push(pos)?; + } + } + } + if let Some(reader) = &d.committed { + facts.extend(reader.live_facts(scan, as_of)?); + } + Ok(facts) + } + + /// History read (every record); tests compare it with live reads and models. + #[cfg(test)] /// Get all facts for a specific entity (index-driven). pub(crate) fn get_facts_by_entity(&self, entity_id: &EntityId) -> Result> { use crate::storage::index::EavtKey; @@ -525,6 +597,8 @@ impl FactStorage { Ok(facts) } + /// History read (every record); tests compare it with live reads and models. + #[cfg(test)] /// Get every stored record for one `(entity, attribute)` pair (index-driven, #323). /// /// Reads only that pair's EAVT range, so other attributes of the same entity @@ -551,6 +625,8 @@ impl FactStorage { Ok(facts) } + /// History read (every record); tests compare it with live reads and models. + #[cfg(test)] /// Get all facts for a specific attribute (index-driven). pub(crate) fn get_facts_by_attribute(&self, attribute: &Attribute) -> Result> { use crate::storage::index::AevtKey; diff --git a/src/query/datalog/executor.rs b/src/query/datalog/executor.rs index 23de7c9d8..942c9b0e3 100644 --- a/src/query/datalog/executor.rs +++ b/src/query/datalog/executor.rs @@ -308,8 +308,10 @@ impl DatalogExecutor { /// `execute_query_with_rules` call. /// /// Step 1 uses selective index-backed fetches when query patterns bind concrete entities - /// or attributes (up to 4 distinct lookups); falls back to `get_all_facts()` otherwise. - /// Step 2 (caching `net_asserted_facts()`) remains a future optimisation opportunity. + /// or attributes (up to 4 distinct lookups), and a full scan otherwise. Without a + /// transaction overlay, and for `:as-of` counters, committed history is already + /// reduced to its net-asserted records on index keys (#379); step 2 still runs over + /// the union with the pending facts, which gives the same result. fn filter_facts_for_query(&self, query: &DatalogQuery) -> Result> { let now = self.read_now(); @@ -318,20 +320,10 @@ impl DatalogExecutor { crate::graph::storage::filter_facts_as_of(facts.iter().cloned().collect(), as_of) } (Some(facts), None) => facts.iter().cloned().collect(), + // A counter is a key bound, so committed history is cut on index keys. + (None, Some(AsOf::Counter(n))) => self.live_source_facts(query, Some(*n))?, (None, Some(as_of)) => self.storage.get_facts_as_of(as_of)?, - (None, None) => { - // Selective fetch is only safe when no rule invocations are present — - // rules require the full fact base to evaluate correctly. - if !query.uses_rules() { - let patterns = collect_all_patterns(&query.where_clauses); - match self.selective_fact_fetch(&patterns, 4)? { - Some(facts) => facts, - None => self.storage.get_all_facts()?, - } - } else { - self.storage.get_all_facts()? - } - } + (None, None) => self.live_source_facts(query, None)?, }; let tx_filtered = source_facts; @@ -360,6 +352,20 @@ impl DatalogExecutor { Ok(Arc::from(valid_filtered)) } + /// Step 1 without a transaction overlay: pending facts up to `as_of` plus the + /// committed facts that survive net-assert at `as_of` (#379), selectively when the + /// patterns allow it. Rules need the whole fact base, so they always get a full scan. + fn live_source_facts(&self, query: &DatalogQuery, as_of: Option) -> Result> { + if !query.uses_rules() { + let patterns = collect_all_patterns(&query.where_clauses); + if let Some(facts) = self.selective_fact_fetch(&patterns, 4, as_of)? { + return Ok(facts); + } + } + self.storage + .get_live_facts(crate::storage::Scan::All, as_of) + } + /// Attempt a selective index-backed fact fetch for the given patterns. /// /// Patterns with a bound entity literal (UUID or keyword → deterministic UUID) are @@ -376,12 +382,15 @@ impl DatalogExecutor { /// A lookup error (a damaged page) is returned, never turned into a full scan. /// /// Results are not deduplicated: the sole caller feeds them to `net_asserted_facts`, - /// which is idempotent under duplicated records. + /// which is idempotent under duplicated records. Each lookup returns the pending facts + /// up to `as_of` and the committed facts live at `as_of` (`FactStorage::get_live_facts`). fn selective_fact_fetch( &self, patterns: &[Pattern], threshold: usize, + as_of: Option, ) -> Result>> { + use crate::storage::Scan; use std::collections::{BTreeMap, BTreeSet}; // Per bound entity: Some(attrs) = only these attributes are referenced; @@ -441,16 +450,16 @@ impl DatalogExecutor { for attr in attrs { all_facts.extend( self.storage - .get_facts_by_entity_attribute_indexed(uid, attr)?, + .get_live_facts(Scan::EntityAttribute(uid, attr), as_of)?, ); } } - _ => all_facts.extend(self.storage.get_facts_by_entity(uid)?), + _ => all_facts.extend(self.storage.get_live_facts(Scan::Entity(uid), as_of)?), } } for attr in &attributes { - all_facts.extend(self.storage.get_facts_by_attribute(attr)?); + all_facts.extend(self.storage.get_live_facts(Scan::Attribute(attr), as_of)?); } Ok(Some(all_facts)) diff --git a/src/storage/btree.rs b/src/storage/btree.rs index 786dc0930..2bb8a19e0 100644 --- a/src/storage/btree.rs +++ b/src/storage/btree.rs @@ -558,8 +558,6 @@ impl<'a> LeafCursor<'a> { /// Move forward to the first entry `>= key`. A key at or before the current /// position leaves the cursor where it is. - // Only tests call this until the streaming engine's joins (#432) do. - #[allow(dead_code)] pub(crate) fn seek(&mut self, key: &[u8]) -> Result<()> { if self.done { return Ok(()); diff --git a/src/storage/keys.rs b/src/storage/keys.rs index 6edbad4ce..dd43e5a31 100644 --- a/src/storage/keys.rs +++ b/src/storage/keys.rs @@ -90,6 +90,11 @@ impl<'a> Reader<'a> { self.pos >= self.buf.len() } + /// Bytes read so far. + pub fn position(&self) -> usize { + self.pos + } + fn byte(&mut self) -> Result { let b = self .buf @@ -361,6 +366,44 @@ impl Reader<'_> { }) } + /// Move past one encoded value without decoding it. + pub fn skip_value(&mut self) -> Result<()> { + match self.byte()? { + TAG_NULL => {} + TAG_BOOL => { + self.byte()?; + } + TAG_INT => { + self.int()?; + } + TAG_FLOAT => { + self.take(8)?; + } + TAG_KEYWORD | TAG_REF => { + self.uint()?; + } + TAG_STRING => loop { + if self.byte()? != 0 { + continue; + } + match self.byte()? { + 0xFF => {} + 0x00 => break, + // hash u64, page u64, slot u16 + 0x01 => { + self.take(18)?; + break; + } + other => { + bail_coded!(ErrorCode::Int049, format!("bad string escape {other:#04x}")) + } + } + }, + tag => bail_coded!(ErrorCode::Int049, format!("bad value tag {tag:#04x}")), + } + Ok(()) + } + fn string(&mut self) -> Result { let mut bytes = Vec::new(); loop { @@ -505,6 +548,31 @@ impl KeyFact { } } +/// Length of the leading triple of an `index` key: its three id/value +/// components, everything before `tx↓`. Entries with the same triple bytes are +/// the history of one `(e, a, v)`, newest first. The value is skipped, not decoded. +pub fn triple_len(index: Index, key: &[u8]) -> Result { + let mut r = Reader::new(key); + match index { + Index::Eavt | Index::Aevt => { + r.uint()?; + r.uint()?; + r.skip_value()?; + } + Index::Avet => { + r.uint()?; + r.skip_value()?; + r.uint()?; + } + Index::Vaet => { + r.uint()?; + r.uint()?; + r.uint()?; + } + } + Ok(r.position()) +} + /// Key prefix of every EAVT entry of entity `e`. pub fn entity_prefix(e: u64) -> Vec { let mut k = Vec::new(); @@ -961,6 +1029,44 @@ mod tests { assert!(non_ref.key(Index::Vaet).is_none(), "VAET holds refs only"); } + /// `triple_len` ends exactly where `tx↓` starts, for every value type and + /// index, and the byte after the triple is below 0xFF, so `triple ‖ 0xFF` + /// sorts after the triple's whole history. + #[test] + fn triple_len_ends_where_the_transaction_starts() { + let mut rng = Rng(4321); + for _ in 0..3000 { + let f = random_fact(&mut rng); + let mut tail = Vec::new(); + put_tx_desc(&mut tail, f.tx_count); + put_int(&mut tail, f.vf); + put_valid_to(&mut tail, f.vt); + tail.push(u8::from(f.asserted)); + for index in Index::ALL { + let Some(k) = f.key(index) else { continue }; + let n = triple_len(index, &k).unwrap(); + assert_eq!(n, k.len() - tail.len(), "triple length"); + assert!(k[n] < 0xFF, "tx byte sorts below 0xFF"); + } + } + let long = long_str(&"x\0y".repeat(40), ValueRef { page: 7, slot: 2 }); + let f = KeyFact { + e: 1, + a: 2, + v: long, + tx_count: 9, + vf: -3, + vt: VALID_TIME_FOREVER, + asserted: false, + }; + let k = f.key(Index::Eavt).unwrap(); + let mut r = Reader::new(&k); + r.uint().unwrap(); + r.uint().unwrap(); + r.value().unwrap(); + assert_eq!(triple_len(Index::Eavt, &k).unwrap(), r.position()); + } + #[test] fn prefixes_bound_exactly_their_entity_or_attribute() { let f = KeyFact { diff --git a/src/storage/mod.rs b/src/storage/mod.rs index bd1ee14bd..ef2049acd 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -357,6 +357,47 @@ pub trait CommittedReader: Send + Sync { ) -> Result>; /// Every committed fact with `attribute` (an AEVT scan). fn facts_for_attribute(&self, attribute: &str) -> Result>; + + /// The committed facts of `scan` that survive net-assert among the + /// committed records with `tx_count <= as_of` (every record when `None`): + /// for each `(e, a, v)`, the newest assertion per valid-time window that is + /// newer than every retraction (#379). + /// + /// Applying `net_asserted_facts` to these plus the pending facts gives the + /// same result as applying it to every record, because each pending fact is + /// newer than each committed one and a record can only be hidden by a newer + /// one. This default reads the history and filters it; `OnDiskReader` + /// decides on index keys and translates only the survivors. + fn live_facts( + &self, + scan: Scan<'_>, + as_of: Option, + ) -> Result> { + let history = match scan { + Scan::All => self.all_facts()?, + Scan::Entity(e) => self.facts_for_entity(e)?, + Scan::EntityAttribute(e, a) => self.facts_for_entity_attribute(e, a)?, + Scan::Attribute(a) => self.facts_for_attribute(a)?, + }; + let in_window = history + .into_iter() + .filter(|f| as_of.is_none_or(|n| f.tx_count <= n)) + .collect(); + Ok(crate::graph::storage::net_asserted_facts(in_window)) + } +} + +/// The facts a read covers. +#[derive(Debug, Clone, Copy)] +pub enum Scan<'a> { + /// Every fact. + All, + /// The facts of one entity. + Entity(&'a crate::graph::types::EntityId), + /// The facts of one `(entity, attribute)` pair. + EntityAttribute(&'a crate::graph::types::EntityId, &'a str), + /// The facts with one attribute. + Attribute(&'a str), } #[cfg(test)] diff --git a/src/storage/persistent_facts.rs b/src/storage/persistent_facts.rs index 28c0e85bc..ebe3e3b2f 100644 --- a/src/storage/persistent_facts.rs +++ b/src/storage/persistent_facts.rs @@ -1457,6 +1457,192 @@ mod tests { ); } + // ── live reads on index keys (#379) ───────────────────────────────────── + + /// The scans a live read can take over the random-history domain. + fn live_scans(entities: &[Uuid], attrs: &[String]) -> Vec> { + use crate::storage::Scan; + let entities: &'static [Uuid] = Box::leak(entities.to_vec().into_boxed_slice()); + let attrs: &'static [String] = Box::leak(attrs.to_vec().into_boxed_slice()); + let mut scans = vec![Scan::All]; + for e in entities { + scans.push(Scan::Entity(e)); + for a in attrs { + scans.push(Scan::EntityAttribute(e, a)); + } + } + for a in attrs { + scans.push(Scan::Attribute(a)); + } + scans + } + + /// Live reads from `pfs` agree with net-assert over the model's records, for + /// every scan and `as_of`. With nothing pending, the committed live read is + /// exactly the net-asserted set, not just equivalent after net-assert. + fn assert_live_reads_match( + pfs: &PersistentFactStorage, + model: &FactStorage, + scans: &[crate::storage::Scan<'_>], + as_ofs: &[Option], + ) { + use crate::graph::storage::net_asserted_facts; + let exact = pfs.storage().get_pending_facts().is_empty(); + for &scan in scans { + for &as_of in as_ofs { + let expected = as_set(net_asserted_facts( + model.get_live_facts(scan, as_of).unwrap(), + )); + let got = pfs.storage().get_live_facts(scan, as_of).unwrap(); + if exact { + assert!( + as_set(got.clone()) == expected, + "committed live read is exact" + ); + } + assert!( + as_set(net_asserted_facts(got)) == expected, + "live read equals net-assert over every record" + ); + } + } + } + + /// Random histories: assertions in several valid-time windows, retractions, + /// re-assertions, several values per attribute, long values, many + /// transactions per checkpoint. Live reads match net-assert over every + /// record at every checkpoint, with pending facts on top, and after reopen. + #[test] + fn live_reads_match_net_assert_on_random_histories() { + use crate::graph::types::TransactOptions; + let mut seed = 0x379u64; + let mut next = move |n: u64| { + seed ^= seed << 13; + seed ^= seed >> 7; + seed ^= seed << 17; + seed % n + }; + let entities: Vec = (0..5).map(entity).collect(); + let attrs: Vec = (0..3).map(|a| format!(":h/{a}")).collect(); + let scans = live_scans(&entities, &attrs); + let mem = MemoryBackend::new(); + let mut pfs = PersistentFactStorage::new(mem.clone(), 32).unwrap(); + let model = FactStorage::new(); + let mut step = |pfs: &mut PersistentFactStorage, + next: &mut dyn FnMut(u64) -> u64| { + let mut tuples = Vec::new(); + for _ in 0..=next(3) { + let v = match next(5) { + 0 => Value::String(format!("{}{}", "long history value ".repeat(5), next(2))), + 1 => Value::Keyword(format!(":k/{}", next(2))), + n => Value::Integer(n as i64), + }; + tuples.push(( + entities[next(5) as usize], + attrs[next(3) as usize].clone(), + v, + )); + } + if next(3) == 0 { + pfs.storage().retract(tuples).unwrap(); + } else { + let opts = match next(4) { + 0 => None, + 1 => Some(TransactOptions::new(Some(10), Some(20))), + 2 => Some(TransactOptions::new(Some(20), None)), + _ => Some(TransactOptions::new(Some(5), Some(15))), + }; + pfs.storage().transact(tuples, opts).unwrap(); + } + pfs.mark_dirty(); + }; + for _round in 0..6 { + for _ in 0..12 { + step(&mut pfs, &mut next); + } + for f in pfs.storage().get_pending_facts() { + model.load_fact(f).unwrap(); + } + pfs.save().unwrap(); + let now = pfs.storage().current_tx_count(); + let as_ofs = [ + None, + Some(0), + Some(now / 3), + Some(now / 2), + Some(now - 1), + Some(now), + ]; + assert_live_reads_match(&pfs, &model, &scans, &as_ofs); + } + // Pending facts on top of committed ones. + for _ in 0..12 { + step(&mut pfs, &mut next); + } + for f in pfs.storage().get_pending_facts() { + model.load_fact(f).unwrap(); + } + let now = pfs.storage().current_tx_count(); + let as_ofs = [None, Some(now / 2), Some(now - 3), Some(now)]; + assert_live_reads_match(&pfs, &model, &scans, &as_ofs); + pfs.save().unwrap(); + drop(pfs); + let pfs = PersistentFactStorage::new(mem, 32).unwrap(); + assert_live_reads_match(&pfs, &model, &scans, &[None, Some(now / 2)]); + } + + /// A point read on a triple with a long history reads a few pages, not the + /// history: the retraction behind the current assertion skips the rest with + /// one seek, and `as_of` enters the triple with one seek. + #[test] + fn live_point_read_skips_superseded_history() { + use crate::storage::Scan; + let log = ReadLog { + inner: MemoryBackend::new(), + read: Arc::new(Mutex::new(Vec::new())), + }; + let reads = log.read.clone(); + let mut pfs = PersistentFactStorage::new(log, 0).unwrap(); + let e = entity(1); + let fact = || vec![(e, ":status".to_string(), Value::Integer(7))]; + for _ in 0..2000 { + pfs.storage().transact(fact(), None).unwrap(); + pfs.storage().retract(fact()).unwrap(); + } + pfs.storage().transact(fact(), None).unwrap(); + let middle = 2000; + pfs.mark_dirty(); + pfs.save().unwrap(); + let s = pfs.storage(); + + reads.lock().unwrap().clear(); + assert_eq!(s.get_facts_by_entity(&e).unwrap().len(), 4001); + let history = std::mem::take(&mut *reads.lock().unwrap()).len(); + + let live = s.get_live_facts(Scan::Entity(&e), None).unwrap(); + assert_eq!(live.len(), 1, "only the current assertion"); + let live_reads = std::mem::take(&mut *reads.lock().unwrap()).len(); + + let at = s.get_live_facts(Scan::Entity(&e), Some(middle)).unwrap(); + assert!(at.is_empty(), "retracted at an even transaction"); + let at = s + .get_live_facts(Scan::Entity(&e), Some(middle - 1)) + .unwrap(); + assert_eq!(at.len(), 1, "asserted at an odd transaction"); + assert_eq!(at[0].tx_count, middle - 1); + let as_of_reads = std::mem::take(&mut *reads.lock().unwrap()).len(); + + assert!(history > 20, "the history spans many leaves"); + assert!( + live_reads * 3 < history, + "live read: {live_reads} pages, history: {history}" + ); + assert!( + as_of_reads * 3 < history * 2, + "two as-of reads: {as_of_reads} pages, history: {history}" + ); + } + fn value_page_count(mem: &MemoryBackend) -> usize { (2..mem.page_count().unwrap()) .filter(|&id| mem.read_page(id).unwrap()[0] == page::PAGE_TYPE_VALUE) diff --git a/src/storage/reader.rs b/src/storage/reader.rs index 036dd823b..dab7e6d2a 100644 --- a/src/storage/reader.rs +++ b/src/storage/reader.rs @@ -9,8 +9,9 @@ use crate::storage::cache::PageCache; use crate::storage::dict::{DictReader, SharedDictCache}; use crate::storage::keys::{self, Index, KeyFact}; use crate::storage::meta::MetaPage; -use crate::storage::{CommittedReader, StorageBackend}; +use crate::storage::{CommittedReader, Scan, StorageBackend}; use anyhow::Result; +use std::collections::HashSet; use std::sync::{Arc, Mutex}; /// [`CommittedReader`] over one committed meta's trees. @@ -57,7 +58,198 @@ impl OnDiskReader { } } +impl OnDiskReader { + /// Facts of `index` entries starting with `prefix` that survive net-assert + /// at `as_of`, decided on the key bytes (#379, spec §5.3). + /// + /// The history of one triple is contiguous and newest first. Entries are + /// taken one transaction at a time: a transaction that holds a retraction + /// hides its own assertions and everything older in the triple. Otherwise + /// each assertion is kept if its valid-time window (`vf vt` bytes) has not + /// been seen in the triple yet. With `as_of`, entries newer than `as_of` + /// are passed over. Entries that are passed over are skipped one by one, and + /// after [`SEEK_AFTER`] of them in one triple the cursor seeks past the run, + /// so a long history costs O(log) page visits while a short one costs no + /// seek. Only kept entries are decoded and translated through the dictionary. + fn live_scan( + &self, + index: Index, + root: u64, + prefix: &[u8], + as_of: Option, + dict: &mut DictReader<'_>, + ) -> Result> { + let mut out = Vec::new(); + if root == 0 { + return Ok(out); + } + let mut cursor = LeafCursor::new(root, Some(prefix), &self.backend, &self.cache)?; + let mut triple: Vec = Vec::new(); + let mut windows: HashSet> = HashSet::new(); + let mut group = TxGroup::default(); + // Entries of the current triple are hidden from here on. + let mut dead = false; + let mut skipped = 0usize; + loop { + let step = match cursor.next_ref()? { + Some((k, _)) if k.starts_with(prefix) => { + if dead && k.starts_with(&triple) { + Step::Pass(Some(0xFF)) + } else { + let n = keys::triple_len(index, k)?; + let (key_triple, rest) = k.split_at(n); + let mut r = keys::Reader::new(rest); + let tx = r.tx_desc()?; + let new_triple = key_triple != triple.as_slice(); + if new_triple { + group.flush(index, &mut windows, dict, &mut out)?; + triple.clear(); + triple.extend_from_slice(key_triple); + windows.clear(); + dead = false; + skipped = 0; + } + if as_of.is_some_and(|limit| tx > limit) { + Step::Pass(None) + } else { + let window = rest + .get(r.position()..rest.len().saturating_sub(1)) + .unwrap_or_default(); + let retraction = k.last() == Some(&0); + Step::Entry { + tx, + window: window.to_vec(), + key: (!retraction).then(|| k.clone()), + } + } + } + } + _ => { + group.flush(index, &mut windows, dict, &mut out)?; + return Ok(out); + } + }; + match step { + Step::Pass(past) => { + skipped = skipped.saturating_add(1); + if skipped >= SEEK_AFTER { + skipped = 0; + let mut target = triple.clone(); + match (past, as_of) { + (Some(b), _) => target.push(b), + (None, Some(limit)) => keys::put_tx_desc(&mut target, limit), + (None, None) => {} + } + cursor.seek(&target)?; + } + } + Step::Entry { tx, window, key } => { + if group.tx != Some(tx) { + if group.flush(index, &mut windows, dict, &mut out)? { + // The retraction hides this older entry too. + dead = true; + skipped = 1; + continue; + } + group = TxGroup { + tx: Some(tx), + ..TxGroup::default() + }; + } + match key { + Some(key) => group.assertions.push((window, key)), + None => group.retracted = true, + } + } + } + } + } +} + +/// Entries of one triple passed over before the walk seeks past them. +const SEEK_AFTER: usize = 8; + +/// What the walk does with one entry. +enum Step { + /// Pass over it; a seek past the run appends this byte to the triple + /// (`0xFF`: past the whole triple), or `tx↓(as_of)` when `None`. + Pass(Option), + /// An entry at or before `as_of`; `key` is kept for assertions only. + Entry { + tx: u64, + window: Vec, + key: Option>, + }, +} + +/// The entries of one triple with one `tx_count`. +#[derive(Default)] +struct TxGroup { + tx: Option, + retracted: bool, + /// `(vf vt bytes, key)` of each assertion. + assertions: Vec<(Vec, Vec)>, +} + +impl TxGroup { + /// Emit the surviving assertions; returns true if the group held a + /// retraction (so every older entry of the triple is hidden too). Resets + /// the group. + fn flush( + &mut self, + index: Index, + windows: &mut HashSet>, + dict: &mut DictReader<'_>, + out: &mut Vec, + ) -> Result { + let group = std::mem::take(self); + if group.retracted { + return Ok(true); + } + for (window, key) in group.assertions { + if windows.insert(window) { + out.push(dict.fact(&KeyFact::decode(index, &key)?)?); + } + } + Ok(false) + } +} + impl CommittedReader for OnDiskReader { + fn live_facts(&self, scan: Scan<'_>, as_of: Option) -> Result> { + let mut dict = self.dict(); + let (index, root, prefix) = match scan { + Scan::All => (Index::Eavt, self.eavt_root, Vec::new()), + Scan::Entity(entity) => { + let Some(e) = dict.eid_of(entity)? else { + return Ok(Vec::new()); + }; + dict.seed_entity(e, *entity); + (Index::Eavt, self.eavt_root, keys::entity_prefix(e)) + } + Scan::EntityAttribute(entity, attribute) => { + let (Some(e), Some(a)) = (dict.eid_of(entity)?, dict.iid_of(attribute)?) else { + return Ok(Vec::new()); + }; + dict.seed_entity(e, *entity); + dict.seed_ident(a, attribute); + ( + Index::Eavt, + self.eavt_root, + keys::entity_attribute_prefix(e, a), + ) + } + Scan::Attribute(attribute) => { + let Some(a) = dict.iid_of(attribute)? else { + return Ok(Vec::new()); + }; + dict.seed_ident(a, attribute); + (Index::Aevt, self.aevt_root, keys::attribute_prefix(a)) + } + }; + self.live_scan(index, root, &prefix, as_of, &mut dict) + } + fn all_facts(&self) -> Result> { if self.eavt_root == 0 { return Ok(Vec::new()); diff --git a/tests/live_reads_test.rs b/tests/live_reads_test.rs new file mode 100644 index 000000000..fbeb4653a --- /dev/null +++ b/tests/live_reads_test.rs @@ -0,0 +1,132 @@ +//! Query results over committed history equal those over the same history held +//! in memory (#379). The file-backed database checkpoints between batches, so +//! its committed facts go through the key-level net-assert walk; the in-memory +//! database runs the same commands with every fact pending. +#![cfg(not(target_arch = "wasm32"))] + +use minigraf::QueryResult; +use minigraf::Value; +use minigraf::db::Minigraf; + +fn rows(db: &Minigraf, query: &str) -> Vec { + match db.execute(query).unwrap() { + QueryResult::QueryResults { results, .. } => { + let mut out: Vec = results + .iter() + .map(|row| { + row.iter() + .map(|v| match v { + Value::String(s) => s.clone(), + Value::Integer(n) => n.to_string(), + Value::Keyword(k) => k.clone(), + Value::Boolean(b) => b.to_string(), + Value::Float(f) => f.to_string(), + Value::Null => "nil".to_string(), + Value::Ref(_) => "ref".to_string(), + }) + .collect::>() + .join("|") + }) + .collect(); + out.sort(); + out + } + _ => panic!("expected query results"), + } +} + +/// Commands in batches; the file-backed database checkpoints after each batch. +fn history() -> Vec> { + let long = "a long status description ".repeat(4); + let mut batches = Vec::new(); + for round in 0..5 { + let mut b = Vec::new(); + for p in 0..6 { + let status = (round + p) % 3; + b.push(format!( + r#"(transact [[:p{p} :person/status :s{status}] [:p{p} :person/age {}] [:p{p} :person/friend :p{}]])"#, + 20 + round, + (p + 1) % 6 + )); + if (round + p) % 2 == 0 { + b.push(format!( + r#"(retract [[:p{p} :person/status :s{status}] [:p{p} :person/age {}]])"#, + 20 + round + )); + } + b.push(format!( + r#"(transact {{:valid-from "2023-01-01" :valid-to "2023-06-30"}} [[:p{p} :person/role "{long}{}"]])"#, + round % 2 + )); + b.push(format!( + r#"(transact {{:valid-from "2024-01-01"}} [[:p{p} :person/role "{long}{}"]])"#, + round % 2 + )); + if round == 3 { + b.push(format!(r#"(retract [[:p{p} :person/role "{long}1"]])"#)); + } + } + batches.push(b); + } + batches +} + +fn queries(max_tx: u64) -> Vec { + let mut q = vec![ + "(query [:find ?s :where [:p1 :person/status ?s]])".to_string(), + "(query [:find ?a :where [:p2 :person/age ?a]])".to_string(), + "(query [:find ?attr ?v :where [:p3 ?attr ?v]])".to_string(), + "(query [:find ?p ?s :where [?p :person/status ?s]])".to_string(), + "(query [:find ?p ?r :valid-at :any-valid-time :where [?p :person/role ?r]])".to_string(), + r#"(query [:find ?p ?r :valid-at "2023-03-01" :where [?p :person/role ?r]])"#.to_string(), + "(query [:find ?p ?f ?a :where [?p :person/friend ?f] [?f :person/age ?a]])".to_string(), + "(query [:find ?e ?a ?v :where [?e ?a ?v]])".to_string(), + ]; + for n in [1, max_tx / 3, max_tx / 2, max_tx - 1, max_tx] { + q.push(format!( + "(query [:find ?s :as-of {n} :where [:p0 :person/status ?s]])" + )); + q.push(format!( + "(query [:find ?p ?a :as-of {n} :valid-at :any-valid-time :where [?p :person/age ?a]])" + )); + q.push(format!( + "(query [:find ?e ?a ?v :as-of {n} :valid-at :any-valid-time :where [?e ?a ?v]])" + )); + } + q +} + +#[test] +fn committed_and_in_memory_histories_answer_alike() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("live.graph"); + let file = Minigraf::open(&path).unwrap(); + let mem = Minigraf::in_memory().unwrap(); + for (i, batch) in history().iter().enumerate() { + for cmd in batch { + file.execute(cmd).unwrap(); + mem.execute(cmd).unwrap(); + } + // Leave the last batch pending so committed and pending facts mix. + if i + 1 < history().len() { + file.checkpoint().unwrap(); + } + } + let rule = "(rule [(knows ?a ?b) [?a :person/friend ?b]])"; + file.execute(rule).unwrap(); + mem.execute(rule).unwrap(); + let max_tx = file.current_tx_count(); + assert_eq!(max_tx, mem.current_tx_count()); + let mut all = queries(max_tx); + all.push("(query [:find ?a ?b :where (knows ?a ?b)])".to_string()); + for q in &all { + assert_eq!(rows(&file, q), rows(&mem, q), "query: {q}"); + } + file.checkpoint().unwrap(); + drop(file); + let file = Minigraf::open(&path).unwrap(); + file.execute(rule).unwrap(); + for q in &all { + assert_eq!(rows(&file, q), rows(&mem, q), "after reopen: {q}"); + } +} From f1ab67ac07a8edf1613ac7de9685b1ebe7f7fbd1 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Tue, 6 Oct 2026 22:29:50 +0530 Subject: [PATCH 2/2] test: drop needless mut in live-read test Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_015GgLEs8YozJjYhwPEYgTUn --- src/storage/persistent_facts.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/storage/persistent_facts.rs b/src/storage/persistent_facts.rs index e5c1144a0..c0479b458 100644 --- a/src/storage/persistent_facts.rs +++ b/src/storage/persistent_facts.rs @@ -1710,8 +1710,8 @@ mod tests { let mem = MemoryBackend::new(); let mut pfs = PersistentFactStorage::new(mem.clone(), 32).unwrap(); let model = FactStorage::new(); - let mut step = |pfs: &mut PersistentFactStorage, - next: &mut dyn FnMut(u64) -> u64| { + let step = |pfs: &mut PersistentFactStorage, + next: &mut dyn FnMut(u64) -> u64| { let mut tuples = Vec::new(); for _ in 0..=next(3) { let v = match next(5) {