feat(planner): accept every read plan against its checked query (RFC 0047)

changelog.d/bm25-scores-ignore-filters.fixed.md

@@ -0,0 +1,7 @@
+- A `bm25()` score no longer depends on the query's filters when rows were
+  written after the property's full-text index was built, or when the index
+  cannot confirm that it covers every row: such a scan now applies its filter
+  after scoring. See the [explain
+  guide][bm25-eligibility-explain].
+
+[bm25-eligibility-explain]: ../docs/user/queries/explain.md

changelog.d/plan-acceptance.added.md

@@ -0,0 +1,8 @@
+- Every read plan is checked against the query before it runs, and `explain`
+  reports what the check covered in a `validation` field: `exact_subset` for a
+  single-binding query with property filters, an optional leading `bm25()`,
+  property sort keys and a limit, whose every rewrite was checked, and
+  `invariants_only` for every other query. See the [explain
+  guide][plan-acceptance-explain].
+
+[plan-acceptance-explain]: ../docs/user/queries/explain.md

changelog.d/tie-break-declared-bindings-first.fixed.md

@@ -0,0 +1,6 @@
+- Rows that tie on every `order` key are now ordered by the identities of the
+  bindings the query declares before those of anonymous traversal endpoints,
+  so a `limit` cuts them as the declared bindings' order says. See the
+  [explain guide][tie-break-explain].
+
+[tie-break-explain]: ../docs/user/queries/explain.md

crates/omnigraph-compiler/src/ir/lower.rs

@@ -25,18 +25,28 @@ struct FreshNames {
     temp: usize,
 }
 
+const ANON_PREFIX: &str = "__anon_";
+const TEMP_PREFIX: &str = "__temp_";
+
 impl FreshNames {
     fn anon(&mut self) -> String {
         self.anon += 1;
-        format!("__anon_{}", self.anon)
+        format!("{ANON_PREFIX}{}", self.anon)
     }
 
     fn temp(&mut self, dst: &str) -> String {
         self.temp += 1;
-        format!("__temp_{}_{}", dst, self.temp)
+        format!("{TEMP_PREFIX}{}_{}", dst, self.temp)
     }
 }
 
+/// Whether `variable` is a name the lowering made up: an anonymous `_`
+/// endpoint or a cycle-closing or renamed temp. The query declares no such
+/// binding, so its identity orders nothing a reader can see.
+pub fn is_fresh_variable(variable: &str) -> bool {
+    variable.starts_with(ANON_PREFIX) || variable.starts_with(TEMP_PREFIX)
+}
+
 /// What every expression site shares: the physical column spellings and
 /// the bindings a variable names. Every emitted expression lowers through
 /// one, so `contains` resolution and `fold` apply everywhere.

crates/omnigraph-compiler/src/ir/mod.rs

@@ -5,6 +5,8 @@ pub(crate) mod lower;
 pub mod untyped;
 pub(crate) mod validate;
 
+pub use lower::is_fresh_variable;
+
 use std::collections::HashMap;
 
 use crate::error::{CompilerError, Result};

crates/omnigraph-compiler/src/lib.rs

@@ -44,6 +44,7 @@ pub const GQ_LANGUAGE_VERSION: (u16, u16) = (2, 2);
 pub use ir::lower::{lower_mutation_query, lower_query};
 pub use lint::{DiagnosticCode, Family, SafetyTier, Severity};
 pub use query::ast::Literal;
+pub use query::checked::CheckedRead;
 pub use query::descriptor::{
     QueryGraphFact, QueryGraphFactKind, QueryOperationDescriptor, QueryResultFieldDescriptor,
     QueryValueKind, describe_query_operation,

crates/omnigraph-compiler/src/query/checked.rs

@@ -0,0 +1,42 @@
+//! A read declaration the type checker accepted, kept beside its checked
+//! query so a later stage derives what the query requires from what it
+//! wrote (RFC 0047, "Requirements come from the checked query"). The checked
+//! query retains every typed expression; the declaration keeps the written
+//! spelling of the score calls a projection or the leading order key names.
+
+use std::sync::Arc;
+
+use crate::catalog::Catalog;
+use crate::error::Result;
+use crate::query::ast::QueryDecl;
+use crate::query::typecheck::{CheckedReadQuery, typecheck_query};
+
+/// One checked read declaration and the checked query the type checker
+/// built for it. Shared by `Arc`: a compiled query is cached and read by
+/// every door.
+#[derive(Debug, Clone)]
+pub struct CheckedRead {
+    decl: Arc<QueryDecl>,
+    query: Arc<CheckedReadQuery>,
+}
+
+impl CheckedRead {
+    /// Type-check `decl` against `catalog`, keeping both.
+    pub fn check(catalog: &Catalog, decl: &QueryDecl) -> Result<Self> {
+        let query = typecheck_query(catalog, decl)?;
+        Ok(Self {
+            decl: Arc::new(decl.clone()),
+            query: Arc::new(query),
+        })
+    }
+
+    /// The declaration as the query wrote it.
+    pub fn decl(&self) -> &QueryDecl {
+        &self.decl
+    }
+
+    /// The checked query the type checker built for [`Self::decl`].
+    pub fn query(&self) -> &CheckedReadQuery {
+        &self.query
+    }
+}

crates/omnigraph-compiler/src/query/mod.rs

@@ -1,4 +1,5 @@
 pub mod ast;
+pub mod checked;
 pub mod codes;
 pub mod descriptor;
 pub mod diagnostic;

crates/omnigraph-compiler/src/query/typecheck.rs

@@ -83,6 +83,19 @@ pub struct CheckedReadQuery {
     pub(crate) limit: Option<u64>,
 }
 
+impl CheckedReadQuery {
+    /// The bindings visible at the end of the top-level match scope.
+    pub fn bindings(&self) -> &HashMap<String, BoundVariable> {
+        &self.scope.bindings
+    }
+
+    /// The traversals of the top-level match scope, in source order;
+    /// a correlated block's traversals belong to its own scope.
+    pub fn traversals(&self) -> impl Iterator<Item = &ResolvedTraversal> {
+        self.scope.traversals()
+    }
+}
+
 /// A declared parameter and the type selected for its name (the last
 /// declaration of a repeated name wins, as the checker's lookup does).
 #[derive(Debug, Clone)]

crates/omnigraph-gqt-core/src/plan.rs

@@ -14,6 +14,7 @@
 //! scan Doc as $d: ranked bm25
 //! scan Doc as $d: ranked nearest fetch 10
 //! scan Doc as $d: ranked nearest fetch 10 nprobes 20
+//! scan Doc as $d: ranked bm25 eligibility after_scoring
 //! expand $d Knows $e: mode indexed_scan
 //! expand $d Knows $e: mode indexed_scan ran csr
 //! contains join $p.text contains $m.number
@@ -24,13 +25,15 @@
 //! hydrate $d: columns [body, title]
 //! pass projection_pushdown
 //! not pass aggregate_pushdown
+//! validation scope exact_subset
 //! ```
 //!
 //! A `ran` claim names the side of the node's declared switch that ran,
 //! read from the execution report of the same run (the last attempt of the
 //! node's row), joined to the explain row by the node's `id`.
 
 use omnigraph_compiler::catalog::Catalog;
+use omnigraph_planner::ValidationScope;
 use omnigraph_planner::optimizer::{
     PASS_ACCESS_PATH, PASS_ADDRESS_SHORT_CIRCUIT, PASS_AGGREGATE_PUSHDOWN, PASS_EXPAND_MODE,
     PASS_FRAGMENT_SCOPE, PASS_JOIN_ALGORITHM, PASS_LATE_MATERIALIZATION, PASS_PREDICATE_PUSHDOWN,
@@ -41,7 +44,7 @@ use serde_json::Value;
 
 use crate::report::Row;
 
-const FORMS: &str = "forms: `scan <Type>[ as $var]: columns [a, b]`, `scan <Type>[ as $var]: not columns [a, b]`, `scan <Type>[ as $var]: filter reads [v.a]`, `scan <Type>[ as $var]: no filter`, `scan <Type>[ as $var]: access id_lookup`, `scan <Type>[ as $var]: access sequential`, `scan <Type>[ as $var]: access index_probe <column>`, `hash join $var[ ran <hash_join|id_lookup>]`, `scan <Type>[ as $var]: ranked <nearest|bm25>[ fetch <n>][ nprobes <n>]`, `scan <Type>[ as $var]: runtime filter <column>`, `scan <Type>[ as $var]: no runtime filter`, `contains join $h.x contains $n.y`, `cross join $h.x contains $n.y`, `expand $src <Edge> $dst: mode <csr|indexed_scan>[ ran <csr|indexed_scan>]`, `filter reads [a.x, b.y]`, `sort tiebreak [$a.@id, $e.@type]`, `rank fuse row tiebreak [$e.@type, $e.@id]`, `expand $a $b: selection alternation [Knows out, Likes in]`, `sort no tiebreak`, `aggregate <column>: <func>([distinct ]<Type>) <accumulator> <overflow> -> <Type>`, `block aggregate <gq>: <func>(<Type>) <accumulator> <overflow> -> <Type>`, `result columns [<name>: <Type>, ...]`, `type <gq>: <Type>`, `cast <gq>: <Type> -> <Type>`, `no cast <gq>`, `hydrate $var: columns [a, b]`, `pass <name>`, `not pass <name>`";
+const FORMS: &str = "forms: `scan <Type>[ as $var]: columns [a, b]`, `scan <Type>[ as $var]: not columns [a, b]`, `scan <Type>[ as $var]: filter reads [v.a]`, `scan <Type>[ as $var]: no filter`, `scan <Type>[ as $var]: access id_lookup`, `scan <Type>[ as $var]: access sequential`, `scan <Type>[ as $var]: access index_probe <column>`, `hash join $var[ ran <hash_join|id_lookup>]`, `scan <Type>[ as $var]: ranked <nearest|bm25>[ fetch <n>][ nprobes <n>]`, `scan <Type>[ as $var]: runtime filter <column>`, `scan <Type>[ as $var]: no runtime filter`, `contains join $h.x contains $n.y`, `cross join $h.x contains $n.y`, `expand $src <Edge> $dst: mode <csr|indexed_scan>[ ran <csr|indexed_scan>]`, `filter reads [a.x, b.y]`, `sort tiebreak [$a.@id, $e.@type]`, `rank fuse row tiebreak [$e.@type, $e.@id]`, `expand $a $b: selection alternation [Knows out, Likes in]`, `sort no tiebreak`, `aggregate <column>: <func>([distinct ]<Type>) <accumulator> <overflow> -> <Type>`, `block aggregate <gq>: <func>(<Type>) <accumulator> <overflow> -> <Type>`, `result columns [<name>: <Type>, ...]`, `type <gq>: <Type>`, `cast <gq>: <Type> -> <Type>`, `no cast <gq>`, `hydrate $var: columns [a, b]`, `pass <name>`, `not pass <name>`, `validation scope <exact_subset|invariants_only>`";
 
 const ID_LOOKUP: &str = "id_lookup";
 const JOIN_SIDES: [&str; 2] = ["hash_join", "id_lookup"];
@@ -129,6 +132,8 @@ pub(crate) enum PlanLine {
         index: String,
         fetch: Option<u64>,
         nprobes: Option<u64>,
+        /// Where the scan applies its eligibility, when claimed.
+        eligibility: Option<String>,
     },
     /// Every selected physical scan is marked with a runtime filter on
     /// `column`, or with none when `column` is `None`.
@@ -190,6 +195,11 @@ pub(crate) enum PlanLine {
         name: String,
         negated: bool,
     },
+    /// Acceptance checked the plan to `scope` (`exact_subset` or
+    /// `invariants_only`), as the document's `validation` reports it.
+    ValidationScope {
+        scope: String,
+    },
 }
 
 #[derive(Debug, Default)]
@@ -241,6 +251,21 @@ pub(crate) fn parse_plan_body(body: &[(usize, &str)]) -> Result<Vec<PlanLine>, S
             });
             continue;
         }
+        if let Some(scope) = line.strip_prefix("validation scope ") {
+            let scope = scope.trim();
+            if ![
+                ValidationScope::ExactSubset.as_str(),
+                ValidationScope::InvariantsOnly.as_str(),
+            ]
+            .contains(&scope)
+            {
+                return Err(refused("names `exact_subset` or `invariants_only`"));
+            }
+            lines.push(PlanLine::ValidationScope {
+                scope: scope.to_string(),
+            });
+            continue;
+        }
         if let Some(gq) = line.strip_prefix("no cast ") {
             let gq = gq.trim();
             if gq.is_empty() {
@@ -585,18 +610,28 @@ pub(crate) fn parse_plan_body(body: &[(usize, &str)]) -> Result<Vec<PlanLine>, S
                 .ok_or_else(|| refused("claims `ranked nearest` or `ranked bm25`"))?;
             let mut fetch = None;
             let mut nprobes = None;
+            let mut eligibility = None;
             while let Some(key) = words.next() {
-                let count = words.next().and_then(|count| count.parse::<u64>().ok());
-                match (key, count) {
-                    ("fetch", Some(count)) if fetch.is_none() && nprobes.is_none() => {
+                let value = words.next();
+                let count = value.and_then(|count| count.parse::<u64>().ok());
+                match (key, count, value) {
+                    ("fetch", Some(count), _)
+                        if fetch.is_none() && nprobes.is_none() && eligibility.is_none() =>
+                    {
                         fetch = Some(count);
                     }
-                    ("nprobes", Some(count)) if nprobes.is_none() => {
+                    ("nprobes", Some(count), _) if nprobes.is_none() && eligibility.is_none() => {
                         nprobes = Some(count);
                     }
+                    ("eligibility", _, Some(placement))
+                        if eligibility.is_none()
+                            && ["before_scoring", "after_scoring"].contains(&placement) =>
+                    {
+                        eligibility = Some(placement.to_string());
+                    }
                     _ => {
                         return Err(refused(
-                            "claims `ranked <kind>[ fetch <n>][ nprobes <n>]`: each key optional, at most once, in that order, followed by a whole number",
+                            "claims `ranked <kind>[ fetch <n>][ nprobes <n>][ eligibility <before_scoring|after_scoring>]`: each key optional, at most once, in that order",
                         ));
                     }
                 }
@@ -607,6 +642,7 @@ pub(crate) fn parse_plan_body(body: &[(usize, &str)]) -> Result<Vec<PlanLine>, S
                 index: kind.to_string(),
                 fetch,
                 nprobes,
+                eligibility,
             });
             continue;
         } else if let Some(access) = claim.strip_prefix("access ") {
@@ -867,6 +903,7 @@ struct PlannedRanking {
     kind: String,
     fetch: Option<u64>,
     nprobes: Option<u64>,
+    eligibility: Option<String>,
 }
 
 /// `<choice>[ ran <choice>]` over the words `allowed`: the planned choice
@@ -1258,6 +1295,10 @@ fn planned_physical(node: &Value, out: &mut PlannedNodes) {
                     nprobes: ranked
                         .get("nprobes")
                         .map(|nprobes| nprobes.as_u64().unwrap_or(0)),
+                    eligibility: ranked
+                        .get("eligibility")
+                        .and_then(Value::as_str)
+                        .map(str::to_string),
                 }),
             runtime_filter: node
                 .get("runtime_filter")
@@ -1428,6 +1469,20 @@ pub(crate) fn plan_mismatch(
         .unwrap_or_default();
     for line in lines {
         match line {
+            PlanLine::ValidationScope { scope } => {
+                let reported = explain
+                    .get("validation")
+                    .and_then(|validation| validation.get("scope"))
+                    .and_then(Value::as_str);
+                if reported != Some(scope.as_str()) {
+                    return Some(format!(
+                        "expect plan: validation scope {scope}; the document reports {}",
+                        reported.map_or("no validation".to_string(), |reported| format!(
+                            "scope {reported}"
+                        ))
+                    ));
+                }
+            }
             PlanLine::Type { gq, ty } => {
                 let typed = match typed_nodes(&nodes.typed) {
                     Ok(nodes) => nodes,
@@ -1640,6 +1695,7 @@ pub(crate) fn plan_mismatch(
                 index: kind,
                 fetch,
                 nprobes,
+                eligibility,
             } => {
                 let selected = match physical_scans(&nodes, type_name, binding.as_deref()) {
                     Ok(selected) => selected,
@@ -1680,6 +1736,15 @@ pub(crate) fn plan_mismatch(
                         });
                     }
                 }
+                if let Some(eligibility) = eligibility {
+                    let placements: Vec<Option<&str>> =
+                        of_kind.iter().map(|r| r.eligibility.as_deref()).collect();
+                    if !placements.contains(&Some(eligibility.as_str())) {
+                        return Some(format!(
+                            "expect plan: the `{kind}` scans of `{type_name}` apply eligibility {placements:?}, expected {eligibility}"
+                        ));
+                    }
+                }
                 if let Some(nprobes) = nprobes {
                     let caps: Vec<Option<u64>> = of_kind.iter().map(|r| r.nprobes).collect();
                     if !caps.contains(&Some(*nprobes)) {
@@ -2451,6 +2516,51 @@ mod tests {
         assert_eq!(check(&lines, &explain()), None);
     }
 
+    /// The scope claim reads the document's `validation`; a different scope
+    /// or a document without one fails, and an unknown scope is refused.
+    #[test]
+    fn validation_scope_claims_read_the_documents_validation() {
+        let lines = parse_plan_body(&[(0, "validation scope invariants_only")]).unwrap();
+        assert_eq!(
+            lines[0],
+            PlanLine::ValidationScope {
+                scope: "invariants_only".to_string()
+            }
+        );
+        let accepted = json!({"validation": {"scope": "invariants_only"}});
+        assert_eq!(check(&lines, &accepted), None);
+        let exact = json!({"validation": {"scope": "exact_subset"}});
+        let mismatch = check(&lines, &exact).unwrap();
+        assert!(
+            mismatch.contains("reports scope exact_subset"),
+            "{mismatch}"
+        );
+        let mismatch = check(&lines, &explain()).unwrap();
+        assert!(mismatch.contains("reports no validation"), "{mismatch}");
+        assert!(parse_plan_body(&[(0, "validation scope everything")]).is_err());
+    }
+
+    /// The eligibility claim reads the ranked scan's declared placement.
+    #[test]
+    fn eligibility_claims_read_the_ranked_scans_placement() {
+        let explain = json!({"physical_plan": {"node": "Scan", "table": "node:Doc", "binding": "d",
+            "ranked": {"kind": "bm25", "fetch": null, "eligibility": "after_scoring"}}});
+        let lines =
+            parse_plan_body(&[(0, "scan Doc as $d: ranked bm25 eligibility after_scoring")])
+                .unwrap();
+        assert_eq!(check(&lines, &explain), None);
+        let lines =
+            parse_plan_body(&[(0, "scan Doc as $d: ranked bm25 eligibility before_scoring")])
+                .unwrap();
+        let mismatch = check(&lines, &explain).unwrap();
+        assert!(mismatch.contains("expected before_scoring"), "{mismatch}");
+        assert!(parse_plan_body(&[(0, "scan Doc: ranked bm25 eligibility sometimes")]).is_err());
+        assert!(
+            parse_plan_body(&[(0, "scan Doc: ranked bm25 eligibility after_scoring fetch 1")])
+                .is_err()
+        );
+    }
+
     #[test]
     fn a_read_column_fails_the_negative_form() {
         let lines = parse_plan_body(&[(0, "scan Doc: not columns [slug]")]).unwrap();
@@ -2945,6 +3055,7 @@ mod tests {
                 index: "nearest".to_string(),
                 fetch: Some(10),
                 nprobes: None,
+                eligibility: None,
             }
         );
         assert_eq!(lines[2].clone(), {
@@ -3018,6 +3129,7 @@ mod tests {
                 index: "nearest".to_string(),
                 fetch: None,
                 nprobes: Some(20),
+                eligibility: None,
             }
         );
         assert_eq!(