perf(planner): run a bounded correlated block once as a key-set anti-join

changelog.d/correlated-block-key-set.breaking.md

@@ -0,0 +1 @@
+- Saved query plans now use envelope version 9, which records each correlated block's strategy. Regenerate plans saved with earlier binaries; older binaries reject version 9 plans.

changelog.d/correlated-block-key-set.performance.md

@@ -0,0 +1,8 @@
+- A correlated block (`not { ... }`, `count { ... }`, and the block aggregates)
+  that reaches the outer row only through one binding's identity and binds one
+  of its own nodes by a key or unique equality now runs once from that node,
+  when its row estimate is lower, instead of expanding every outer row's
+  binding. Explain reports the choice as the `AntiJoin` row's `strategy`. See
+  the [query guide][key-set-blocks].
+
+[key-set-blocks]: ../docs/user/queries/index.md#correlated-blocks

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

@@ -3206,14 +3206,16 @@ fn check_restored_plan(bound: &BoundPlan, restored: &BoundPlan) -> Result<(), St
                 PhysicalNode::AntiJoin {
                     predicate,
                     aggregate,
+                    strategy,
                     ..
                 },
                 Some(PhysicalNode::AntiJoin {
                     predicate: back,
                     aggregate: back_spec,
+                    strategy: back_strategy,
                     ..
                 }),
-            ) => predicate == back && aggregate == back_spec,
+            ) => predicate == back && aggregate == back_spec && strategy == back_strategy,
             (
                 PhysicalNode::Projection { .. }
                 | PhysicalNode::Aggregate { .. }

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

@@ -11,6 +11,7 @@
 //! scan Doc as $d: access id_lookup
 //! hash join $d
 //! hash join $d ran id_lookup
+//! anti join $x: key_set
 //! scan Doc as $d: ranked bm25
 //! scan Doc as $d: ranked nearest fetch 10
 //! scan Doc as $d: ranked nearest fetch 10 nprobes 20
@@ -35,20 +36,21 @@
 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,
-    PASS_PROJECTION_PUSHDOWN, PASS_RESOLVE,
+    PASS_ACCESS_PATH, PASS_ADDRESS_SHORT_CIRCUIT, PASS_AGGREGATE_PUSHDOWN, PASS_ANTI_JOIN_STRATEGY,
+    PASS_EXPAND_MODE, PASS_FRAGMENT_SCOPE, PASS_JOIN_ALGORITHM, PASS_LATE_MATERIALIZATION,
+    PASS_PREDICATE_PUSHDOWN, PASS_PROJECTION_PUSHDOWN, PASS_RESOLVE,
 };
 use serde::{Deserialize, Serialize};
 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>`, `validation scope <exact_subset|invariants_only>`";
+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>]`, `anti join $var: <correlated|key_set>`, `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"];
 const EXPAND_MODES: [&str; 2] = ["csr", "indexed_scan"];
+const ANTI_JOIN_STRATEGIES: [&str; 2] = ["correlated", "key_set"];
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub(crate) struct AggregateClaim {
@@ -123,6 +125,12 @@ pub(crate) enum PlanLine {
         binding: String,
         ran: Option<String>,
     },
+    /// Every physical `AntiJoin` correlated on `$outer_var` reaches its
+    /// block's matches by `strategy` (`correlated` or `key_set`).
+    AntiJoin {
+        outer_var: String,
+        strategy: String,
+    },
     /// Every selected physical scan is ranked, and one of them by the named
     /// index, asking it for `fetch` candidates and under the probe cap
     /// `nprobes` (`0` no cap) when claimed; an `rrf()` has one scan per arm.
@@ -238,6 +246,7 @@ pub(crate) fn parse_plan_body(body: &[(usize, &str)]) -> Result<Vec<PlanLine>, S
                 PASS_JOIN_ALGORITHM,
                 PASS_EXPAND_MODE,
                 PASS_ACCESS_PATH,
+                PASS_ANTI_JOIN_STRATEGY,
                 omnigraph_planner::scan_access::PASS_SCAN_ACCESS,
                 omnigraph_planner::scan_access::PASS_KEY_TO_ID,
             ]
@@ -459,6 +468,26 @@ pub(crate) fn parse_plan_body(body: &[(usize, &str)]) -> Result<Vec<PlanLine>, S
             });
             continue;
         }
+        if let Some(rest) = line.strip_prefix("anti join ") {
+            let claim = rest.split_once(':').and_then(|(outer_var, strategy)| {
+                let outer_var = outer_var
+                    .trim()
+                    .strip_prefix('$')
+                    .filter(|v| identifier(v))?;
+                let strategy = strategy.trim();
+                ANTI_JOIN_STRATEGIES
+                    .contains(&strategy)
+                    .then(|| (outer_var.to_string(), strategy.to_string()))
+            });
+            let Some((outer_var, strategy)) = claim else {
+                return Err(refused("claims `anti join $var: <correlated|key_set>`"));
+            };
+            lines.push(PlanLine::AntiJoin {
+                outer_var,
+                strategy,
+            });
+            continue;
+        }
         let join = if let Some(rest) = line.strip_prefix("contains join ") {
             Some((rest, "contains join"))
         } else {
@@ -489,7 +518,7 @@ pub(crate) fn parse_plan_body(body: &[(usize, &str)]) -> Result<Vec<PlanLine>, S
             (rest, true)
         } else {
             return Err(refused(
-                "knows fourteen line heads: `scan`, `hash join`, `contains join`, `cross join`, `expand`, `filter`, `sort`, `rank fuse`, `aggregate`, `result columns`, `type`, `cast`, `no cast`, `pass`",
+                "knows fifteen line heads: `scan`, `hash join`, `anti join`, `contains join`, `cross join`, `expand`, `filter`, `sort`, `rank fuse`, `aggregate`, `result columns`, `type`, `cast`, `no cast`, `pass`",
             ));
         };
         let Some((selector, claim)) = rest.split_once(':') else {
@@ -955,6 +984,8 @@ struct PlannedNodes {
     modes: Vec<PlannedExpandMode>,
     accesses: Vec<PlannedAccess>,
     joins: Vec<PlannedJoin>,
+    /// `(outer_var, strategy)` of every physical `AntiJoin`.
+    anti_joins: Vec<(String, String)>,
     contains_joins: Vec<PlannedContainsJoin>,
     cross_joins: Vec<Vec<String>>,
     sorts: Vec<Result<Vec<String>, String>>,
@@ -1229,24 +1260,27 @@ fn planned_physical(node: &Value, out: &mut PlannedNodes) {
                 "expect plan: Aggregate has no aggregates list".to_string()
             )),
         },
-        Some("AntiJoin") => match node.get("aggregate") {
-            Some(Value::Null) => {}
-            Some(entry) => {
-                let mut entry = entry.clone();
-                if let Some(fields) = entry.as_object_mut()
-                    && let Some(gq) = fields.remove("gq")
-                {
-                    fields.insert("column".into(), gq);
+        Some("AntiJoin") => {
+            out.anti_joins.push((text("outer_var"), text("strategy")));
+            match node.get("aggregate") {
+                Some(Value::Null) => {}
+                Some(entry) => {
+                    let mut entry = entry.clone();
+                    if let Some(fields) = entry.as_object_mut()
+                        && let Some(gq) = fields.remove("gq")
+                    {
+                        fields.insert("column".into(), gq);
+                    }
+                    out.block_aggregates
+                        .push(serde_json::from_value(entry).map_err(|error| {
+                            format!("expect plan: malformed block aggregate entry: {error}")
+                        }));
                 }
-                out.block_aggregates
-                    .push(serde_json::from_value(entry).map_err(|error| {
-                        format!("expect plan: malformed block aggregate entry: {error}")
-                    }));
+                None => out.block_aggregates.push(Err(
+                    "expect plan: AntiJoin has no aggregate declaration".into(),
+                )),
             }
-            None => out.block_aggregates.push(Err(
-                "expect plan: AntiJoin has no aggregate declaration".into(),
-            )),
-        },
+        }
         Some("Expand") => {
             out.selections.push((
                 text("src"),
@@ -1918,6 +1952,32 @@ pub(crate) fn plan_mismatch(
                     }
                 }
             }
+            PlanLine::AntiJoin {
+                outer_var,
+                strategy,
+            } => {
+                let selected: Vec<&str> = nodes
+                    .anti_joins
+                    .iter()
+                    .filter(|(var, _)| var == outer_var)
+                    .map(|(_, planned)| planned.as_str())
+                    .collect();
+                if selected.is_empty() {
+                    let known: Vec<String> = nodes
+                        .anti_joins
+                        .iter()
+                        .map(|(var, _)| format!("${var}"))
+                        .collect();
+                    return Some(format!(
+                        "expect plan: no anti join correlated on `${outer_var}` in the physical plan; it correlates {known:?}"
+                    ));
+                }
+                if let Some(planned) = selected.iter().find(|planned| **planned != strategy) {
+                    return Some(format!(
+                        "expect plan: the anti join on `${outer_var}` is `{planned}`, expected `{strategy}`"
+                    ));
+                }
+            }
             PlanLine::Scan {
                 type_name,
                 binding,
@@ -2845,6 +2905,55 @@ mod tests {
         }
     }
 
+    #[test]
+    fn anti_join_claims_read_each_blocks_strategy() {
+        let explain = json!({
+            "physical_plan": {"node": "Projection", "inputs": [
+                {"node": "AntiJoin", "outer_var": "x", "strategy": "key_set", "aggregate": null, "inputs": [
+                    {"node": "AntiJoin", "outer_var": "p", "strategy": "correlated", "aggregate": null, "inputs": [
+                        {"node": "Scan", "table": "node:Person", "binding": "p"},
+                        {"node": "OuterReference", "outer_var": "p"}
+                    ]},
+                    {"node": "Scan", "table": "node:Source", "binding": "s0"}
+                ]}
+            ]},
+            "passes": ["resolve", "anti_join_strategy"],
+        });
+        let lines = parse_plan_body(&[
+            (0, "anti join $x: key_set"),
+            (1, "anti join $p: correlated"),
+            (2, "pass anti_join_strategy"),
+        ])
+        .unwrap();
+        assert_eq!(
+            lines[0],
+            PlanLine::AntiJoin {
+                outer_var: "x".to_string(),
+                strategy: "key_set".to_string(),
+            }
+        );
+        assert_eq!(check(&lines, &explain), None);
+        for (claim, message) in [
+            (
+                "anti join $x: correlated",
+                "the anti join on `$x` is `key_set`, expected `correlated`",
+            ),
+            ("anti join $q: key_set", "no anti join correlated on `$q`"),
+        ] {
+            let lines = parse_plan_body(&[(0, claim)]).unwrap();
+            let mismatch = check(&lines, &explain).unwrap();
+            assert!(mismatch.contains(message), "{mismatch}");
+        }
+        for refused in [
+            "anti join x: key_set",
+            "anti join $x key_set",
+            "anti join $x: hash_join",
+            "anti join $x:",
+        ] {
+            assert!(parse_plan_body(&[(0, refused)]).is_err(), "{refused}");
+        }
+    }
+
     #[test]
     fn scalar_access_claims_require_matching_typed_leaves_on_every_scan() {
         let leaf = json!({"kind":"search","index":"by_age","column":"age","search":"age = 1"});

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

@@ -1953,6 +1953,7 @@ mod result_contract {
                     op: CompOp::Gt,
                     right,
                 },
+                strategy: omnigraph_planner::AntiJoinStrategy::Correlated,
             });
             bound.plan.set_root(id);
             assert_eq!(check_bound_plan_round_trip(&bound), Ok(()));

crates/omnigraph-gqt/README.md

@@ -481,6 +481,7 @@ scan Doc: not columns [embedding]
 scan Doc as $d: filter reads [d.state]
 scan Doc as $e: no filter
 hash join $e
+anti join $p: key_set
 expand $d Knows $e: mode indexed_scan
 filter reads [d.rank, e.rank]
 sort tiebreak [$d, $e]
@@ -574,7 +575,12 @@ has another access path or is the build side of a hash join. A `hash
 join $var` line claims that a physical `HashJoin` reaches `$var`'s rows by
 reading its table once as the build side the traversal probes (pass
 `access_path` when the cost model decided); it fails when the plan holds no
-hash join over that binding. The `hash join` and `expand` lines take a
+hash join over that binding. An `anti join $var: <correlated|key_set>` line
+claims the strategy of every physical `AntiJoin` correlated on `$var` (a `not`,
+`exists`, `count` or aggregate block): `correlated` starts its inner tree from
+the outer rows, `key_set` runs it once from the block's own bounded binding
+(pass `anti_join_strategy`); it fails when no such block is in the physical
+plan or one took the other strategy. The `hash join` and `expand` lines take a
 trailing `ran <side>` (`hash join $e ran id_lookup`, `mode indexed_scan ran
 csr`): the side of the node's declared switch the run took, read from the
 execution report of the same run (the last attempt of the node's row, joined