adityamukho · Oct 14, 2026 · Oct 14, 2026 · Oct 14, 2026 · Oct 14, 2026
@@ -56,6 +56,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Internal
 
 - **Single-pattern queries run on a new streaming engine** (#473, part 2: #544, first PR). A query whose `:where` is one pattern and whose `:find` holds plain variables now runs on operators that read index keys in batches of 1,024 rows and turn ids into values only for the rows they return (#530, for these queries). A pattern with a constant attribute and value, `[?e :a C]`, reads the AVET range of `C` instead of every fact of the attribute (#518). Measured at 1M facts (#394 envelope, `[?e :name "…"]`): 344 ms → 20 µs and 142 → 3 MiB; an attribute scan of 166,666 rows 577 → 268 ms and 172 → 57 MiB; point and `:as-of` queries 10–30 % faster. Every other query, and reads inside a `WriteTransaction`, run on the existing executor as before. Results are unchanged: in tests, every query the engine runs is also run by the executor and the answers must agree, and CI runs the whole suite that way (`test-oracle` feature).
+- **Expressions and pseudo-attributes run on the streaming engine** (#473, part 2: #544, second PR). The engine now also runs expression clauses after its pattern (`[(> ?v 21)]`, `[(+ ?v 1) ?w]`, registered predicates), per-fact pseudo-attribute patterns on the pattern's entity (`:db/valid-from`, `:db/valid-to`, `:db/tx-count`, `:db/tx-id`), and keyword, UUID and float constants in the value position. A constant reads only the AVET ranges it can match: a keyword reads its own range and the range of refs to the entity it names, a UUID reads its ref range and the attribute's keywords, a float reads the attribute's floats. The executor's matching rules are kept (a keyword matches a ref to the entity it names, a UUID matches a keyword that names it, floats match within `f64::EPSILON`), and so are its errors (INT-026, QRY-007). `:db/valid-at`, which the executor matches against every fact of the entity, stays on the executor until joins (#545).
 - **Committed reads hide superseded history on index keys** (#379). A query's committed facts are now reduced to the ones that survive net-assert (each fact's assertion from its newest transaction, unless that transaction retracted it; #435) while walking the index keys, before any fact is built. Only surviving entries are decoded and translated through the dictionary. A transaction that retracted a triple hides everything older in it, and a long run of hidden entries is skipped with one `seek`. `:as-of N` point and attribute queries now take the same index path instead of reading every fact in the file. On a checkpointed file with 2,000 superseded values, a point query takes 1.03 ms instead of 4.61 ms, an attribute scan 1.24 ms instead of 5.87 ms, and an `:as-of` point query 23 µs instead of 5.6 ms (at depth 1). Results are unchanged: the executor still applies net-assert to committed and pending facts together, and a pending fact is always newer than a committed one.
 - **B+tree reads no longer follow leaf sibling links** (v8 storage format, PR 1 of 5; #374). Every index scan uses a cursor that walks down from the root and keeps the path, with a forward `seek` that climbs only as far as needed. This prepares for copy-on-write index pages, which cannot keep sibling links. The file format and query results are unchanged. Concurrent 100k-fact scans take 25–37 % less time and checkpointing after one fact at 10k facts 9 % less; single point lookups and attribute scans are 2–6 % slower (about 0.4 µs per lookup). A corrupted child pointer now fails with STG-013 (wrong page type) or INT-049 (cycle) instead of following a bad link.
 

@@ -139,7 +139,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), beside the executor until #551: `plan.rs` (which queries it runs: one pattern, plain `:find` variables, today), `scan.rs` (one pattern over a committed key range via `CommittedReader::key_source`/`live_keys`, then the pending range by whole runs), `cell.rs`/`batch.rs` (id cells in columnar batches of 1,024), `ids.rs` (DICT and provisional ids), `project.rs` (ids → values for returned rows only), `oracle.rs` (with `cfg(test)` or the `test-oracle` feature, every engine query also runs on the executor and must agree). `DatalogExecutor::execute_query` tries it first.
+3a. **`src/query/engine/`** — Streaming query engine (#473), beside the executor until #551: `plan.rs` (which queries it runs: one pattern, then per-fact pseudo-attribute patterns on its entity and expression clauses in `optimizer::plan` order, plain `:find` variables, today), `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), `filter.rs` (`Filter`/`Bind` and pseudo-attribute columns over batches), `cell.rs`/`batch.rs` (id cells in columnar batches of 1,024), `ids.rs` (DICT and provisional ids), `project.rs` (ids → values for returned rows only), `oracle.rs` (with `cfg(test)` or the `test-oracle` feature, every engine query also runs on the executor and must agree). `DatalogExecutor::execute_query` tries it first.
 
 4. **`src/temporal.rs`** — UTC-only timestamp parsing (avoids chrono CVE GHSA-wcg3-cvx6-7396)
 
@@ -204,7 +204,7 @@ Auto-migrates v7 → v8 on open (spec §9, with a backup meta page). v1–v6 are
 
 ## Test Coverage
 
-**1477 tests** (1467 passing, 10 ignored; unit + integration + doc).
+**1481 tests** (1471 passing, 10 ignored; unit + integration + doc).
 See `docs/TEST_COVERAGE.md` for the full per-file breakdown.
 
 **Testing conventions** — see the Testing Conventions section below before writing any tests.

@@ -131,7 +131,7 @@ let r2 = pq.execute(&[("tx", BindValue::TxCount(2)), ("entity", BindValue::Entit
 
 ```bash
 cargo run          # interactive Datalog REPL
-cargo test         # run 1477 tests
+cargo test         # run 1481 tests
 cargo run < demos/demo_recursive.txt   # recursive rules demo
 ```
 

@@ -189,6 +189,25 @@ for what it ships.
   attribute: keyword only; other constants → no rows, as `match_component` gives).
 - Oracle tests for each (the issue's "cross-kind matching rules pinned by oracle tests").
 
+### PR 2 deviations
+- `:db/valid-at` patterns stay on the executor. Its matcher binds `:db/valid-at` by
+  scanning every fact of the entity (rows are a bag, one per fact), so it is a join, not
+  a constant column (#545).
+- A pseudo-attribute pattern runs on the engine only if its entity is the scan
+  pattern's entity term (the same variable, or a constant naming the same entity), it
+  is planned after the scan pattern, and no expression rebinds that variable: the
+  executor's hidden-key fast path applies only then; otherwise it joins (#545).
+- Steps run in `optimizer::plan` order, and a binding expression overwrites a bound
+  variable as the executor does (`[(+ ?v 1) ?v]` is accepted by the parser).
+- `INT-026` and `QRY-007` are raised before the engine runs, by the executor's own
+  checks (`check_clauses`); the executor path keeps its order of errors.
+- `attribute_string_prefix` is not ported: an exact long string resolves its value ref
+  through the DICT and uses `attribute_value_prefix`, so nothing would call it.
+- With the attribute unbound, cross-kind constants read EAVT (the entity's prefix, or
+  the whole index) with a residual test, not VAET: a keyword or UUID constant also
+  matches keyword values, which VAET does not hold.
+- `KeySource::tx_id_of` added for `:db/tx-id` on committed rows (decision 5 listed it).
+
 ### 10. Lazy cursor (PR 3)
 - `Cursor` holds the operator tree and the snapshot; `next_batch(max_rows)` pulls
   internal batches until `max_rows` rows are ready; errors surface from `next_batch`

@@ -15,6 +15,39 @@ use anyhow::Result;
 use std::collections::HashMap;
 use std::sync::{Arc, RwLock};
 
+/// `INT-026`: per-fact pseudo-attributes require `:any-valid-time`.
+fn check_per_fact_pseudo_attrs(query: &DatalogQuery) -> Result<()> {
+    if query_uses_per_fact_pseudo_attr(query)
+        && !matches!(query.valid_at, Some(ValidAt::AnyValidTime))
+    {
+        return Err(err_coded!(ErrorCode::Int026));
+    }
+    Ok(())
+}
+
+/// `QRY-007`: surface unknown UDF predicates as errors before processing any
+/// rows (matches the behaviour of the former apply_expr_clauses post-pass).
+fn check_udf_predicates(query: &DatalogQuery, registry: &FunctionRegistry) -> Result<()> {
+    for clause in &query.where_clauses {
+        if let WhereClause::Expr {
+            expr: Expr::UnaryOp(UnaryOp::Udf(name), _),
+            ..
+        } = clause
+            && registry.get_predicate(name).is_none()
+        {
+            bail_coded!(ErrorCode::Qry007, name);
+        }
+    }
+    Ok(())
+}
+
+/// The clause checks of the executor's own path, for a query the streaming
+/// engine runs: its errors come first, from the same code (#473).
+fn check_clauses(query: &DatalogQuery, registry: &FunctionRegistry) -> Result<()> {
+    check_per_fact_pseudo_attrs(query)?;
+    check_udf_predicates(query, registry)
+}
+
 /// Returns true if any where clause (at any depth) contains a per-fact
 /// pseudo-attribute pattern (ValidFrom / ValidTo / TxCount / TxId).
 /// Used to enforce the `:any-valid-time` requirement.
@@ -514,8 +547,15 @@ impl DatalogExecutor {
         if self.facts_override.is_none()
             && let Some(plan) = crate::query::engine::plan(&query)
         {
+            let registry = self
+                .functions
+                .read()
+                .map_err(|_| err_coded!(ErrorCode::Qry008))?;
+            check_clauses(&query, &registry)?;
             let snapshot = self.read_snapshot()?;
-            if let Some(result) = crate::query::engine::run(&snapshot, plan) {
+            let result = crate::query::engine::run(&snapshot, plan, &registry);
+            drop(registry);
+            if let Some(result) = result {
                 #[cfg(any(test, feature = "test-oracle"))]
                 crate::query::engine::oracle::check(
                     &result,
@@ -556,12 +596,7 @@ impl DatalogExecutor {
             None => Value::Integer(now),
         };
 
-        // Hard-error: per-fact pseudo-attrs require :any-valid-time.
-        if query_uses_per_fact_pseudo_attr(&query)
-            && !matches!(query.valid_at, Some(ValidAt::AnyValidTime))
-        {
-            return Err(err_coded!(ErrorCode::Int026));
-        }
+        check_per_fact_pseudo_attrs(&query)?;
 
         // Apply temporal filters before pattern matching
         let filtered_facts = self.filter_facts_for_query(&query)?;
@@ -575,18 +610,7 @@ impl DatalogExecutor {
             .read()
             .map_err(|_| err_coded!(ErrorCode::Qry008))?;
 
-        // Pre-validate UDF predicate names: surface unknown predicates as errors before
-        // processing any rows (matches the behaviour of the former apply_expr_clauses post-pass).
-        for clause in &query.where_clauses {
-            if let WhereClause::Expr {
-                expr: Expr::UnaryOp(UnaryOp::Udf(name), _),
-                ..
-            } = clause
-                && registry.get_predicate(name).is_none()
-            {
-                bail_coded!(ErrorCode::Qry007, name);
-            }
-        }
+        check_udf_predicates(&query, &registry)?;
 
         // Collect Pattern, Expr, Not, and NotJoin top-level clauses for the planner.
         // Or/OrJoin are extracted separately below and applied as a post-pass.
@@ -733,12 +757,7 @@ impl DatalogExecutor {
             None => Value::Integer(now),
         };
 
-        // Hard-error: per-fact pseudo-attrs require :any-valid-time.
-        if query_uses_per_fact_pseudo_attr(&query)
-            && !matches!(query.valid_at, Some(ValidAt::AnyValidTime))
-        {
-            return Err(err_coded!(ErrorCode::Int026));
-        }
+        check_per_fact_pseudo_attrs(&query)?;
 
         // Apply temporal filters before evaluating recursive rules
         let filtered_facts = self.filter_facts_for_query(&query)?;

@@ -27,4 +27,21 @@ impl Batch {
         }
         self.len += 1;
     }
+
+    /// Keep the rows whose entry in `keep` is true.
+    pub(crate) fn retain(&mut self, keep: &[bool]) {
+        for col in &mut self.cols {
+            let mut k = keep.iter();
+            col.retain(|_| k.next().copied().unwrap_or(false));
+        }
+        self.len = keep.iter().filter(|&&k| k).count();
+    }
+
+    /// Set the column of `slot`: the next new slot, or one already filled.
+    pub(crate) fn set_col(&mut self, slot: usize, col: Vec<Cell>) {
+        match self.cols.get_mut(slot) {
+            Some(c) => *c = col,
+            None => self.cols.push(col),
+        }
+    }
 }
@@ -1,6 +1,10 @@
 //! Cells: the values operators exchange (#473 spec §5.2).
 
+use super::ids::IdSpace;
+use crate::graph::types::Value;
+use crate::storage::KeySource;
 use crate::storage::keys::{KeyValue, ValueRef};
+use anyhow::Result;
 use std::sync::Arc;
 
 /// One value in a row. Entities and keywords are ids in the query's
@@ -42,4 +46,21 @@ impl Cell {
             KeyValue::Ref(eid) => Cell::Eid(*eid),
         }
     }
+
+    /// The cell of a value, with ids from `ids`.
+    pub(crate) fn of_value(
+        ids: &mut IdSpace,
+        keys: Option<&dyn KeySource>,
+        value: &Value,
+    ) -> Result<Cell> {
+        Ok(match value {
+            Value::Null => Cell::Val(Scalar::Null),
+            Value::Boolean(b) => Cell::Val(Scalar::Bool(*b)),
+            Value::Integer(n) => Cell::Val(Scalar::Int(*n)),
+            Value::Float(f) => Cell::Val(Scalar::Float(*f)),
+            Value::String(s) => Cell::Val(Scalar::Str(Arc::from(s.as_str()))),
+            Value::Keyword(k) => Cell::Iid(ids.iid(keys, k)?),
+            Value::Ref(u) => Cell::Eid(ids.eid(keys, u)?),
+        })
+    }
 }
@@ -0,0 +1,82 @@
+//! `Filter` and `Bind` (#473 spec §6.1), and the pseudo-attribute patterns
+//! on a scanned fact: the clauses after the scan, one batch at a time.
+//!
+//! An expression sees only the variables it reads, turned into values for
+//! the row; a binding expression's result becomes a cell again.
+
+use super::Ctx;
+use super::batch::Batch;
+use super::cell::{Cell, Scalar};
+use super::plan::{Step, Term, Vars};
+use super::project::value;
+use crate::query::datalog::executor::{eval_expr, is_truthy};
+use crate::query::datalog::types::Expr;
+use anyhow::Result;
+use std::collections::HashMap;
+
+/// Apply `step` to `batch`.
+pub(crate) fn apply(step: &Step, mut batch: Batch, ctx: &mut Ctx<'_>) -> Result<Batch> {
+    match step {
+        Step::Pseudo { col, value } => {
+            let meta = batch.cols.get(*col).cloned().unwrap_or_default();
+            match value {
+                Term::Any => {}
+                Term::Bind(slot) => batch.set_col(*slot, meta),
+                Term::Const(n) => {
+                    let want = n.map(|n| Cell::Val(Scalar::Int(n)));
+                    let keep: Vec<bool> = meta.iter().map(|c| want.as_ref() == Some(c)).collect();
+                    batch.retain(&keep);
+                }
+                Term::Eq(slot) => {
+                    let other = batch.cols.get(*slot);
+                    let keep: Vec<bool> = meta
+                        .iter()
+                        .enumerate()
+                        .map(|(i, c)| other.and_then(|o| o.get(i)) == Some(c))
+                        .collect();
+                    batch.retain(&keep);
+                }
+            }
+        }
+        Step::Filter(expr, vars) => {
+            let mut keep = Vec::with_capacity(batch.len);
+            for i in 0..batch.len {
+                keep.push(eval(expr, vars, &batch, i, ctx)?.is_some_and(|v| is_truthy(&v)));
+            }
+            batch.retain(&keep);
+        }
+        Step::Bind(expr, vars, slot) => {
+            let mut keep = Vec::with_capacity(batch.len);
+            let mut out = Vec::with_capacity(batch.len);
+            for i in 0..batch.len {
+                match eval(expr, vars, &batch, i, ctx)? {
+                    Some(v) => {
+                        out.push(Cell::of_value(&mut ctx.ids, ctx.keys, &v)?);
+                        keep.push(true);
+                    }
+                    None => keep.push(false),
+                }
+            }
+            batch.retain(&keep);
+            batch.set_col(*slot, out);
+        }
+    }
+    Ok(batch)
+}
+
+/// `expr` over row `i`, or `None` where the executor's evaluation fails.
+fn eval(
+    expr: &Expr,
+    vars: &Vars,
+    batch: &Batch,
+    i: usize,
+    ctx: &mut Ctx<'_>,
+) -> Result<Option<crate::graph::types::Value>> {
+    let mut binding = HashMap::with_capacity(vars.len());
+    for (name, slot) in vars {
+        if let Some(cell) = batch.cols.get(*slot).and_then(|c| c.get(i)) {
+            binding.insert(name.clone(), value(cell, ctx)?);
+        }
+    }
+    Ok(eval_expr(expr, &binding, Some(ctx.functions)).ok())
+}
@@ -8,6 +8,7 @@
 
 mod batch;
 mod cell;
+mod filter;
 mod ids;
 #[cfg(any(test, feature = "test-oracle"))]
 pub(crate) mod oracle;
@@ -19,6 +20,7 @@ mod tests;
 
 use crate::graph::storage::Snapshot;
 use crate::query::datalog::executor::QueryResult;
+use crate::query::datalog::functions::FunctionRegistry;
 use crate::storage::KeySource;
 use anyhow::Result;
 use ids::IdSpace;
@@ -40,30 +42,53 @@ pub(crate) struct Ctx<'a> {
     /// database.
     keys: Option<&'a dyn KeySource>,
     ids: IdSpace,
+    /// Predicates for expression clauses.
+    functions: &'a FunctionRegistry,
 }
 
 /// Run `plan` over `snapshot`, or `None` when the snapshot's committed
 /// reader has no keys (a test reader) and the executor must run it.
-pub(crate) fn run(snapshot: &Snapshot, plan: Plan) -> Option<Result<QueryResult>> {
+pub(crate) fn run(
+    snapshot: &Snapshot,
+    plan: Plan,
+    functions: &FunctionRegistry,
+) -> Option<Result<QueryResult>> {
     let keys = match &snapshot.committed {
         Some(reader) => Some(reader.key_source()?),
         None => None,
     };
     #[cfg(test)]
     RUNS.with(|r| r.set(r.get() + 1));
-    Some(execute(snapshot, keys, plan))
+    Some(execute(snapshot, keys, plan, functions))
 }
 
-fn execute(snapshot: &Snapshot, keys: Option<&dyn KeySource>, plan: Plan) -> Result<QueryResult> {
-    let Plan { scan, find, vars } = plan;
+fn execute(
+    snapshot: &Snapshot,
+    keys: Option<&dyn KeySource>,
+    plan: Plan,
+    functions: &FunctionRegistry,
+) -> Result<QueryResult> {
+    let Plan {
+        scan,
+        steps,
+        find,
+        vars,
+    } = plan;
     let mut results = Vec::new();
     let mut ctx = Ctx {
         snapshot,
         keys,
         ids: IdSpace::new(keys),
+        functions,
     };
     let mut scan = scan::Scan::new(scan);
-    while let Some(batch) = scan.next_batch(&mut ctx)? {
+    while let Some(mut batch) = scan.next_batch(&mut ctx)? {
+        for step in &steps {
+            if batch.len == 0 {
+                break;
+            }
+            batch = filter::apply(step, batch, &mut ctx)?;
+        }
         results.extend(project::project(&batch, &find, &mut ctx)?);
     }
     Ok(QueryResult::QueryResults { vars, results })