From 35d3fdafab451154ffabdb02b50cda4be1967205 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Fri, 23 Oct 2026 20:05:40 +0530 Subject: [PATCH 1/7] docs(plan): value probes for joins on a bound value (#721) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0169pNQsLcCEn48te9X9WuZW --- .../plans/2026-10-23-721-value-probes.md | 56 +++++++++++++++++++ 1 file changed, 56 insertions(+) create mode 100644 docs/superpowers/plans/2026-10-23-721-value-probes.md diff --git a/docs/superpowers/plans/2026-10-23-721-value-probes.md b/docs/superpowers/plans/2026-10-23-721-value-probes.md new file mode 100644 index 000000000..877e5ed6c --- /dev/null +++ b/docs/superpowers/plans/2026-10-23-721-value-probes.md @@ -0,0 +1,56 @@ +# Value probes (#721) — Plan + +Issue: #721. One PR to `v3`, merge commit, no closing keyword (the issue is closed by hand on +merge). #551 PR 3 waits for it, so its envelope records the fixed recursion numbers. + +## Problem + +A join whose inner pattern shares only its value with the rows so far (`[?x :manager ?y]` with +`?y` bound) is a `HashJoin` that holds or streams every fact of the attribute. In a rule's delta +run it does so every round: #700 orders the ancestors shape's delta run that way, and +`recursive_ancestors` went from 2 ms to 1.8 s at 1M facts and 35 s at 10M. The engine probes +by entity only; a value probe was left out because the pending store has no value index +(#545 plan, decision 3). + +## Decisions + +1. **`JoinKind::ProbeValue(k)`**: when the inner pattern's entity is not bound, its attribute is a + constant, its value is key `k`, and it reads no derived relation, the join probes per outer + row instead of hashing. An entity probe still wins when both apply. A pattern that reads a + relation keeps the `HashJoin` (relations are indexed by entity only). The operator is still + an `IndexNestedLoopJoin`. +2. **Committed side: AVET ranges of the values that join-equal the outer cell** (the join rules, + spec §5.4 of #473): + - a keyword ident `i`: `Keyword(i)` and `Ref(e)` of the entity `i` names, where each is a + DICT id; + - an entity `e`: `Ref(e)` and `Keyword(i)` for the keyword naming `e`, found through a map + from named entity to keyword ident, built once per scan from the attribute's keyword range + (at most what today's `HashJoin` reads, and only for entity cells); + - a float `f`: `Float(f)`, and `Float(-f)` when `f` is zero; NaN reads nothing; + - any other scalar: its committed key value, if the generation can hold it. + The ranges are a superset of the join: the join still checks every key, the probed one too, so + answers are the `HashJoin`'s. +3. **Pending side: a map from join key to the attribute's live pending facts**, built on the first + probe from one `pending_live` read of the attribute and charged to the meter. The shadow set + (committed triples a pending record decides) is built once per scan too, not per probe. +4. **No answer changes.** The frozen answers (`tests/answers/`), the plan-independence matrix and + the model-based tests must pass unchanged. + +## Steps + +Each step builds and passes clippy on all targets and features and leaves the suite passing, +plain and with `--features test-hooks`; tests run on the bench machine. + +1. **Plan.** This file. +2. **Planner:** `JoinKind::ProbeValue`, chosen per decision 1; a pattern with a derived relation + is downgraded to a `HashJoin` before the build-side choice. Test: the ancestors shape's delta + run (`(boss :leaf ?b)` after the magic-set rewrite) has no `HashJoin` over a stored pattern, + and the #694 fan-out shape keeps no keyless join. +3. **Scan and join:** `Scan::probe_value` (decisions 2, 3), the join's per-row probe and key + check. Tests: a value join reads only the probed keys (operator counters) on a checkpointed + file, with pending facts, and after a pending retraction of a committed fact; keyword ↔ ref in + both directions, an entity cell against keyword values, `0.0`/`-0.0`, NaN, a long string + committed and pending, `:as-of` and `:valid-at`; the same answers as with probes off. +4. **Measure and document:** Criterion `recursion` benches and the #394 envelope at 100K/1M/10M + against `v3`, A/B on the bench machine; `CHANGELOG.md` (`Fixed`, #721), `CLAUDE.md` (engine + notes), `docs/TEST_COVERAGE.md`, test counts; release binary size (gate 1,200 KiB). From f88de9751a1e9dc06c28efc0feaa067c6c3e871b Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Fri, 23 Oct 2026 20:20:50 +0530 Subject: [PATCH 2/7] fix(engine): probe a join on a bound value through AVET (#721) A join whose inner pattern shares only its value with the rows so far, under a constant attribute, was a HashJoin over every fact of the attribute. In a rule's delta run it was rebuilt every round: since #700 the ancestors shape (boss :leaf ?b) took 1.8 s at 1M facts instead of 2 ms. It is now an IndexNestedLoopJoin on the value: per outer row, the AVET ranges of the values that join-equal it (a keyword and the ref to the entity it names, either way; 0.0 and -0.0), and the attribute's pending facts with its join key, read once per scan with the shadow set. The join checks every key, so answers are the hash join's. A pattern that reads a relation, or has a constant entity, keeps its hash join. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0169pNQsLcCEn48te9X9WuZW --- src/query/engine/ids.rs | 5 + src/query/engine/join.rs | 31 ++- src/query/engine/meter.rs | 5 + src/query/engine/plan.rs | 23 ++- src/query/engine/scan.rs | 334 +++++++++++++++++++++++++++++---- src/query/engine/tests.rs | 181 +++++++++++++++++- tests/answers/engine_tests.txt | 9 + 7 files changed, 538 insertions(+), 50 deletions(-) diff --git a/src/query/engine/ids.rs b/src/query/engine/ids.rs index 08f1caff6..7a2fd1b33 100644 --- a/src/query/engine/ids.rs +++ b/src/query/engine/ids.rs @@ -210,6 +210,11 @@ impl IdSpace { (eid < self.dict_eids).then_some(eid) } + /// `iid` as a DICT ident id, or `None` for a provisional one. + pub(crate) fn dict_iid(&self, iid: u64) -> Option { + u32::try_from(iid).ok().filter(|_| iid < self.dict_iids) + } + /// The ident with id `iid`. pub(crate) fn name(&mut self, keys: Option<&dyn KeySource>, iid: u64) -> Result> { let maps = self.maps(iid >= self.dict_iids); diff --git a/src/query/engine/join.rs b/src/query/engine/join.rs index 1727ce3f4..ffd6f65a5 100644 --- a/src/query/engine/join.rs +++ b/src/query/engine/join.rs @@ -81,7 +81,7 @@ enum Side { impl Join { pub(crate) fn new(outer: Box, spec: JoinSpec, meter: &Arc) -> Self { let op = match spec.kind { - JoinKind::Probe(_) => "IndexNestedLoopJoin", + JoinKind::Probe(_) | JoinKind::ProbeValue(_) => "IndexNestedLoopJoin", JoinKind::Hash { .. } => "HashJoin", }; Join { @@ -214,16 +214,17 @@ impl Join { Ok(Some((uuid, ctx.ids.dict_eid(e)))) } - /// Whether `outer` and `inner` agree on every key but the probed one. + /// Whether `outer` and `inner` agree on every key but the probed one, + /// `None` for every key. fn agree( &self, outer: &[Cell], inner: &[Cell], - probed: usize, + probed: Option, ctx: &mut Ctx<'_>, ) -> Result { for (n, k) in self.keys.iter().enumerate() { - if n == probed { + if Some(n) == probed { continue; } let (Some(o), Some(i)) = (outer.get(k.outer), inner.get(k.inner)) else { @@ -237,19 +238,22 @@ impl Join { } /// Join outer rows into `out` until it is full; `false` once the outer - /// side has ended. + /// side has ended. The probe is by entity, or by value (#721), whose + /// ranges may hold values the key does not equal, so its key is checked. fn next_probe( &mut self, p: &mut Probe, probed: usize, + by_value: bool, out: &mut Batch, ctx: &mut Ctx<'_>, ) -> Result { + let skip = (!by_value).then_some(probed); while out.len < BATCH_ROWS { if let Some(o) = p.rows.get(p.at) { if let Some(m) = p.matches.get(p.next) { p.next += 1; - if self.agree(o, m, probed, ctx)? { + if self.agree(o, m, skip, ctx)? { self.emit(o, m, out); } } else if p.probing { @@ -262,6 +266,14 @@ impl Join { } } p.next = 0; + } else if by_value { + match self.keys.get(probed).and_then(|k| o.get(k.outer)) { + Some(cell) => { + self.inner.probe_value(cell, &self.held, ctx)?; + p.probing = true; + } + None => p.at += 1, + } } else { match self.probe_of(o, probed, ctx)? { Some((uuid, eid)) => { @@ -343,7 +355,7 @@ impl Operator for Join { let mut out = Batch::new(self.outer_width + self.new.len()); if let State::Start = self.state { self.state = match self.kind { - JoinKind::Probe(_) => State::Probe(Probe::default()), + JoinKind::Probe(_) | JoinKind::ProbeValue(_) => State::Probe(Probe::default()), JoinKind::Hash { build_inner } => { let built = if build_inner { Side::Inner @@ -367,8 +379,9 @@ impl Operator for Join { } // Taken out while it runs; an error leaves the join done. let more = match (std::mem::replace(&mut self.state, State::Done), self.kind) { - (State::Probe(mut p), JoinKind::Probe(k)) => { - let more = self.next_probe(&mut p, k, &mut out, ctx)?; + (State::Probe(mut p), JoinKind::Probe(k) | JoinKind::ProbeValue(k)) => { + let by_value = matches!(self.kind, JoinKind::ProbeValue(_)); + let more = self.next_probe(&mut p, k, by_value, &mut out, ctx)?; self.state = State::Probe(p); more } diff --git a/src/query/engine/meter.rs b/src/query/engine/meter.rs index f224cf8de..e1521066c 100644 --- a/src/query/engine/meter.rs +++ b/src/query/engine/meter.rs @@ -63,6 +63,11 @@ pub(crate) struct Held { } impl Held { + /// A new holder on the same meter, for the same operator, holding nothing. + pub(crate) fn sibling(&self) -> Self { + Held::new(&self.meter, self.op) + } + pub(crate) fn new(meter: &Arc, op: &'static str) -> Self { Held { meter: meter.clone(), diff --git a/src/query/engine/plan.rs b/src/query/engine/plan.rs index f8c0aa302..0bfeec206 100644 --- a/src/query/engine/plan.rs +++ b/src/query/engine/plan.rs @@ -252,6 +252,11 @@ pub(crate) enum JoinKind { /// `IndexNestedLoopJoin`: the inner pattern's entity is `keys[k]`; one /// entity range per outer row. Probe(usize), + /// `IndexNestedLoopJoin` on the value (#721): the inner pattern's + /// attribute is a constant and its value is `keys[k]`; per outer row, + /// the AVET ranges of the values that join-equal it, and the pending + /// facts with its key. + ProbeValue(usize), /// `HashJoin`: a table of the side with the smaller estimate, probed by /// the other. Hash { build_inner: bool }, @@ -1003,6 +1008,11 @@ impl Builder { self.steps.append(&mut self.front); } else { let mut join = join(&scanned, &attribute, slots, &rebind, valid)?; + // Relations are indexed by entity only: a value join on + // one hashes (#721). + if derived.is_some() && matches!(join.kind, JoinKind::ProbeValue(_)) { + join.kind = JoinKind::Hash { build_inner: true }; + } join.inner.derived = derived; for &(m, var) in &wants { let local = join.inner.width; @@ -1340,6 +1350,7 @@ fn join( let mut new = Vec::new(); let mut replace = Vec::new(); let mut probe = None; + let mut probe_value = None; // Each variable at its first position in the pattern. for (position, (edn, kind)) in [ (&q.entity, OTHER), @@ -1363,6 +1374,11 @@ fn join( let held = slots.kinds.get(outer).copied().unwrap_or(OTHER); if position == 0 { probe = Some(keys.len()); + } else if position == 2 + && matches!(a, Term::Const(_)) + && !matches!(e, Term::Const(_)) + { + probe_value = Some(keys.len()); } keys.push(Key { inner, @@ -1393,9 +1409,10 @@ fn join( width: local.len(), derived: None, }, - kind: match probe { - Some(k) => JoinKind::Probe(k), - None => JoinKind::Hash { build_inner: true }, + kind: match (probe, probe_value) { + (Some(k), _) => JoinKind::Probe(k), + (None, Some(k)) => JoinKind::ProbeValue(k), + (None, None) => JoinKind::Hash { build_inner: true }, }, keys, new, diff --git a/src/query/engine/scan.rs b/src/query/engine/scan.rs index 5f825c4d2..054536694 100644 --- a/src/query/engine/scan.rs +++ b/src/query/engine/scan.rs @@ -11,7 +11,8 @@ //! row is visible at every time and has no metadata (`nil`). use super::batch::Batch; -use super::cell::{Cell, Scalar}; +use super::cell::{Cell, JoinKey, Scalar}; +use super::meter::{HASH_ENTRY_OVERHEAD, Held, key_bytes, value_bytes}; use super::plan::{Meta, ScanSpec, Term, Valid, ValueConst, names}; use super::relation::Range as Rows; use super::{BATCH_ROWS, Counters, Ctx, Operator}; @@ -21,6 +22,8 @@ use crate::storage::keys::{self, Index, KeyFact, KeyValue, SHORT_STRING_MAX, Val use crate::storage::{KeySource, Scan as Range}; use anyhow::Result; use std::collections::{HashMap, HashSet}; +use std::mem::size_of; +use std::sync::Arc; pub(crate) struct Scan { spec: ScanSpec, @@ -33,14 +36,33 @@ pub(crate) struct Scan { state: State, keys_buf: Vec, facts_buf: Vec, + /// What a value probe reads besides its key ranges (#721), built on the + /// first one, and the join key of the current one. + values: Option, + value_key: Option, /// `rows_in` counts the live keys and pending facts read. counters: Counters, } +/// The reads a value probe shares with every later one (#721), charged to +/// `held`. +struct Values { + /// The attribute's live pending facts, by the join key of their value. + pending: HashMap>, + /// The committed triples a pending record of the attribute decides. + shadow: Arc>>, + /// The keyword idents among the attribute's committed values, by the + /// entity each names; read on the first probe by an entity. + named: Option>>, + held: Held, +} + enum State { Start, Committed(Committed), Pending(Option), + /// A value probe's pending facts. + Bucket(Vec), Derived(Derived), Done, } @@ -67,7 +89,7 @@ struct Committed { /// The value constant when the ranges do not fix it. test: Option, /// Triples a pending record decides, as [`keys::entity_attribute_value_prefix`]. - shadow: HashSet>, + shadow: Arc>>, } /// The attribute and value constants of a scan, against the DICT. @@ -119,6 +141,8 @@ impl Scan { state: State::Start, keys_buf: Vec::new(), facts_buf: Vec::new(), + values: None, + value_key: None, counters: Counters::new("Scan"), } } @@ -127,8 +151,229 @@ impl Scan { /// only: one probe of an `IndexNestedLoopJoin`. pub(crate) fn probe(&mut self, uuid: EntityId, eid: Option) { self.probe = Some((uuid, eid)); + self.value_key = None; self.state = State::Start; } + + /// Start again, over the facts whose value may join-equal `cell` only: + /// one probe of an `IndexNestedLoopJoin` on the value (#721). The + /// committed facts come from the AVET ranges of the values equal to + /// `cell` under the join rules, the pending ones from the attribute's + /// facts with its key; the join checks every key, so these need only + /// hold every match. `held` is charged for what the probes share. + pub(crate) fn probe_value( + &mut self, + cell: &Cell, + held: &Held, + ctx: &mut Ctx<'_>, + ) -> Result<()> { + self.probe = None; + self.value_key = None; + self.state = State::Done; + let Term::Const(attribute) = &self.spec.a else { + return Ok(()); + }; + if self.spec.never() { + return Ok(()); + } + if self.consts.is_none() + && let Some(keys) = ctx.keys + { + self.consts = Some(consts(&self.spec, keys)?); + } + if self.values.is_none() { + self.values = Some(values(attribute, held.sibling(), ctx)?); + } + let Some(key) = JoinKey::of(cell, true, ctx)? else { + return Ok(()); + }; + let a = match (&self.consts, ctx.keys) { + (Some(Some(Consts { a: Some(a), .. })), Some(_)) => Some(*a), + _ => None, + }; + let ranges = match a { + Some(a) => self.value_ranges(cell, a, ctx)?, + None => Vec::new(), + }; + self.value_key = Some(key); + self.state = if ranges.is_empty() { + self.bucket() + } else { + let shadow = self + .values + .as_ref() + .map_or_else(Arc::default, |v| v.shadow.clone()); + State::Committed(Committed { + ranges, + next: 0, + resume: None, + test: None, + shadow, + }) + }; + Ok(()) + } + + /// The AVET ranges of attribute `a` holding every committed value that + /// join-equals `cell`. + fn value_ranges( + &mut self, + cell: &Cell, + a: u32, + ctx: &mut Ctx<'_>, + ) -> Result)>> { + let Some(keys) = ctx.keys else { + return Ok(Vec::new()); + }; + let mut kvs = Vec::new(); + match cell { + // A keyword, and a ref to the entity it names. + Cell::Iid(i) => { + if let Some(di) = ctx.ids.dict_iid(*i) { + kvs.push(KeyValue::Keyword(di)); + } + let e = ctx.ids.keyword_eid(ctx.keys, *i)?; + if let Some(de) = ctx.ids.dict_eid(e) { + kvs.push(KeyValue::Ref(de)); + } + } + // A ref, and the keyword that names the entity. + Cell::Eid(e) => { + if let Some(de) = ctx.ids.dict_eid(*e) { + kvs.push(KeyValue::Ref(de)); + } + if let Some(v) = &mut self.values + && v.named.is_none() + { + v.named = Some(named(a, &mut v.held, ctx)?); + } + if let Some(idents) = self + .values + .as_ref() + .and_then(|v| v.named.as_ref()) + .and_then(|n| n.get(e)) + { + kvs.extend(idents.iter().map(|&i| KeyValue::Keyword(i))); + } + } + // `-0.0` equals `0.0`; NaN has no key and never gets here. + Cell::Val(Scalar::Float(f)) => { + kvs.push(KeyValue::Float(*f)); + if *f == 0.0 { + kvs.push(KeyValue::Float(-*f)); + } + } + Cell::Val(_) => { + let v = super::project::value(cell, ctx)?; + kvs.extend(key_value(keys, &v)?); + } + } + Ok(kvs + .iter() + .map(|v| (Index::Avet, keys::attribute_value_prefix(a, v))) + .collect()) + } + + /// After the committed ranges: a value probe's pending facts, or the + /// pending range. + fn bucket(&self) -> State { + match (&self.value_key, &self.values) { + (Some(k), Some(v)) => State::Bucket(v.pending.get(k).cloned().unwrap_or_default()), + (Some(_), None) => State::Done, + _ => State::Pending(None), + } + } +} + +/// The live pending facts of `attribute` by the join key of their value, +/// and the committed triples its pending records decide (#721). +fn values(attribute: &str, mut held: Held, ctx: &mut Ctx<'_>) -> Result { + let mut pending: HashMap> = HashMap::new(); + let mut resume = None; + let mut buf = Vec::new(); + loop { + buf.clear(); + let more = ctx.snapshot.pending_live( + Range::Attribute(attribute), + ctx.bound, + &mut resume, + BATCH_ROWS, + &mut buf, + )?; + for f in buf.drain(..) { + let cell = Cell::of_value(ctx.ids, ctx.keys, &f.value)?; + let Some(k) = JoinKey::of(&cell, true, ctx)? else { + continue; + }; + held.charge( + HASH_ENTRY_OVERHEAD + + key_bytes(&k) + + size_of::() + + f.entity.as_bytes().len() + + f.attribute.len() + + value_bytes(&f.value), + )?; + pending.entry(k).or_default().push(f); + } + if !more { + break; + } + } + let mut shadow = HashSet::new(); + if let Some(keys) = ctx.keys { + for f in ctx + .snapshot + .pending_records(Range::Attribute(attribute), ctx.bound)? + { + if let Some(t) = committed_triple(keys, &f)? { + held.charge(HASH_ENTRY_OVERHEAD + size_of::>() + t.len())?; + shadow.insert(t); + } + } + } + Ok(Values { + pending, + shadow: Arc::new(shadow), + named: None, + held, + }) +} + +/// The keyword idents among the live committed values of attribute `a`, by +/// the entity each names (#721). +fn named(a: u32, held: &mut Held, ctx: &mut Ctx<'_>) -> Result>> { + let mut by_entity: HashMap> = HashMap::new(); + let Some(keys) = ctx.keys else { + return Ok(by_entity); + }; + let prefix = keys::attribute_kind_prefix(a, ValueKind::Keyword); + let mut seen = HashSet::new(); + let mut resume: Option> = None; + let mut buf = Vec::new(); + loop { + buf.clear(); + resume = keys.live_keys( + Index::Avet, + &prefix, + resume.as_deref(), + ctx.bound, + BATCH_ROWS, + &mut buf, + )?; + for kf in &buf { + if let KeyValue::Keyword(i) = kf.v + && seen.insert(i) + { + let e = ctx.ids.keyword_eid(ctx.keys, u64::from(i))?; + held.charge(HASH_ENTRY_OVERHEAD + size_of::<(u64, Vec)>() + size_of::())?; + by_entity.entry(e).or_default().push(i); + } + } + if resume.is_none() { + break; + } + } + Ok(by_entity) } impl Operator for Scan { @@ -156,11 +401,11 @@ impl Operator for Scan { } State::Committed(c) => { let Some(keys) = ctx.keys else { - self.state = State::Pending(None); + self.state = self.bucket(); continue; }; let Some((index, prefix)) = c.ranges.get(c.next) else { - self.state = State::Pending(None); + self.state = self.bucket(); continue; }; self.keys_buf.clear(); @@ -221,37 +466,7 @@ impl Operator for Scan { &mut self.facts_buf, )?; self.counters.rows_in += self.facts_buf.len() as u64; - let valid = valid_at(self.spec.valid, ctx); - let mut row = vec![None; self.spec.width]; - for f in &self.facts_buf { - if !visible(valid, f.valid_from, f.valid_to) - || matches!(&self.spec.v, Term::Const(v) if !v.matches(&f.value)) - { - continue; - } - let keys = ctx.keys; - let ids = &mut *ctx.ids; - if fill(&self.spec.e, &mut row, || { - Ok(Cell::Eid(ids.eid(keys, &f.entity)?)) - })? && fill(&self.spec.a, &mut row, || { - Ok(Cell::Iid(ids.iid(keys, &f.attribute)?)) - })? && fill(&self.spec.v, &mut row, || { - Cell::of_value(ids, keys, &f.value) - })? { - for &(m, slot) in &self.spec.meta { - let n = match m { - Meta::ValidFrom => Some(f.valid_from), - Meta::ValidTo => Some(f.valid_to), - Meta::TxCount => Some(f.tx_count.cast_signed()), - Meta::TxId => Some(f.tx_id.cast_signed()), - Meta::ValidAt => valid, - }; - set(&mut row, slot, n); - } - batch.push(&mut row); - } - row.iter_mut().for_each(|c| *c = None); - } + push_facts(&self.spec, &self.facts_buf, &mut batch, ctx)?; if !more { self.state = if self.spec.derived.is_some() { State::Derived(Derived::default()) @@ -260,6 +475,15 @@ impl Operator for Scan { }; } } + State::Bucket(facts) => { + let rest = facts.split_off(facts.len().min(BATCH_ROWS)); + let now = std::mem::replace(facts, rest); + self.counters.rows_in += now.len() as u64; + push_facts(&self.spec, &now, &mut batch, ctx)?; + if now.is_empty() { + self.state = State::Done; + } + } State::Derived(d) => { if !derived(&self.spec, self.probe.as_ref(), d, &mut batch, ctx)? { self.state = State::Done; @@ -277,6 +501,7 @@ impl Operator for Scan { /// Start again; the constants resolved against the DICT are kept. fn rewind(&mut self, _seed: &[Cell]) { self.probe = None; + self.value_key = None; self.state = State::Start; } @@ -284,6 +509,8 @@ impl Operator for Scan { self.state = State::Done; self.keys_buf = Vec::new(); self.facts_buf = Vec::new(); + self.values = None; + self.value_key = None; } fn counters(&self, out: &mut Vec) { @@ -291,6 +518,41 @@ impl Operator for Scan { } } +/// Push the rows of pending `facts` that match `spec` and are visible. +fn push_facts(spec: &ScanSpec, facts: &[Fact], batch: &mut Batch, ctx: &mut Ctx<'_>) -> Result<()> { + let valid = valid_at(spec.valid, ctx); + let mut row = vec![None; spec.width]; + for f in facts { + if !visible(valid, f.valid_from, f.valid_to) + || matches!(&spec.v, Term::Const(v) if !v.matches(&f.value)) + { + continue; + } + let keys = ctx.keys; + let ids = &mut *ctx.ids; + if fill(&spec.e, &mut row, || { + Ok(Cell::Eid(ids.eid(keys, &f.entity)?)) + })? && fill(&spec.a, &mut row, || { + Ok(Cell::Iid(ids.iid(keys, &f.attribute)?)) + })? && fill(&spec.v, &mut row, || Cell::of_value(ids, keys, &f.value))? + { + for &(m, slot) in &spec.meta { + let n = match m { + Meta::ValidFrom => Some(f.valid_from), + Meta::ValidTo => Some(f.valid_to), + Meta::TxCount => Some(f.tx_count.cast_signed()), + Meta::TxId => Some(f.tx_id.cast_signed()), + Meta::ValidAt => valid, + }; + set(&mut row, slot, n); + } + batch.push(&mut row); + } + row.iter_mut().for_each(|c| *c = None); + } + Ok(()) +} + /// Read derived rows of `spec` into `batch` until it is full; `false` once /// every relation has been read. fn derived( @@ -457,7 +719,7 @@ fn open( next: 0, resume: None, test, - shadow, + shadow: Arc::new(shadow), })) } diff --git a/src/query/engine/tests.rs b/src/query/engine/tests.rs index c222c42ef..2700b2fbe 100644 --- a/src/query/engine/tests.rs +++ b/src/query/engine/tests.rs @@ -862,6 +862,7 @@ fn planner_chooses_the_join() { .filter_map(|s| match s { Step::Join(j) => Some(match j.kind { JoinKind::Probe(_) => "probe".to_string(), + JoinKind::ProbeValue(_) => "probe value".to_string(), JoinKind::Hash { build_inner } => format!("hash inner={build_inner}"), }), _ => None, @@ -872,9 +873,15 @@ fn planner_chooses_the_join() { kinds("(query [:find ?e :where [?e :a ?x] [?e :b ?y]])"), ["probe"] ); + // On a bound value with a constant attribute, a probe of its AVET + // ranges (#721); with no constant attribute, a hash join. assert_eq!( kinds("(query [:find ?e :where [?e :a ?x] [?f :b ?x]])"), - ["hash inner=true"] + ["probe value"] + ); + assert_eq!( + kinds("(query [:find ?e :where [?e :a ?x] [?f ?b ?x]])"), + ["hash inner=false"] ); assert_eq!( kinds("(query [:find ?e :where [?e :a 1] [?f :b ?y] [?f :c ?e]])"), @@ -891,6 +898,12 @@ fn planner_chooses_the_join() { kinds("(query [:find ?e :where [?e :a 1] [?f :b ?y] [?f :c ?e]])"), ["hash inner=true", "probe"] ); + // A constant entity reads one entity's range: a hash join, not a + // probe of the value's. + assert_eq!( + kinds("(query [:find ?e :where [?e :a ?x] [:k :b ?x]])"), + ["hash inner=true"] + ); } #[test] @@ -2237,6 +2250,170 @@ fn rule_delta_runs_join_on_a_bound_variable() { assert_eq!(rows(&db, "(query [:find ?to :where (reach :n1 ?to)])"), 10); } +/// #721: after the magic-set rewrite of `(boss :leaf ?b)` over a tree, the +/// run from the `boss` delta joins `[?x :manager ?y]` on its value. It +/// probes the AVET range of each `?y` instead of hashing every `:manager` +/// fact in every round; a pattern that reads a relation keeps its hash join. +#[test] +fn rule_delta_probes_a_bound_value() { + use super::plan::{JoinKind, Step}; + use crate::query::datalog::parser::parse_datalog_command; + use crate::query::datalog::rules::RuleRegistry; + use crate::query::datalog::types::DatalogCommand; + let mut registry = RuleRegistry::new(); + for rule in [ + "(rule [(boss ?x ?y) [?x :manager ?y]])", + "(rule [(boss ?x ?z) [?x :manager ?y] (boss ?y ?z)])", + ] { + let DatalogCommand::Rule(r) = parse_datalog_command(rule).unwrap() else { + panic!("expected a rule"); + }; + registry.register_rule("boss".to_string(), r).unwrap(); + } + let DatalogCommand::Query(query) = + parse_datalog_command("(query [:find ?b :where (boss :p99 ?b)])").unwrap() + else { + panic!("expected a query"); + }; + let (rewritten, seeds) = + crate::query::datalog::magic_sets::rewrite(&query, ®istry).expect("magic sets apply"); + let attrs: Vec<&str> = seeds.iter().map(|(_, a, _)| a.as_str()).collect(); + let plan = super::plan_rules(&query, &rewritten, &attrs).expect("the engine plans it"); + let mut value_probes = 0; + for (_, rules) in &plan.strata { + for rule in rules { + for run in &rule.deltas { + for step in &run.plan.steps { + let Step::Join(j) = step else { continue }; + match j.kind { + JoinKind::ProbeValue(_) => { + assert!( + j.inner.derived.is_none(), + "a relation is not probed by value" + ); + value_probes += 1; + } + JoinKind::Hash { .. } => assert!( + j.inner.derived.is_some(), + "a delta run of {} hashes a stored attribute", + rule.predicate + ), + JoinKind::Probe(_) => {} + } + } + } + } + } + assert!(value_probes >= 1, "the boss delta probes :manager by value"); + + // The answers on the tree: :p99's managers are :p9 and :p0. + let db = Minigraf::in_memory().unwrap(); + let facts: String = (1..200) + .map(|i| format!("[:p{i} :manager :p{}]", i / 10)) + .collect(); + db.execute(&format!("(transact [{facts}])")).unwrap(); + db.execute("(rule [(boss ?x ?y) [?x :manager ?y]])") + .unwrap(); + db.execute("(rule [(boss ?x ?z) [?x :manager ?y] (boss ?y ?z)])") + .unwrap(); + assert_eq!(rows(&db, "(query [:find ?b :where (boss :p99 ?b)])"), 2); + assert_eq!(rows(&db, "(query [:find ?b :where (boss :p199 ?b)])"), 3); +} + +/// #721: a join on a bound value probes the AVET ranges of the values that +/// join-equal it, not every fact of the attribute, on committed and on +/// pending facts. +#[test] +fn value_probe_reads_only_the_probed_keys() { + let dir = tempfile::tempdir().unwrap(); + let db = Minigraf::open(dir.path().join("probe.graph")).unwrap(); + let facts: String = (0..1000) + .map(|i| format!("[:e{i} :boss :b{}]", i % 100)) + .collect(); + db.execute(&format!("(transact [{facts}])")).unwrap(); + db.checkpoint().unwrap(); + // Pending: one more report of :b5, and one of :b5's committed reports + // retracted. + db.execute("(transact [[:n1 :boss :b5]])").unwrap(); + db.execute("(retract [[:e105 :boss :b5]])").unwrap(); + let q = "(query [:find ?s :where [:e5 :boss ?b] [?s :boss ?b]])"; + assert_eq!( + rows(&db, q), + 10, + "nine of :b5's ten committed reports, and :n1" + ); + // The counters of the open stream; its one probe has read every match. + let mut cursor = db.query(q).unwrap(); + cursor.next_batch(1).unwrap().expect("a row"); + let c = cursor.counters(); + assert!( + c.iter().any(|c| c.op == "IndexNestedLoopJoin"), + "a probe, not a hash join" + ); + let read: u64 = c.iter().filter(|c| c.op == "Scan").map(|c| c.rows_in).sum(); + assert!(read < 30, "reads the probed keys only, not 1,000 facts"); +} + +/// #721: a value probe matches as a join does: a keyword and a ref to the +/// entity it names, from either side; `0.0` and `-0.0`; long strings; with +/// `:as-of` and valid time; in memory, checkpointed and pending. +#[test] +fn value_probe_matches_as_joins_do() { + let x = uuid_of(":x"); + let long = long(1); + let first = format!( + "(transact [[:a :v :x] [:b :w :x] [:x :name \"X\"] \ + [:f1 :n 0.0] [:f2 :m -0.0] [:f3 :m 0.0] [:f4 :m 1.0] \ + [:l1 :t \"{long}\"] [:l2 :u \"{long}\"]])" + ); + let second = format!("(transact [[:c :w #uuid \"{x}\"] [:d :w :y] [:l3 :u \"{long}\"]])"); + let expired = "(transact {:valid-from \"2020-01-01\" :valid-to \"2021-01-01\"} [[:old :w :x] [:q :v :x]])"; + let queries = [ + // A keyword outer value: the keyword and the ref. + ("(query [:find ?s :where [:a :v ?k] [?s :w ?k]])", 2), + // An entity outer value: the ref and the keyword naming it. + ("(query [:find ?s :where [?k :name \"X\"] [?s :w ?k]])", 2), + ("(query [:find ?s :where [:f1 :n ?z] [?s :m ?z]])", 2), + ("(query [:find ?s :where [:l1 :t ?t] [?s :u ?t]])", 2), + // Before the second transaction: the keyword only. + ( + "(query [:find ?s :as-of 1 :where [:a :v ?k] [?s :w ?k]])", + 1, + ), + // In 2020: only the facts valid then. + ( + "(query [:find ?s :valid-at \"2020-06-01\" :where [:q :v ?k] [?s :w ?k]])", + 1, + ), + ]; + let check = |db: &Minigraf, mode: &str| { + for (q, n) in queries { + assert_eq!(rows(db, q), n, "{mode}: {q}"); + } + }; + + let db = Minigraf::in_memory().unwrap(); + for cmd in [first.as_str(), second.as_str(), expired] { + db.execute(cmd).unwrap(); + } + check(&db, "in memory"); + + let dir = tempfile::tempdir().unwrap(); + for (name, checkpoint_after) in [("all.graph", 3), ("half.graph", 1)] { + let db = Minigraf::open(dir.path().join(name)).unwrap(); + for (i, cmd) in [first.as_str(), second.as_str(), expired] + .into_iter() + .enumerate() + { + db.execute(cmd).unwrap(); + if i + 1 == checkpoint_after { + db.checkpoint().unwrap(); + } + } + check(&db, name); + } +} + // ── #549: query memory accounting ────────────────────────────────────────── /// A checkpointed database of `n` entities, each with integers `:a` and @@ -2548,7 +2725,7 @@ fn planner_builds_a_keyless_join_on_the_smaller_product() { .filter_map(|s| match s { Step::Join(j) => match j.kind { JoinKind::Hash { build_inner } => Some(build_inner), - JoinKind::Probe(_) => None, + JoinKind::Probe(_) | JoinKind::ProbeValue(_) => None, }, _ => None, }) diff --git a/tests/answers/engine_tests.txt b/tests/answers/engine_tests.txt index 194163fff..468d67ecb 100644 --- a/tests/answers/engine_tests.txt +++ b/tests/answers/engine_tests.txt @@ -4945,6 +4945,8 @@ not_bodies_plan_as_clause_lists 07a2cd1cfc45bcc6 2 54e8aae90ee45459 not_bodies_plan_as_clause_lists 92ae19a63d955c9d 3 334b7473ff8bc18a not_bodies_plan_as_clause_lists 9c4ccbd1c930e739 3 334b7473ff8bc18a planner_errors_keep_their_codes 8d2f701fd7f4a458 1 d800c42686702fe9 +rule_delta_probes_a_bound_value 7aac9bf2085f7f70 2 504c6f55a0f9d00e +rule_delta_probes_a_bound_value a2bb83359e06a257 3 c49c7d5bab6f3ac5 rule_delta_runs_join_on_a_bound_variable 3a840c32501fd782 110 78d43ae4a90656ea rule_delta_runs_join_on_a_bound_variable ae359baa9829c163 10 2b10d79d79ce1507 rule_queries_read_derived_rows_in_batches 012b6bf9d9c1ae59 1 c13a0e3963062b18 @@ -4957,6 +4959,13 @@ rule_query_routing 4fb26550baf3a6d1 1 75741c9011909650 rule_query_routing 57c18e0e4c2d3d59 2 02b1d1046853e1c4 rule_query_routing 9cdc76b397fe1ac5 3 2b9101d4b11ba754 rule_query_routing e3c9735c8586e200 0 cbf29ce484222325 +value_probe_matches_as_joins_do 04701079619bafd4 2 b11858569b739dda +value_probe_matches_as_joins_do 436327ca6645dc60 2 b11858569b739dda +value_probe_matches_as_joins_do 565efa7a38dac1a4 2 ecfa7f988c5996fa +value_probe_matches_as_joins_do 7739c10f98f3604c 2 70dd264d4a2a703b +value_probe_matches_as_joins_do a8b0b8fb4cec0d4f 1 af63b82e992110a1 +value_probe_matches_as_joins_do efbdf901cf8540a7 1 75741c9011909650 +value_probe_reads_only_the_probed_keys 7a28c763b652fd24 10 ccca6be15b98175d window_routing 34f4a4cc8f6e63ca 4 7edb2d3ac44b6289 window_routing b16fb0833e2b5faf 4 ad00ad205380e0a2 window_routing d1a4aa676b518bd6 0 cbf29ce484222325 From a657b04b0bd100266be5f55f9eae084b5f1e9018 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Fri, 23 Oct 2026 22:27:30 +0530 Subject: [PATCH 3/7] fix(engine): value probes read once per query; hash join when an entity's keyword is unknown (#721) The pending facts and shadow set of a value probe are read once per query, not per scan, so a rule run in every round reads them once. An entity cell must also find the keyword that names it, and a keyword's entity is a hash of its name: no index leads back. The query records the keywords it makes entities from (rule heads, and the keyword constants its rule calls name, the magic seeds); a probe by an entity without one, on an attribute with committed keyword values, turns the join into the hash join the planner chose before, on the side it named. recursive_ancestors (bench machine, p50): 1.1 ms at 100K, 1.4 ms at 1M, 1.8 ms at 10M (v3: 101 ms, 1.8 s, 35 s). join_3_selective 145 ms -> 0.2 ms at 1M; join_3_scan unchanged. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0169pNQsLcCEn48te9X9WuZW --- src/query/datalog/executor.rs | 13 ++ src/query/engine/ids.rs | 4 + src/query/engine/join.rs | 59 ++++++-- src/query/engine/meter.rs | 13 +- src/query/engine/mod.rs | 13 +- src/query/engine/plan.rs | 18 ++- src/query/engine/recursive.rs | 18 ++- src/query/engine/scan.rs | 262 ++++++++++++++++++--------------- src/query/engine/tests.rs | 27 +++- tests/answers/engine_tests.txt | 1 + 10 files changed, 283 insertions(+), 145 deletions(-) diff --git a/src/query/datalog/executor.rs b/src/query/datalog/executor.rs index 58f77c994..9e7f06def 100644 --- a/src/query/datalog/executor.rs +++ b/src/query/datalog/executor.rs @@ -399,6 +399,18 @@ impl DatalogExecutor { query_memory: self.query_memory_limit, }; let snapshot = self.read_snapshot()?; + // The keywords the query's rule calls name as entities (#721). + let keywords: Vec = query + .where_clauses + .iter() + .filter_map(|c| match c { + WhereClause::RuleInvocation { args, .. } => match args.first() { + Some(EdnValue::Keyword(k)) => Some(k.clone()), + _ => None, + }, + _ => None, + }) + .collect(); // The query's predicates, then its rules' (#551 plan, decision 9). let mut names = Vec::new(); crate::query::engine::udfs(&query.where_clauses, &mut names); @@ -408,6 +420,7 @@ impl DatalogExecutor { snapshot.clone(), plan, &seeds, + &keywords, self.functions.clone(), limits, )?; diff --git a/src/query/engine/ids.rs b/src/query/engine/ids.rs index 7a2fd1b33..abbd1057e 100644 --- a/src/query/engine/ids.rs +++ b/src/query/engine/ids.rs @@ -37,6 +37,8 @@ pub(crate) struct IdSpace { /// What `provisional` and `dict` hold (#549 spec §3.2). held_provisional: Held, held_dict: Held, + /// What value probes read once for the whole query (#721). + pub(crate) probes: super::scan::Probes, } #[derive(Default)] @@ -64,6 +66,7 @@ impl IdSpace { dict: Maps::default(), held_provisional: Held::new(meter, "IdSpace"), held_dict: Held::new(meter, "IdSpace"), + probes: super::scan::Probes::new(meter), } } @@ -78,6 +81,7 @@ impl IdSpace { self.end_pull(); self.provisional = Maps::default(); self.held_provisional.release(); + self.probes = super::scan::Probes::new(self.held_provisional.meter()); } /// Count an entry of `bytes` about to go into the maps that hold diff --git a/src/query/engine/join.rs b/src/query/engine/join.rs index ffd6f65a5..6ff6401cb 100644 --- a/src/query/engine/join.rs +++ b/src/query/engine/join.rs @@ -31,6 +31,9 @@ pub(crate) struct Join { state: State, /// The build side's table (#549 spec §3.2). held: Held, + /// A value probe that could not find a row's matches (#721): the rows + /// of the outer batch from that row on, joined by a hash join instead. + fallback: Option>, /// `rows_in` counts the outer rows received. counters: Counters, } @@ -81,7 +84,7 @@ enum Side { impl Join { pub(crate) fn new(outer: Box, spec: JoinSpec, meter: &Arc) -> Self { let op = match spec.kind { - JoinKind::Probe(_) | JoinKind::ProbeValue(_) => "IndexNestedLoopJoin", + JoinKind::Probe(_) | JoinKind::ProbeValue { .. } => "IndexNestedLoopJoin", JoinKind::Hash { .. } => "HashJoin", }; Join { @@ -94,6 +97,7 @@ impl Join { outer_width: spec.outer_width, state: State::Start, held: Held::new(meter, op), + fallback: None, counters: Counters::new(op), } } @@ -151,10 +155,14 @@ impl Join { })) } - /// Read all of `side` into a table. - fn build(&mut self, side: Side, ctx: &mut Ctx<'_>) -> Result { + /// Read `first`, then all of `side`, into a table. + fn build(&mut self, side: Side, first: Vec, ctx: &mut Ctx<'_>) -> Result
{ let mut table = Table::default(); - while let Some(rows) = self.pull(side, ctx)? { + let mut first = Some(first); + while let Some(rows) = match first.take() { + Some(rows) => Some(rows), + None => self.pull(side, ctx)?, + } { for row in rows { let Some(key) = self.key(&row, side, ctx)? else { continue; @@ -269,7 +277,10 @@ impl Join { } else if by_value { match self.keys.get(probed).and_then(|k| o.get(k.outer)) { Some(cell) => { - self.inner.probe_value(cell, &self.held, ctx)?; + if !self.inner.probe_value(cell, ctx)? { + self.fallback = Some(p.rows.split_off(p.at)); + return Ok(true); + } p.probing = true; } None => p.at += 1, @@ -355,14 +366,14 @@ impl Operator for Join { let mut out = Batch::new(self.outer_width + self.new.len()); if let State::Start = self.state { self.state = match self.kind { - JoinKind::Probe(_) | JoinKind::ProbeValue(_) => State::Probe(Probe::default()), + JoinKind::Probe(_) | JoinKind::ProbeValue { .. } => State::Probe(Probe::default()), JoinKind::Hash { build_inner } => { let built = if build_inner { Side::Inner } else { Side::Outer }; - let table = self.build(built, ctx)?; + let table = self.build(built, Vec::new(), ctx)?; if table.buckets.is_empty() { State::Done } else { @@ -379,11 +390,39 @@ impl Operator for Join { } // Taken out while it runs; an error leaves the join done. let more = match (std::mem::replace(&mut self.state, State::Done), self.kind) { - (State::Probe(mut p), JoinKind::Probe(k) | JoinKind::ProbeValue(k)) => { - let by_value = matches!(self.kind, JoinKind::ProbeValue(_)); + (State::Probe(mut p), JoinKind::Probe(k) | JoinKind::ProbeValue { key: k, .. }) => { + let by_value = matches!(self.kind, JoinKind::ProbeValue { .. }); let more = self.next_probe(&mut p, k, by_value, &mut out, ctx)?; self.state = State::Probe(p); - more + match self.fallback.take() { + // For this row and the rest, the hash join the planner + // chose before #721, built on the side it named. It + // stays one for this operator. + Some(rows) => { + let build_inner = match self.kind { + JoinKind::ProbeValue { build_inner, .. } => build_inner, + _ => true, + }; + self.inner.rewind(&[]); + let (table, rows) = if build_inner { + (self.build(Side::Inner, Vec::new(), ctx)?, rows) + } else { + (self.build(Side::Outer, rows, ctx)?, Vec::new()) + }; + self.kind = JoinKind::Hash { build_inner }; + let mut h = Hash { + table, + rows, + at: 0, + bucket: None, + next: 0, + }; + let more = self.next_hash(&mut h, build_inner, &mut out, ctx)?; + self.state = State::Hash(h); + more + } + None => more, + } } (State::Hash(mut h), JoinKind::Hash { build_inner }) => { let more = self.next_hash(&mut h, build_inner, &mut out, ctx)?; diff --git a/src/query/engine/meter.rs b/src/query/engine/meter.rs index e1521066c..c3a8c0920 100644 --- a/src/query/engine/meter.rs +++ b/src/query/engine/meter.rs @@ -63,9 +63,9 @@ pub(crate) struct Held { } impl Held { - /// A new holder on the same meter, for the same operator, holding nothing. - pub(crate) fn sibling(&self) -> Self { - Held::new(&self.meter, self.op) + /// The meter this holder counts on. + pub(crate) fn meter(&self) -> &Arc { + &self.meter } pub(crate) fn new(meter: &Arc, op: &'static str) -> Self { @@ -113,6 +113,13 @@ impl Held { ); } + /// Take over what `other` counts, which goes on being counted until this + /// holder lets it go. + pub(crate) fn absorb(&mut self, mut other: Held) { + self.bytes = self.bytes.saturating_add(other.bytes); + other.bytes = 0; + } + /// Stop counting every byte of this holder. pub(crate) fn release(&mut self) { self.uncharge(self.bytes); diff --git a/src/query/engine/mod.rs b/src/query/engine/mod.rs index a87968d38..29b26c39d 100644 --- a/src/query/engine/mod.rs +++ b/src/query/engine/mod.rs @@ -191,11 +191,14 @@ fn resolve( /// Open the rule query `plan` over `snapshot`: derive its relations, with /// the magic seeds `seeds` as rows, then open its body (#548). An error from -/// the rules fails the query here. +/// the rules fails the query here. `keywords` are the keyword constants the +/// query's rule calls name as entities: value probes find their facts from +/// the entity (#721). pub(crate) fn open_rules( snapshot: Snapshot, plan: plan::RulesPlan, seeds: &[(EntityId, String, Value)], + keywords: &[String], functions: Arc>, limits: recursive::Limits, ) -> Result { @@ -210,6 +213,14 @@ pub(crate) fn open_rules( .iter() .map(|a| ids.iid(keys, a)) .collect::>>()?; + for k in keywords { + let e = crate::query::datalog::matcher::edn_to_entity_id( + &crate::query::datalog::types::EdnValue::Keyword(k.clone()), + ) + .map_err(|e| err_coded!(ErrorCode::Int049, format!("query engine: {e}")))?; + let (e, i) = (ids.eid(keys, &e)?, ids.iid(keys, k)?); + ids.probes.note(e, i)?; + } let mut rels = relation::Relations::new(iids, &meter); for (e, a, v) in seeds { if let Some(&rel) = plan.evaluated.by_attr.get(a) { diff --git a/src/query/engine/plan.rs b/src/query/engine/plan.rs index 0bfeec206..befb68d0a 100644 --- a/src/query/engine/plan.rs +++ b/src/query/engine/plan.rs @@ -253,10 +253,11 @@ pub(crate) enum JoinKind { /// entity range per outer row. Probe(usize), /// `IndexNestedLoopJoin` on the value (#721): the inner pattern's - /// attribute is a constant and its value is `keys[k]`; per outer row, + /// attribute is a constant and its value is `keys[key]`; per outer row, /// the AVET ranges of the values that join-equal it, and the pending - /// facts with its key. - ProbeValue(usize), + /// facts with its key. When a probe cannot find a row's matches, a + /// `HashJoin` built on the side `build_inner` names. + ProbeValue { key: usize, build_inner: bool }, /// `HashJoin`: a table of the side with the smaller estimate, probed by /// the other. Hash { build_inner: bool }, @@ -1010,7 +1011,7 @@ impl Builder { let mut join = join(&scanned, &attribute, slots, &rebind, valid)?; // Relations are indexed by entity only: a value join on // one hashes (#721). - if derived.is_some() && matches!(join.kind, JoinKind::ProbeValue(_)) { + if derived.is_some() && matches!(join.kind, JoinKind::ProbeValue { .. }) { join.kind = JoinKind::Hash { build_inner: true }; } join.inner.derived = derived; @@ -1025,7 +1026,9 @@ impl Builder { } metas.push((m, slot)); } - if let JoinKind::Hash { build_inner } = &mut join.kind { + if let JoinKind::Hash { build_inner } + | JoinKind::ProbeValue { build_inner, .. } = &mut join.kind + { // With the test switch off, always the inner pattern // (decision 3 of the #596 plan). After a seed and no // pattern, the seed (both costs of nothing are 0). @@ -1411,7 +1414,10 @@ fn join( }, kind: match (probe, probe_value) { (Some(k), _) => JoinKind::Probe(k), - (None, Some(k)) => JoinKind::ProbeValue(k), + (None, Some(k)) => JoinKind::ProbeValue { + key: k, + build_inner: true, + }, (None, None) => JoinKind::Hash { build_inner: true }, }, keys, diff --git a/src/query/engine/recursive.rs b/src/query/engine/recursive.rs index 207a64073..ea2e50641 100644 --- a/src/query/engine/recursive.rs +++ b/src/query/engine/recursive.rs @@ -22,6 +22,7 @@ use crate::query::datalog::aggregate::type_name; use crate::query::datalog::evaluator::{derived_facts_exceeded, iterations_exceeded}; use crate::query::datalog::functions::FunctionRegistry; use crate::query::datalog::matcher::{edn_to_entity_id, edn_to_value}; +use crate::query::datalog::types::EdnValue; use crate::storage::{KeySource, TxBound}; use anyhow::Result; use std::collections::HashSet; @@ -230,7 +231,16 @@ fn entity<'c>( } // As the executor reads a head (`evaluator.rs`). let k = match edn_to_entity_id(c) { - Ok(u) => Cell::Eid(ctx.ids.eid(ctx.keys, &u)?), + Ok(u) => { + let e = ctx.ids.eid(ctx.keys, &u)?; + // Value probes find this keyword's facts from the + // entity (#721). + if let EdnValue::Keyword(name) = c { + let i = ctx.ids.iid(ctx.keys, name)?; + ctx.ids.probes.note(e, i)?; + } + Cell::Eid(e) + } Err(_) => { let kind = edn_to_value(c).map_or("symbol", |v| type_name(&v)); return Err(err_coded!(ErrorCode::Qry013, predicate, kind)); @@ -241,7 +251,11 @@ fn entity<'c>( } HeadArg::Slot(s) => match cell(*s) { Some(Cell::Eid(e)) => return Ok(Cell::Eid(*e)), - Some(Cell::Iid(i)) => return Ok(Cell::Eid(ctx.ids.keyword_eid(ctx.keys, *i)?)), + Some(Cell::Iid(i)) => { + let e = ctx.ids.keyword_eid(ctx.keys, *i)?; + ctx.ids.probes.note(e, *i)?; + return Ok(Cell::Eid(e)); + } Some(Cell::Val(v)) => match v { // The executor reads a `nil` back as a symbol. Scalar::Null => "symbol", diff --git a/src/query/engine/scan.rs b/src/query/engine/scan.rs index 054536694..415ee3c09 100644 --- a/src/query/engine/scan.rs +++ b/src/query/engine/scan.rs @@ -12,7 +12,7 @@ use super::batch::Batch; use super::cell::{Cell, JoinKey, Scalar}; -use super::meter::{HASH_ENTRY_OVERHEAD, Held, key_bytes, value_bytes}; +use super::meter::{HASH_ENTRY_OVERHEAD, Held, Meter, key_bytes, value_bytes}; use super::plan::{Meta, ScanSpec, Term, Valid, ValueConst, names}; use super::relation::Range as Rows; use super::{BATCH_ROWS, Counters, Ctx, Operator}; @@ -36,25 +36,57 @@ pub(crate) struct Scan { state: State, keys_buf: Vec, facts_buf: Vec, - /// What a value probe reads besides its key ranges (#721), built on the - /// first one, and the join key of the current one. - values: Option, + /// What a value probe reads besides its key ranges (#721), from the + /// query's [`Probes`], and the join key of the current probe. + values: Option>, value_key: Option, /// `rows_in` counts the live keys and pending facts read. counters: Counters, } -/// The reads a value probe shares with every later one (#721), charged to -/// `held`. +/// What value probes read once for the whole query (#721), so that a rule +/// run in every round, or a body run per outer row, reads it once. +pub(crate) struct Probes { + /// Per attribute, its pending facts and the committed triples they + /// decide. + values: HashMap>, + /// Per attribute id, whether any of its live committed values is a + /// keyword. + keywords: HashMap, + /// The keyword that names an entity, where the query made the entity + /// from the keyword (a rule head, #548 spec §5.3). A keyword's entity + /// is a hash of its name, so no index leads back from the entity. + named: HashMap, + held: Held, +} + +impl Probes { + pub(crate) fn new(meter: &Arc) -> Self { + Probes { + values: HashMap::new(), + keywords: HashMap::new(), + named: HashMap::new(), + held: Held::new(meter, "IndexNestedLoopJoin"), + } + } + + /// Record that the keyword with id `iid` names the entity `eid`. + pub(crate) fn note(&mut self, eid: u64, iid: u64) -> Result<()> { + if let std::collections::hash_map::Entry::Vacant(v) = self.named.entry(eid) { + self.held + .charge(HASH_ENTRY_OVERHEAD + size_of::<(u64, u64)>())?; + v.insert(iid); + } + Ok(()) + } +} + +/// An attribute's reads for value probes. struct Values { - /// The attribute's live pending facts, by the join key of their value. + /// Its live pending facts, by the join key of their value. pending: HashMap>, - /// The committed triples a pending record of the attribute decides. + /// The committed triples a pending record of it decides. shadow: Arc>>, - /// The keyword idents among the attribute's committed values, by the - /// entity each names; read on the first probe by an entity. - named: Option>>, - held: Held, } enum State { @@ -160,21 +192,18 @@ impl Scan { /// committed facts come from the AVET ranges of the values equal to /// `cell` under the join rules, the pending ones from the attribute's /// facts with its key; the join checks every key, so these need only - /// hold every match. `held` is charged for what the probes share. - pub(crate) fn probe_value( - &mut self, - cell: &Cell, - held: &Held, - ctx: &mut Ctx<'_>, - ) -> Result<()> { + /// hold every match. `false`, and nothing to read, when the scan cannot + /// find them: `cell` is an entity no keyword the query met names, and + /// the attribute has committed keyword values, any of which may name it. + pub(crate) fn probe_value(&mut self, cell: &Cell, ctx: &mut Ctx<'_>) -> Result { self.probe = None; self.value_key = None; self.state = State::Done; let Term::Const(attribute) = &self.spec.a else { - return Ok(()); + return Ok(true); }; if self.spec.never() { - return Ok(()); + return Ok(true); } if self.consts.is_none() && let Some(keys) = ctx.keys @@ -182,17 +211,20 @@ impl Scan { self.consts = Some(consts(&self.spec, keys)?); } if self.values.is_none() { - self.values = Some(values(attribute, held.sibling(), ctx)?); + self.values = Some(values(attribute, ctx)?); } let Some(key) = JoinKey::of(cell, true, ctx)? else { - return Ok(()); + return Ok(true); }; let a = match (&self.consts, ctx.keys) { (Some(Some(Consts { a: Some(a), .. })), Some(_)) => Some(*a), _ => None, }; let ranges = match a { - Some(a) => self.value_ranges(cell, a, ctx)?, + Some(a) => match value_ranges(cell, a, ctx)? { + Some(r) => r, + None => return Ok(false), + }, None => Vec::new(), }; self.value_key = Some(key); @@ -211,67 +243,7 @@ impl Scan { shadow, }) }; - Ok(()) - } - - /// The AVET ranges of attribute `a` holding every committed value that - /// join-equals `cell`. - fn value_ranges( - &mut self, - cell: &Cell, - a: u32, - ctx: &mut Ctx<'_>, - ) -> Result)>> { - let Some(keys) = ctx.keys else { - return Ok(Vec::new()); - }; - let mut kvs = Vec::new(); - match cell { - // A keyword, and a ref to the entity it names. - Cell::Iid(i) => { - if let Some(di) = ctx.ids.dict_iid(*i) { - kvs.push(KeyValue::Keyword(di)); - } - let e = ctx.ids.keyword_eid(ctx.keys, *i)?; - if let Some(de) = ctx.ids.dict_eid(e) { - kvs.push(KeyValue::Ref(de)); - } - } - // A ref, and the keyword that names the entity. - Cell::Eid(e) => { - if let Some(de) = ctx.ids.dict_eid(*e) { - kvs.push(KeyValue::Ref(de)); - } - if let Some(v) = &mut self.values - && v.named.is_none() - { - v.named = Some(named(a, &mut v.held, ctx)?); - } - if let Some(idents) = self - .values - .as_ref() - .and_then(|v| v.named.as_ref()) - .and_then(|n| n.get(e)) - { - kvs.extend(idents.iter().map(|&i| KeyValue::Keyword(i))); - } - } - // `-0.0` equals `0.0`; NaN has no key and never gets here. - Cell::Val(Scalar::Float(f)) => { - kvs.push(KeyValue::Float(*f)); - if *f == 0.0 { - kvs.push(KeyValue::Float(-*f)); - } - } - Cell::Val(_) => { - let v = super::project::value(cell, ctx)?; - kvs.extend(key_value(keys, &v)?); - } - } - Ok(kvs - .iter() - .map(|v| (Index::Avet, keys::attribute_value_prefix(a, v))) - .collect()) + Ok(true) } /// After the committed ranges: a value probe's pending facts, or the @@ -286,8 +258,13 @@ impl Scan { } /// The live pending facts of `attribute` by the join key of their value, -/// and the committed triples its pending records decide (#721). -fn values(attribute: &str, mut held: Held, ctx: &mut Ctx<'_>) -> Result { +/// and the committed triples its pending records decide (#721): read once +/// per query. +fn values(attribute: &str, ctx: &mut Ctx<'_>) -> Result> { + if let Some(v) = ctx.ids.probes.values.get(attribute) { + return Ok(v.clone()); + } + let mut held = Held::new(ctx.ids.probes.held.meter(), "IndexNestedLoopJoin"); let mut pending: HashMap> = HashMap::new(); let mut resume = None; let mut buf = Vec::new(); @@ -331,49 +308,94 @@ fn values(attribute: &str, mut held: Held, ctx: &mut Ctx<'_>) -> Result } } } - Ok(Values { + let v = Arc::new(Values { pending, shadow: Arc::new(shadow), - named: None, - held, - }) + }); + held.charge(HASH_ENTRY_OVERHEAD + size_of::<(String, Arc)>() + attribute.len())?; + let probes = &mut ctx.ids.probes; + probes.held.absorb(held); + probes.values.insert(attribute.to_string(), v.clone()); + Ok(v) } -/// The keyword idents among the live committed values of attribute `a`, by -/// the entity each names (#721). -fn named(a: u32, held: &mut Held, ctx: &mut Ctx<'_>) -> Result>> { - let mut by_entity: HashMap> = HashMap::new(); +/// Key ranges as `(index, prefix)`. +type Ranges = Vec<(Index, Vec)>; + +/// The AVET ranges of attribute `a` holding every committed value that +/// join-equals `cell`, or `None` when they cannot be found (see +/// [`Scan::probe_value`]). +fn value_ranges(cell: &Cell, a: u32, ctx: &mut Ctx<'_>) -> Result> { let Some(keys) = ctx.keys else { - return Ok(by_entity); + return Ok(Some(Vec::new())); }; - let prefix = keys::attribute_kind_prefix(a, ValueKind::Keyword); - let mut seen = HashSet::new(); - let mut resume: Option> = None; - let mut buf = Vec::new(); - loop { - buf.clear(); - resume = keys.live_keys( - Index::Avet, - &prefix, - resume.as_deref(), - ctx.bound, - BATCH_ROWS, - &mut buf, - )?; - for kf in &buf { - if let KeyValue::Keyword(i) = kf.v - && seen.insert(i) - { - let e = ctx.ids.keyword_eid(ctx.keys, u64::from(i))?; - held.charge(HASH_ENTRY_OVERHEAD + size_of::<(u64, Vec)>() + size_of::())?; - by_entity.entry(e).or_default().push(i); + let mut kvs = Vec::new(); + match cell { + // A keyword, and a ref to the entity it names. + Cell::Iid(i) => { + if let Some(di) = ctx.ids.dict_iid(*i) { + kvs.push(KeyValue::Keyword(di)); + } + let e = ctx.ids.keyword_eid(ctx.keys, *i)?; + if let Some(de) = ctx.ids.dict_eid(e) { + kvs.push(KeyValue::Ref(de)); } } - if resume.is_none() { - break; + // A ref, and the keyword that names the entity. + Cell::Eid(e) => { + if let Some(de) = ctx.ids.dict_eid(*e) { + kvs.push(KeyValue::Ref(de)); + } + match ctx.ids.probes.named.get(e).copied() { + Some(i) => kvs.extend(ctx.ids.dict_iid(i).map(KeyValue::Keyword)), + None if has_keywords(a, ctx)? => return Ok(None), + None => {} + } } + // `-0.0` equals `0.0`; NaN has no key and never gets here. + Cell::Val(Scalar::Float(f)) => { + kvs.push(KeyValue::Float(*f)); + if *f == 0.0 { + kvs.push(KeyValue::Float(-*f)); + } + } + Cell::Val(_) => { + let v = super::project::value(cell, ctx)?; + kvs.extend(key_value(keys, &v)?); + } + } + Ok(Some( + kvs.iter() + .map(|v| (Index::Avet, keys::attribute_value_prefix(a, v))) + .collect(), + )) +} + +/// Whether any live committed value of attribute `a` is a keyword: read +/// once per query. +fn has_keywords(a: u32, ctx: &mut Ctx<'_>) -> Result { + if let Some(&k) = ctx.ids.probes.keywords.get(&a) { + return Ok(k); } - Ok(by_entity) + let Some(keys) = ctx.keys else { + return Ok(false); + }; + let mut buf = Vec::new(); + keys.live_keys( + Index::Avet, + &keys::attribute_kind_prefix(a, ValueKind::Keyword), + None, + ctx.bound, + 1, + &mut buf, + )?; + let k = !buf.is_empty(); + let probes = &mut ctx.ids.probes; + probes + .held + .charge(HASH_ENTRY_OVERHEAD + size_of::<(u32, bool)>())?; + probes.keywords.insert(a, k); + Ok(k) } impl Operator for Scan { diff --git a/src/query/engine/tests.rs b/src/query/engine/tests.rs index 2700b2fbe..302c4fa99 100644 --- a/src/query/engine/tests.rs +++ b/src/query/engine/tests.rs @@ -862,7 +862,7 @@ fn planner_chooses_the_join() { .filter_map(|s| match s { Step::Join(j) => Some(match j.kind { JoinKind::Probe(_) => "probe".to_string(), - JoinKind::ProbeValue(_) => "probe value".to_string(), + JoinKind::ProbeValue { .. } => "probe value".to_string(), JoinKind::Hash { build_inner } => format!("hash inner={build_inner}"), }), _ => None, @@ -2286,7 +2286,7 @@ fn rule_delta_probes_a_bound_value() { for step in &run.plan.steps { let Step::Join(j) = step else { continue }; match j.kind { - JoinKind::ProbeValue(_) => { + JoinKind::ProbeValue { .. } => { assert!( j.inner.derived.is_none(), "a relation is not probed by value" @@ -2318,6 +2318,27 @@ fn rule_delta_probes_a_bound_value() { .unwrap(); assert_eq!(rows(&db, "(query [:find ?b :where (boss :p99 ?b)])"), 2); assert_eq!(rows(&db, "(query [:find ?b :where (boss :p199 ?b)])"), 3); + + // On a file, half checkpointed: the probes read committed keyword + // values through the keywords the rule heads named, and pending ones. + let dir = tempfile::tempdir().unwrap(); + let db = Minigraf::open(dir.path().join("tree.graph")).unwrap(); + let reports = |odd: usize| -> String { + (1..200) + .filter(|i| i % 2 == odd) + .map(|i| format!("[:p{i} :manager :p{}]", i / 10)) + .collect() + }; + db.execute(&format!("(transact [{}])", reports(0))).unwrap(); + db.checkpoint().unwrap(); + db.execute(&format!("(transact [{}])", reports(1))).unwrap(); + db.execute("(rule [(boss ?x ?y) [?x :manager ?y]])") + .unwrap(); + db.execute("(rule [(boss ?x ?z) [?x :manager ?y] (boss ?y ?z)])") + .unwrap(); + assert_eq!(rows(&db, "(query [:find ?b :where (boss :p99 ?b)])"), 2); + assert_eq!(rows(&db, "(query [:find ?b :where (boss :p199 ?b)])"), 3); + assert_eq!(rows(&db, "(query [:find ?b :where (boss :p198 ?b)])"), 3); } /// #721: a join on a bound value probes the AVET ranges of the values that @@ -2725,7 +2746,7 @@ fn planner_builds_a_keyless_join_on_the_smaller_product() { .filter_map(|s| match s { Step::Join(j) => match j.kind { JoinKind::Hash { build_inner } => Some(build_inner), - JoinKind::Probe(_) | JoinKind::ProbeValue(_) => None, + JoinKind::Probe(_) | JoinKind::ProbeValue { .. } => None, }, _ => None, }) diff --git a/tests/answers/engine_tests.txt b/tests/answers/engine_tests.txt index 468d67ecb..5d9f383f6 100644 --- a/tests/answers/engine_tests.txt +++ b/tests/answers/engine_tests.txt @@ -4945,6 +4945,7 @@ not_bodies_plan_as_clause_lists 07a2cd1cfc45bcc6 2 54e8aae90ee45459 not_bodies_plan_as_clause_lists 92ae19a63d955c9d 3 334b7473ff8bc18a not_bodies_plan_as_clause_lists 9c4ccbd1c930e739 3 334b7473ff8bc18a planner_errors_keep_their_codes 8d2f701fd7f4a458 1 d800c42686702fe9 +rule_delta_probes_a_bound_value 0650b7d1d80b9354 3 c49c7d5bab6f3ac5 rule_delta_probes_a_bound_value 7aac9bf2085f7f70 2 504c6f55a0f9d00e rule_delta_probes_a_bound_value a2bb83359e06a257 3 c49c7d5bab6f3ac5 rule_delta_runs_join_on_a_bound_variable 3a840c32501fd782 110 78d43ae4a90656ea From 99cc9695531af20291146a7245cf80a219cca0f2 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Fri, 23 Oct 2026 22:28:33 +0530 Subject: [PATCH 4/7] bench: recursion/ancestors; docs for value probes (#721) The nightly recursion group gains the ancestors shape, whose rounds join on a value, next to fanout. CLAUDE.md and TEST_COVERAGE describe value probes; the plan records what measuring changed. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0169pNQsLcCEn48te9X9WuZW --- CLAUDE.md | 4 +-- benches/helpers/mod.rs | 33 +++++++++++++++++++ benches/minigraf_bench.rs | 18 ++++++++++ docs/TEST_COVERAGE.md | 5 +-- .../plans/2026-10-23-721-value-probes.md | 16 +++++++++ 5 files changed, 72 insertions(+), 4 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 6dde6b598..cbad3424c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -147,7 +147,7 @@ cargo run < demos/demo_negation.txt - `prepared.rs`: `BindValue`, `PreparedQuery` — parse-once/execute-many with named `$slot` bind slots; `prepare_query`, `substitute` - `magic_sets.rs`: Magic-sets rewriting for demand-driven recursive rule evaluation (not applied to rules with `not`/`not-join`) -3a. **`src/query/engine/`** — Streaming query engine (#473), runs every query (#551): `plan.rs` (the operator tree of every query, or the executor's error code for it: patterns, expression clauses, `not`/`not-join` and `or`/`or-join`, and rule invocations, read as patterns of their relations (`plan_rules`, #548; a magic-set head constant is `HeadArg::Const`), in `optimizer::plan` order, then the `or`s in text order, then the deferred `not`s; an `or` branch or a `not` body is planned as a `:where` is; a clause list with no pattern first starts from `Seed`; a constant no cell equals is `Term::Never`, a scan that reads nothing; `:as-of` resolved when the query opens (`Snapshot::tx_bound`: a counter bound when `tx_id` rises with `tx_count`, else `TxBound::Timestamp`: committed `tx_count` ranges and pending `tx_id <= t`); a per-fact pseudo-attribute pattern reads the metadata column its source's scan fills (the hidden variable the parser named, #594), and `:db/valid-at` joins as `[e _ _]`; plain `:find` variables, aggregates with `:with`, and window functions, over an aggregate's groups too), `scan.rs` (one pattern over committed key ranges via `CommittedReader::key_source`/`live_keys`, AVET ranges for cross-kind constants, then the pending range by whole runs; one entity per probe for a join), `join.rs` (`IndexNestedLoopJoin` when the pattern's entity is bound, `HashJoin` otherwise; `JoinKey`: floats by IEEE `==`, a keyword equals a ref to the entity it names), `filter.rs` (`Filter`/`Bind` and pseudo-attribute steps over batches; a `Bind` on a bound slot keeps the rows where the result is join-equal, #596), `anti.rs` (`AntiJoin`, #546: a body with the same rows for every outer row runs once into a hash set of its imported-variable keys; any other body, or one whose first pattern's entity is imported, runs per outer row from `seed.rs`'s one-row `Seed`, via `Operator::rewind`; never deduplicates), `aggregate.rs` (`Aggregate`, #547: a hash table of groups keyed on cells as `Value`'s `Ord` compares values, one entry when ungrouped; `count`/`count-distinct` on cells, other aggregates on values via `datalog::aggregate::Agg`; groups sorted by value and finished in that order; a float group shows its least value by `value_order`), `window.rs` (`Window`, #547: holds each input row's passed-on cells and its named variables' values, then computes every window with `datalog::aggregate::windows`, the executor's code, and returns the rows in input order), `union.rs`/`distinct.rs` (`Union` and `Distinct`, #546: each `or` branch planned as a `not` body is, hashed into a table or run per outer row; then one row per distinct binding, as the executor's `Value`s compare), `cell.rs`/`batch.rs` (id cells in columnar batches of 1,024), `ids.rs` (DICT and provisional ids, `keyword_eid`), `project.rs` (ids → values for returned rows only), `relation.rs`/`recursive.rs` (#548: rule queries; a relation per evaluated predicate holds its derived rows, a set by `encode_value` identity, appended per round so `Old`/`Delta`/`Full` are ranges; `Recursive` runs each stratum's semi-naive rounds, one run per same-stratum top-level invocation, a rule reading its stratum otherwise run in full every round; scans read derived rows after committed and pending ones, with `nil` metadata), `meter.rs` (#549: one `Meter` per query, shared by every holder of state through a `Held` — hash join build sides, `not`/`or` tables, distinct sets, aggregate groups, window rows, rule relations and their temporaries, `IdSpace`; each entry charged by a per-entry estimate before it is inserted, containers grown with `try_reserve`, released on `rewind`/`close`/drop; over `OpenOptions::query_memory_limit`, or a refused allocation, is `QRY-012`; `Stream::close` drops the operator tree, keeping its counters). `mod.rs`: `Operator` (`next_batch`/`rewind`/`close`/`counters`), `Stream` (the operator tree over an owned `Snapshot`, pulled one batch at a time). `DatalogExecutor::open` / `execute_query` run every query on it: a cursor gets the `Stream` (`Answer::Stream`), `execute` drains it. +3a. **`src/query/engine/`** — Streaming query engine (#473), runs every query (#551): `plan.rs` (the operator tree of every query, or the executor's error code for it: patterns, expression clauses, `not`/`not-join` and `or`/`or-join`, and rule invocations, read as patterns of their relations (`plan_rules`, #548; a magic-set head constant is `HeadArg::Const`), in `optimizer::plan` order, then the `or`s in text order, then the deferred `not`s; an `or` branch or a `not` body is planned as a `:where` is; a clause list with no pattern first starts from `Seed`; a constant no cell equals is `Term::Never`, a scan that reads nothing; `:as-of` resolved when the query opens (`Snapshot::tx_bound`: a counter bound when `tx_id` rises with `tx_count`, else `TxBound::Timestamp`: committed `tx_count` ranges and pending `tx_id <= t`); a per-fact pseudo-attribute pattern reads the metadata column its source's scan fills (the hidden variable the parser named, #594), and `:db/valid-at` joins as `[e _ _]`; plain `:find` variables, aggregates with `:with`, and window functions, over an aggregate's groups too), `scan.rs` (one pattern over committed key ranges via `CommittedReader::key_source`/`live_keys`, AVET ranges for cross-kind constants, then the pending range by whole runs; one entity per probe for a join), `join.rs` (`IndexNestedLoopJoin` when the pattern's entity is bound, or, under a constant attribute and entity not constant, its value (#721: per outer row the AVET ranges of the values that join-equal it and the attribute's pending facts with its key, read once per query in `IdSpace::probes`; an entity cell needs the keyword naming it, known only where the query made the entity from a keyword (rule heads, magic seeds), else the join becomes the planner's `HashJoin`); `HashJoin` otherwise, and on a relation; `JoinKey`: floats by IEEE `==`, a keyword equals a ref to the entity it names), `filter.rs` (`Filter`/`Bind` and pseudo-attribute steps over batches; a `Bind` on a bound slot keeps the rows where the result is join-equal, #596), `anti.rs` (`AntiJoin`, #546: a body with the same rows for every outer row runs once into a hash set of its imported-variable keys; any other body, or one whose first pattern's entity is imported, runs per outer row from `seed.rs`'s one-row `Seed`, via `Operator::rewind`; never deduplicates), `aggregate.rs` (`Aggregate`, #547: a hash table of groups keyed on cells as `Value`'s `Ord` compares values, one entry when ungrouped; `count`/`count-distinct` on cells, other aggregates on values via `datalog::aggregate::Agg`; groups sorted by value and finished in that order; a float group shows its least value by `value_order`), `window.rs` (`Window`, #547: holds each input row's passed-on cells and its named variables' values, then computes every window with `datalog::aggregate::windows`, the executor's code, and returns the rows in input order), `union.rs`/`distinct.rs` (`Union` and `Distinct`, #546: each `or` branch planned as a `not` body is, hashed into a table or run per outer row; then one row per distinct binding, as the executor's `Value`s compare), `cell.rs`/`batch.rs` (id cells in columnar batches of 1,024), `ids.rs` (DICT and provisional ids, `keyword_eid`), `project.rs` (ids → values for returned rows only), `relation.rs`/`recursive.rs` (#548: rule queries; a relation per evaluated predicate holds its derived rows, a set by `encode_value` identity, appended per round so `Old`/`Delta`/`Full` are ranges; `Recursive` runs each stratum's semi-naive rounds, one run per same-stratum top-level invocation, a rule reading its stratum otherwise run in full every round; scans read derived rows after committed and pending ones, with `nil` metadata), `meter.rs` (#549: one `Meter` per query, shared by every holder of state through a `Held` — hash join build sides, `not`/`or` tables, distinct sets, aggregate groups, window rows, rule relations and their temporaries, `IdSpace`; each entry charged by a per-entry estimate before it is inserted, containers grown with `try_reserve`, released on `rewind`/`close`/drop; over `OpenOptions::query_memory_limit`, or a refused allocation, is `QRY-012`; `Stream::close` drops the operator tree, keeping its counters). `mod.rs`: `Operator` (`next_batch`/`rewind`/`close`/`counters`), `Stream` (the operator tree over an owned `Snapshot`, pulled one batch at a time). `DatalogExecutor::open` / `execute_query` run every query on it: a cursor gets the `Stream` (`Answer::Stream`), `execute` drains it. 4. **`src/temporal.rs`** — UTC-only timestamp parsing (avoids chrono CVE GHSA-wcg3-cvx6-7396) @@ -212,7 +212,7 @@ Auto-migrates v7 → v8 on open (spec §9, with a backup meta page). v1–v6 are ## Test Coverage -**1709 tests** (1697 passing, 12 ignored; unit + integration + doc). Engine answers once checked by the executor are frozen in `tests/answers/` (#551). +**1712 tests** (1700 passing, 12 ignored; unit + integration + doc). Engine answers once checked by the executor are frozen in `tests/answers/` (#551). See `docs/TEST_COVERAGE.md` for the full per-file breakdown. **Testing conventions** — see the Testing Conventions section below before writing any tests. diff --git a/benches/helpers/mod.rs b/benches/helpers/mod.rs index 918b115ba..582aadffb 100644 --- a/benches/helpers/mod.rs +++ b/benches/helpers/mod.rs @@ -154,6 +154,39 @@ pub fn fanout_graph(width: usize, depth: usize) -> Arc { Arc::new(db) } +/// In-memory DB with the fan-out tree of [`fanout_graph`] pointing up: each +/// node has `[:n :up :n]`, and `(anc ?x ?a)` is every +/// ancestor of `?x` (#721). Returns the database and the last leaf's id. +pub fn ancestor_tree(width: usize, depth: usize) -> (Arc, usize) { + let db = Minigraf::in_memory().unwrap(); + let mut edges: Vec<(usize, usize)> = Vec::new(); + let mut current_level = vec![0usize]; + let mut next_id = 1usize; + for _ in 0..depth { + let mut next_level = Vec::new(); + for &parent in ¤t_level { + for _ in 0..width { + edges.push((next_id, parent)); + next_level.push(next_id); + next_id += 1; + } + } + current_level = next_level; + } + for chunk in edges.chunks(200) { + let mut cmd = String::from("(transact ["); + for (child, parent) in chunk { + cmd.push_str(&format!("[:n{} :up :n{}]", child, parent)); + } + cmd.push_str("])"); + db.execute(&cmd).unwrap(); + } + db.execute("(rule [(anc ?x ?a) [?x :up ?a]])").unwrap(); + db.execute("(rule [(anc ?x ?a) [?x :up ?p] (anc ?p ?a)])") + .unwrap(); + (Arc::new(db), next_id - 1) +} + /// In-memory DB with n value facts AND n-1 chain reference facts. /// Total ≈ 2n facts. Used for join_3pattern benchmarks. /// diff --git a/benches/minigraf_bench.rs b/benches/minigraf_bench.rs index 97273a6cd..25c9e7378 100644 --- a/benches/minigraf_bench.rs +++ b/benches/minigraf_bench.rs @@ -323,6 +323,24 @@ fn bench_recursion(c: &mut Criterion) { ); group.finish(); } + + // ancestors: the same tree pointing up, from a leaf (#721). After the + // magic-set rewrite, each round joins `[?x :up ?p]` on its value. + { + let mut group = c.benchmark_group("recursion/ancestors"); + group.sample_size(10); + let (label, width, depth) = ("w10_d4", 10usize, 4usize); + group.bench_with_input( + BenchmarkId::from_parameter(label), + &(width, depth), + |b, &(width, depth)| { + let (db, leaf) = helpers::ancestor_tree(width, depth); + let q = format!("(query [:find ?a :where (anc :n{leaf} ?a)])"); + b.iter(|| db.execute(&q).unwrap()); + }, + ); + group.finish(); + } } // ── Task 8: open/ ───────────────────────────────────────────────────────────── diff --git a/docs/TEST_COVERAGE.md b/docs/TEST_COVERAGE.md index 2144525f3..d2e311a9e 100644 --- a/docs/TEST_COVERAGE.md +++ b/docs/TEST_COVERAGE.md @@ -4,9 +4,9 @@ | | Default features | With `--features test-hooks` | |---|---|---| -| **Passing** | 1,697 | 1,713 | +| **Passing** | 1,700 | 1,716 | | **Ignored** | 12 | 13 | -| **Total** | 1,709 | 1,726 | +| **Total** | 1,712 | 1,729 | **Verified**: 2026-10-23 with `cargo test --quiet`. @@ -414,6 +414,7 @@ These tests compare the database against an independent answer rather than a sin - Each in memory, on a file with pending facts and after a checkpoint, against their frozen answers. - Engine unit tests check routing (`:as-of` timestamps, a rule body with no pattern, a three-argument call and a write transaction) and derived rows read across batches before and after a checkpoint. `relation.rs` unit tests check row identity and the `Old` / `Delta` / `Full` ranges. - Every delta run of the magic-set rewrite of a recursive rule joins each clause on a variable bound before it, with no keyless join (`rule_delta_runs_join_on_a_bound_variable`, #694). An optimizer unit test checks `plan_after`'s connected order. + - A join on a bound value probes its AVET ranges and pending facts (#721): the delta run of the ancestors shape probes `[?x :manager ?y]` by value, in memory and on a half-checkpointed file (`rule_delta_probes_a_bound_value`); a value join reads only the probed keys, through a pending retraction (`value_probe_reads_only_the_probed_keys`); keyword and ref from either side, an entity cell no keyword was recorded for (the hash-join fallback), `0.0`/`-0.0`, long strings, `:as-of` and `:valid-at`, in memory, checkpointed and pending (`value_probe_matches_as_joins_do`). - **`:db/valid-at` in rule bodies** (`tests/recursive_rules_test.rs`, #617): 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. - **Rules calling `not` / `or` rules** (`tests/not_join_test.rs`, #628) - 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. diff --git a/docs/superpowers/plans/2026-10-23-721-value-probes.md b/docs/superpowers/plans/2026-10-23-721-value-probes.md index 877e5ed6c..d417e16b2 100644 --- a/docs/superpowers/plans/2026-10-23-721-value-probes.md +++ b/docs/superpowers/plans/2026-10-23-721-value-probes.md @@ -36,6 +36,22 @@ by entity only; a value probe was left out because the pending store has no valu 4. **No answer changes.** The frozen answers (`tests/answers/`), the plan-independence matrix and the model-based tests must pass unchanged. +## Deviations found while measuring + +5. **Shared reads are per query, not per scan.** A rule's runs rebuild their operators every + round, so per-scan pending maps reread the attribute every round (an in-memory database is all + pending). `IdSpace::probes` (`scan::Probes`) holds them for the whole query. +6. **No reverse map for entity cells (replaces decision 2's entity case).** A keyword's entity is + a hash of its name, so finding the keyword that names an entity meant reading the attribute's + whole keyword range: `recursive_ancestors` stayed at 123 ms (1M) and `join_3_scan` grew to + 15.6 s and 681 MiB at 10M. Instead the query records the keywords it makes entities from (rule + heads, and the keyword constants its rule calls name, which become the magic seeds). A probe by + an entity with no recorded keyword, on an attribute with committed keyword values, turns the + join into the hash join the planner chose before #721, on the side it named + (`JoinKind::ProbeValue { key, build_inner }`), for that row and the rest. +7. **Criterion `recursion/ancestors`** joins the nightly `recursion` group, next to `fanout`, so a + change of this shape shows on Bencher. + ## Steps Each step builds and passes clippy on all targets and features and leaves the suite passing, From 442a85e4b45b5b1d61b96f849491043522fc021f Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Fri, 23 Oct 2026 23:34:19 +0530 Subject: [PATCH 5/7] fix(engine): pending value-probe reads by encoded value, one candidate list (#721) The attribute's pending facts are keyed by encode_value, so building the per-query map hashes no keyword names; each probe looks up the values that join-equal its cell, the same list that gives the committed AVET ranges. An entity cell with no recorded keyword falls back to the hash join when the attribute has pending keyword values too. CHANGELOG entry with the envelope and Criterion numbers. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0169pNQsLcCEn48te9X9WuZW --- CHANGELOG.md | 1 + .../plans/2026-10-23-721-value-probes.md | 7 + src/query/engine/ids.rs | 5 - src/query/engine/scan.rs | 128 +++++++++--------- 4 files changed, 71 insertions(+), 70 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8cb115d23..33e7fba12 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -61,6 +61,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **A float constant in a pattern matches equal floats only** (#706). A float written in a pattern's value position, or bound to a `$slot` of a prepared query, matched every stored float within an absolute `f64::EPSILON` (about `2.2e-16`), so `[?e :m/x 0.0]` returned an entity whose `:m/x` is `1e-17`, while `[?e :m/x ?v] [(= ?v 0.0)]` did not. Above 1.0 the epsilon is smaller than the gap between adjacent floats, so only floats smaller than about `2.2e-16` in magnitude were affected: every one of them matched `0.0`, and `1e-20` matched `5e-20`. Float constants now compare by IEEE `==`, as joins, predicates and the indexes do: `0.0` matches `-0.0` and nothing else changes. The fix covers patterns in queries, rule bodies and `not` / `or` bodies, on the executor and the streaming engine. Nothing stored was wrong. The same fix ships in v2.0.10. - **WAL entries of more than 1,000,000 facts replay** (#686). A WAL entry of more than 1,000,000 facts was taken as damage: replay stopped there, and the next open cut the WAL off at that entry, so after a crash, kill or close without a checkpoint that transaction and every later one in the WAL were lost, though each had committed. The writer had no such limit. Since #550 no v3 transaction writes such an entry, but a WAL left by v2.x can hold one. Replay now takes an entry's fact count as damage only when the bytes after it cannot hold that many facts. Affected every release since v0.5.0. Also fixed in v2.0.10. - **A recursive rule query no longer joins a rule's new facts with every magic-set fact** (#694). On the engine (#548), each round after the first runs a recursive rule from the facts the last round derived, then joins the rest of its body. The rest kept the order the planner chose for the whole body, so the next clause could share no variable with the new facts: after the magic-set rewrite of `(reach :n0 ?to)`, the body `(magic ?from) [?from :next ?mid] (reach ?mid ?to)` joined each new `reach` fact with every `magic` fact, then probed `:next` once per pair. The rest of the body is now planned with the new facts' variables bound, each next pattern sharing a variable with those bound before it when one can. Answers are unchanged. Nightly `recursion` benchmarks (bench profile): `fanout/w10_d3` 4.94 s → 22.6 ms (it was about 94 ms on the executor before #548), `chain/depth_100` 699 ms → 41.5 ms, `chain/depth_10` 1.15 ms → 0.68 ms. +- **A join on a bound value probes the index instead of reading the whole attribute** (#721). #694's fix ordered a recursive rule's delta run so that the next pattern shares a variable with the new facts, but when that variable is the pattern's value (`[?x :manager ?y]` after the delta `(boss ?y ?z)`) the join was a `HashJoin` holding every `:manager` fact, rebuilt every round: `(boss :leaf ?b)` over a management tree took 1.8 s at 1M facts and 35 s at 10M instead of 2–3 ms. Such a join, on a constant attribute, is now an `IndexNestedLoopJoin` on the value: per row, the AVET ranges of the values that join-equal it (a keyword and a ref to the entity it names, either way; `0.0` and `-0.0`) and the attribute's uncheckpointed facts with those values, read once per query. A keyword's entity is a hash of its name, so a row whose value is an entity finds the keyword that names it only where the query made the entity from that keyword (a rule head, or a keyword its rule calls name); otherwise, on an attribute with keyword values, the join becomes the hash join it was before. Answers are unchanged. Measured with the #394 envelope (bench machine, p50 and peak RSS at 100K / 1M / 10M facts): `recursive_ancestors` 100 ms / 1.80 s / 35.0 s → 1.1 / 1.4 / 1.9 ms, 610 MiB → 4 MiB at 10M; `join_3_selective` (`[:p :manager ?b] [?s :manager ?b] [?s :name ?n]`) 10.4 ms / 144 ms / 2.36 s → 0.21 / 0.20 / 0.20 ms; other shapes unchanged. Nightly `recursion` benchmarks (in memory, bench profile): the new `ancestors/w10_d4` 97.7 → 15.1 ms, `chain/depth_100` 41.4 → 20.8 ms, `fanout/w10_d3` 22.7 → 20.8 ms. In memory every fact is uncheckpointed, and the first value probe reads the attribute's uncheckpointed facts once per query (12 ms for 11,110 facts), so `ancestors` stays above its 0.27 ms before #694. - **`BrowserDb` errors carry their error code** (#673). The browser API (`@minigraf/browser`) rejected with the error's top-level message only, such as `Unknown command: nosuchcommand`, while every other API reports `[PRS-001] Unknown command: nosuchcommand`. Every `BrowserDb` rejection now goes through `MinigrafError`, so it starts with its code. Host errors that have no code yet (no IndexedDB, an import blob whose length is not a multiple of the page size) are unchanged. - **Prepared queries use the database's rule limit** (#549). `PreparedQuery::execute` and `query` ran a rule query with the default `max_derived_facts` of 1,000,000 whatever `OpenOptions` set; they now use the database's limit, as `execute` does, and its `query_memory_limit`. - **An `or` branch in a rule body knows the registered predicates** (#548). A rule with `or`/`or-join` evaluated its branches with the built-in functions only, so a predicate registered with `register_predicate` in a branch (`(or [(big? ?v)] [?x :special true])`) was unknown there. Branches now get the database's functions, as the rest of the rule body does. diff --git a/docs/superpowers/plans/2026-10-23-721-value-probes.md b/docs/superpowers/plans/2026-10-23-721-value-probes.md index d417e16b2..974973d26 100644 --- a/docs/superpowers/plans/2026-10-23-721-value-probes.md +++ b/docs/superpowers/plans/2026-10-23-721-value-probes.md @@ -51,6 +51,13 @@ by entity only; a value probe was left out because the pending store has no valu (`JoinKind::ProbeValue { key, build_inner }`), for that row and the rest. 7. **Criterion `recursion/ancestors`** joins the nightly `recursion` group, next to `fanout`, so a change of this shape shows on Bencher. +8. **Pending facts by encoded value, and one candidate list.** The pending map is keyed by + `encode_value` of each fact's value, not by its join key, so building it hashes no keyword + names; a probe looks up each value that join-equals the cell (`scan::candidates`), the same + list that gives the committed AVET ranges. An entity cell with no recorded keyword also falls + back when the attribute has pending keyword values. In memory, where every fact is pending, + the map is still one read of the attribute per query: `recursion/ancestors` 15 ms against + 0.27 ms before #694 (97.7 ms on `v3`). A pending value index would remove it; #545 left it out. ## Steps diff --git a/src/query/engine/ids.rs b/src/query/engine/ids.rs index abbd1057e..8305aebd8 100644 --- a/src/query/engine/ids.rs +++ b/src/query/engine/ids.rs @@ -214,11 +214,6 @@ impl IdSpace { (eid < self.dict_eids).then_some(eid) } - /// `iid` as a DICT ident id, or `None` for a provisional one. - pub(crate) fn dict_iid(&self, iid: u64) -> Option { - u32::try_from(iid).ok().filter(|_| iid < self.dict_iids) - } - /// The ident with id `iid`. pub(crate) fn name(&mut self, keys: Option<&dyn KeySource>, iid: u64) -> Result> { let maps = self.maps(iid >= self.dict_iids); diff --git a/src/query/engine/scan.rs b/src/query/engine/scan.rs index 415ee3c09..d33a4ae6d 100644 --- a/src/query/engine/scan.rs +++ b/src/query/engine/scan.rs @@ -11,13 +11,14 @@ //! row is visible at every time and has no metadata (`nil`). use super::batch::Batch; -use super::cell::{Cell, JoinKey, Scalar}; -use super::meter::{HASH_ENTRY_OVERHEAD, Held, Meter, key_bytes, value_bytes}; +use super::cell::{Cell, Scalar}; +use super::meter::{HASH_ENTRY_OVERHEAD, Held, Meter, value_bytes}; use super::plan::{Meta, ScanSpec, Term, Valid, ValueConst, names}; use super::relation::Range as Rows; use super::{BATCH_ROWS, Counters, Ctx, Operator}; use crate::graph::storage::PendingResume; use crate::graph::types::{EntityId, Fact, Value}; +use crate::storage::index::encode_value; use crate::storage::keys::{self, Index, KeyFact, KeyValue, SHORT_STRING_MAX, ValueKind}; use crate::storage::{KeySource, Scan as Range}; use anyhow::Result; @@ -39,7 +40,8 @@ pub(crate) struct Scan { /// What a value probe reads besides its key ranges (#721), from the /// query's [`Probes`], and the join key of the current probe. values: Option>, - value_key: Option, + /// The encoded values whose pending facts the current probe reads. + value_key: Option>>, /// `rows_in` counts the live keys and pending facts read. counters: Counters, } @@ -83,8 +85,10 @@ impl Probes { /// An attribute's reads for value probes. struct Values { - /// Its live pending facts, by the join key of their value. - pending: HashMap>, + /// Its live pending facts, by their encoded value. + pending: HashMap, Vec>, + /// Whether any of them has a keyword value. + keywords: bool, /// The committed triples a pending record of it decides. shadow: Arc>>, } @@ -213,21 +217,23 @@ impl Scan { if self.values.is_none() { self.values = Some(values(attribute, ctx)?); } - let Some(key) = JoinKey::of(cell, true, ctx)? else { - return Ok(true); - }; let a = match (&self.consts, ctx.keys) { (Some(Some(Consts { a: Some(a), .. })), Some(_)) => Some(*a), _ => None, }; - let ranges = match a { - Some(a) => match value_ranges(cell, a, ctx)? { - Some(r) => r, - None => return Ok(false), - }, - None => Vec::new(), + let pending_keywords = self.values.as_ref().is_some_and(|v| v.keywords); + let Some(candidates) = candidates(cell, a, pending_keywords, ctx)? else { + return Ok(false); }; - self.value_key = Some(key); + let mut ranges = Vec::new(); + if let (Some(a), Some(keys)) = (a, ctx.keys) { + for v in &candidates { + if let Some(kv) = key_value(keys, v)? { + ranges.push((Index::Avet, keys::attribute_value_prefix(a, &kv))); + } + } + } + self.value_key = Some(candidates.iter().map(encode_value).collect()); self.state = if ranges.is_empty() { self.bucket() } else { @@ -250,7 +256,13 @@ impl Scan { /// pending range. fn bucket(&self) -> State { match (&self.value_key, &self.values) { - (Some(k), Some(v)) => State::Bucket(v.pending.get(k).cloned().unwrap_or_default()), + (Some(ks), Some(v)) => State::Bucket( + ks.iter() + .filter_map(|k| v.pending.get(k)) + .flatten() + .cloned() + .collect(), + ), (Some(_), None) => State::Done, _ => State::Pending(None), } @@ -265,7 +277,8 @@ fn values(attribute: &str, ctx: &mut Ctx<'_>) -> Result> { return Ok(v.clone()); } let mut held = Held::new(ctx.ids.probes.held.meter(), "IndexNestedLoopJoin"); - let mut pending: HashMap> = HashMap::new(); + let mut pending: HashMap, Vec> = HashMap::new(); + let mut keywords = false; let mut resume = None; let mut buf = Vec::new(); loop { @@ -278,13 +291,12 @@ fn values(attribute: &str, ctx: &mut Ctx<'_>) -> Result> { &mut buf, )?; for f in buf.drain(..) { - let cell = Cell::of_value(ctx.ids, ctx.keys, &f.value)?; - let Some(k) = JoinKey::of(&cell, true, ctx)? else { - continue; - }; + keywords |= matches!(f.value, Value::Keyword(_)); + let k = encode_value(&f.value); held.charge( HASH_ENTRY_OVERHEAD - + key_bytes(&k) + + size_of::>() + + k.len() + size_of::() + f.entity.as_bytes().len() + f.attribute.len() @@ -310,6 +322,7 @@ fn values(attribute: &str, ctx: &mut Ctx<'_>) -> Result> { } let v = Arc::new(Values { pending, + keywords, shadow: Arc::new(shadow), }); held.charge(HASH_ENTRY_OVERHEAD + size_of::<(String, Arc)>() + attribute.len())?; @@ -319,56 +332,41 @@ fn values(attribute: &str, ctx: &mut Ctx<'_>) -> Result> { Ok(v) } -/// Key ranges as `(index, prefix)`. -type Ranges = Vec<(Index, Vec)>; - -/// The AVET ranges of attribute `a` holding every committed value that -/// join-equals `cell`, or `None` when they cannot be found (see -/// [`Scan::probe_value`]). -fn value_ranges(cell: &Cell, a: u32, ctx: &mut Ctx<'_>) -> Result> { - let Some(keys) = ctx.keys else { - return Ok(Some(Vec::new())); - }; - let mut kvs = Vec::new(); - match cell { - // A keyword, and a ref to the entity it names. +/// The values that join-equal `cell` (#721): a keyword and a ref to the +/// entity it names, either way; `0.0` and `-0.0`; nothing for NaN. `None` +/// when they cannot be found: `cell` is an entity no keyword the query met +/// names, and the attribute (`a`, when committed) has keyword values, any of +/// which may name it. +fn candidates( + cell: &Cell, + a: Option, + pending_keywords: bool, + ctx: &mut Ctx<'_>, +) -> Result>> { + Ok(Some(match cell { Cell::Iid(i) => { - if let Some(di) = ctx.ids.dict_iid(*i) { - kvs.push(KeyValue::Keyword(di)); - } let e = ctx.ids.keyword_eid(ctx.keys, *i)?; - if let Some(de) = ctx.ids.dict_eid(e) { - kvs.push(KeyValue::Ref(de)); - } + vec![ + Value::Keyword(ctx.ids.name(ctx.keys, *i)?.to_string()), + Value::Ref(ctx.ids.uuid(ctx.keys, e)?), + ] } - // A ref, and the keyword that names the entity. Cell::Eid(e) => { - if let Some(de) = ctx.ids.dict_eid(*e) { - kvs.push(KeyValue::Ref(de)); - } + let mut v = vec![Value::Ref(ctx.ids.uuid(ctx.keys, *e)?)]; match ctx.ids.probes.named.get(e).copied() { - Some(i) => kvs.extend(ctx.ids.dict_iid(i).map(KeyValue::Keyword)), - None if has_keywords(a, ctx)? => return Ok(None), - None => {} + Some(i) => v.push(Value::Keyword(ctx.ids.name(ctx.keys, i)?.to_string())), + None if pending_keywords => return Ok(None), + None => match a { + Some(a) if has_keywords(a, ctx)? => return Ok(None), + _ => {} + }, } + v } - // `-0.0` equals `0.0`; NaN has no key and never gets here. - Cell::Val(Scalar::Float(f)) => { - kvs.push(KeyValue::Float(*f)); - if *f == 0.0 { - kvs.push(KeyValue::Float(-*f)); - } - } - Cell::Val(_) => { - let v = super::project::value(cell, ctx)?; - kvs.extend(key_value(keys, &v)?); - } - } - Ok(Some( - kvs.iter() - .map(|v| (Index::Avet, keys::attribute_value_prefix(a, v))) - .collect(), - )) + Cell::Val(Scalar::Float(f)) if f.is_nan() => Vec::new(), + Cell::Val(Scalar::Float(f)) if *f == 0.0 => vec![Value::Float(*f), Value::Float(-*f)], + Cell::Val(_) => vec![super::project::value(cell, ctx)?], + })) } /// Whether any live committed value of attribute `a` is a keyword: read From 8bc9313a665d9e54873309f51900ff66901a0aa7 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Fri, 23 Oct 2026 23:36:31 +0530 Subject: [PATCH 6/7] test: a value probe that falls back mid-batch joins every row once (#721) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0169pNQsLcCEn48te9X9WuZW --- CLAUDE.md | 2 +- docs/TEST_COVERAGE.md | 6 +++--- src/query/engine/tests.rs | 31 +++++++++++++++++++++++++++++++ tests/answers/engine_tests.txt | 1 + 4 files changed, 36 insertions(+), 4 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index cbad3424c..356050f62 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -212,7 +212,7 @@ Auto-migrates v7 → v8 on open (spec §9, with a backup meta page). v1–v6 are ## Test Coverage -**1712 tests** (1700 passing, 12 ignored; unit + integration + doc). Engine answers once checked by the executor are frozen in `tests/answers/` (#551). +**1713 tests** (1701 passing, 12 ignored; unit + integration + doc). Engine answers once checked by the executor are frozen in `tests/answers/` (#551). See `docs/TEST_COVERAGE.md` for the full per-file breakdown. **Testing conventions** — see the Testing Conventions section below before writing any tests. diff --git a/docs/TEST_COVERAGE.md b/docs/TEST_COVERAGE.md index d2e311a9e..0e1e931ad 100644 --- a/docs/TEST_COVERAGE.md +++ b/docs/TEST_COVERAGE.md @@ -4,9 +4,9 @@ | | Default features | With `--features test-hooks` | |---|---|---| -| **Passing** | 1,700 | 1,716 | +| **Passing** | 1,701 | 1,717 | | **Ignored** | 12 | 13 | -| **Total** | 1,712 | 1,729 | +| **Total** | 1,713 | 1,730 | **Verified**: 2026-10-23 with `cargo test --quiet`. @@ -414,7 +414,7 @@ These tests compare the database against an independent answer rather than a sin - Each in memory, on a file with pending facts and after a checkpoint, against their frozen answers. - Engine unit tests check routing (`:as-of` timestamps, a rule body with no pattern, a three-argument call and a write transaction) and derived rows read across batches before and after a checkpoint. `relation.rs` unit tests check row identity and the `Old` / `Delta` / `Full` ranges. - Every delta run of the magic-set rewrite of a recursive rule joins each clause on a variable bound before it, with no keyless join (`rule_delta_runs_join_on_a_bound_variable`, #694). An optimizer unit test checks `plan_after`'s connected order. - - A join on a bound value probes its AVET ranges and pending facts (#721): the delta run of the ancestors shape probes `[?x :manager ?y]` by value, in memory and on a half-checkpointed file (`rule_delta_probes_a_bound_value`); a value join reads only the probed keys, through a pending retraction (`value_probe_reads_only_the_probed_keys`); keyword and ref from either side, an entity cell no keyword was recorded for (the hash-join fallback), `0.0`/`-0.0`, long strings, `:as-of` and `:valid-at`, in memory, checkpointed and pending (`value_probe_matches_as_joins_do`). + - A join on a bound value probes its AVET ranges and pending facts (#721): the delta run of the ancestors shape probes `[?x :manager ?y]` by value, in memory and on a half-checkpointed file (`rule_delta_probes_a_bound_value`); a value join reads only the probed keys, through a pending retraction (`value_probe_reads_only_the_probed_keys`); keyword and ref from either side, an entity cell no keyword was recorded for (the hash-join fallback), `0.0`/`-0.0`, long strings, `:as-of` and `:valid-at`, in memory, checkpointed and pending (`value_probe_matches_as_joins_do`); a fallback in the middle of a batch joins the rows before it, the rest of the batch and later batches once each (`value_probe_falls_back_mid_batch`). - **`:db/valid-at` in rule bodies** (`tests/recursive_rules_test.rs`, #617): 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. - **Rules calling `not` / `or` rules** (`tests/not_join_test.rs`, #628) - 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. diff --git a/src/query/engine/tests.rs b/src/query/engine/tests.rs index 302c4fa99..217ae72db 100644 --- a/src/query/engine/tests.rs +++ b/src/query/engine/tests.rs @@ -2435,6 +2435,37 @@ fn value_probe_matches_as_joins_do() { } } +/// #721: a value probe that meets an entity no recorded keyword names, in +/// the middle of a batch, finishes as a hash join: the rows probed before +/// it, the rest of that batch and every later batch each join once. +#[test] +fn value_probe_falls_back_mid_batch() { + // Six :tag values are refs, the others keywords, so probes by keyword + // run before the first ref; each names the :x of one :w fact. + let facts = |lo: usize, hi: usize| -> String { + (lo..hi) + .map(|i| { + let k = if i % 500 == 7 { + format!("#uuid \"{}\"", uuid_of(&format!(":x{i}"))) + } else { + format!(":x{i}") + }; + format!("[:s{i} :tag {k}] [:t{i} :w :x{i}]") + }) + .collect() + }; + let q = "(query [:find ?s ?t :where [?s :tag ?k] [?t :w ?k]])"; + let db = Minigraf::in_memory().unwrap(); + db.execute(&format!("(transact [{}])", facts(0, 3000))).unwrap(); + assert_eq!(rows(&db, q), 3000, "in memory"); + let dir = tempfile::tempdir().unwrap(); + let db = Minigraf::open(dir.path().join("mixed.graph")).unwrap(); + db.execute(&format!("(transact [{}])", facts(0, 2000))).unwrap(); + db.checkpoint().unwrap(); + db.execute(&format!("(transact [{}])", facts(2000, 3000))).unwrap(); + assert_eq!(rows(&db, q), 3000, "two thirds checkpointed"); +} + // ── #549: query memory accounting ────────────────────────────────────────── /// A checkpointed database of `n` entities, each with integers `:a` and diff --git a/tests/answers/engine_tests.txt b/tests/answers/engine_tests.txt index 5d9f383f6..fb2b03828 100644 --- a/tests/answers/engine_tests.txt +++ b/tests/answers/engine_tests.txt @@ -4960,6 +4960,7 @@ rule_query_routing 4fb26550baf3a6d1 1 75741c9011909650 rule_query_routing 57c18e0e4c2d3d59 2 02b1d1046853e1c4 rule_query_routing 9cdc76b397fe1ac5 3 2b9101d4b11ba754 rule_query_routing e3c9735c8586e200 0 cbf29ce484222325 +value_probe_falls_back_mid_batch a43724d3cbf8b581 3000 dc6af9a5f7ceefe9 value_probe_matches_as_joins_do 04701079619bafd4 2 b11858569b739dda value_probe_matches_as_joins_do 436327ca6645dc60 2 b11858569b739dda value_probe_matches_as_joins_do 565efa7a38dac1a4 2 ecfa7f988c5996fa From 2965c642987d25a272c90a444e574fa070795b76 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Fri, 23 Oct 2026 23:50:56 +0530 Subject: [PATCH 7/7] style: rustfmt (#721) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0169pNQsLcCEn48te9X9WuZW --- src/query/engine/tests.rs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/query/engine/tests.rs b/src/query/engine/tests.rs index 217ae72db..00bedf83e 100644 --- a/src/query/engine/tests.rs +++ b/src/query/engine/tests.rs @@ -2456,13 +2456,16 @@ fn value_probe_falls_back_mid_batch() { }; let q = "(query [:find ?s ?t :where [?s :tag ?k] [?t :w ?k]])"; let db = Minigraf::in_memory().unwrap(); - db.execute(&format!("(transact [{}])", facts(0, 3000))).unwrap(); + db.execute(&format!("(transact [{}])", facts(0, 3000))) + .unwrap(); assert_eq!(rows(&db, q), 3000, "in memory"); let dir = tempfile::tempdir().unwrap(); let db = Minigraf::open(dir.path().join("mixed.graph")).unwrap(); - db.execute(&format!("(transact [{}])", facts(0, 2000))).unwrap(); + db.execute(&format!("(transact [{}])", facts(0, 2000))) + .unwrap(); db.checkpoint().unwrap(); - db.execute(&format!("(transact [{}])", facts(2000, 3000))).unwrap(); + db.execute(&format!("(transact [{}])", facts(2000, 3000))) + .unwrap(); assert_eq!(rows(&db, q), 3000, "two thirds checkpointed"); }