From 41b9fa780dcb11ec61ebe11fc908936083377188 Mon Sep 17 00:00:00 2001 From: Ragnor Comerford Date: Mon, 12 Oct 2026 21:31:17 -0700 Subject: [PATCH 1/2] perf(planner): run a bounded correlated block once as a key-set anti-join A correlated block whose inner pattern reaches the outer row only through one binding's identity, and binds one of its own nodes by a key or unique equality or a list membership, now has a second inner tree: the same pattern rooted at that node, each Expand toward it reversed, ending on the correlation binding's id. Pass anti_join_strategy prices both trees by their summed row estimates and lowers the cheaper one; the physical AntiJoin records strategy correlated or key_set, and AntiJoinMaskExec folds a key-set inner tree into one aggregate per outer key as it streams. Saved bound plans move to version 8. GQT gains the plan line `anti join $var: `. --- .../correlated-block-key-set.breaking.md | 1 + .../correlated-block-key-set.performance.md | 8 + crates/omnigraph-gqt-core/src/lib.rs | 4 +- crates/omnigraph-gqt-core/src/plan.rs | 151 +++++++-- crates/omnigraph-gqt-core/src/tests.rs | 1 + crates/omnigraph-gqt/README.md | 8 +- ...ti_join_key_set_from_a_bounded_binding.gqt | 289 ++++++++++++++++ ..._join_key_set_reverses_hops_and_counts.gqt | 169 ++++++++++ crates/omnigraph-planner/src/bound.rs | 2 +- crates/omnigraph-planner/src/cost.rs | 65 +++- crates/omnigraph-planner/src/key_set.rs | 316 ++++++++++++++++++ crates/omnigraph-planner/src/lib.rs | 10 +- crates/omnigraph-planner/src/mirror.rs | 7 +- crates/omnigraph-planner/src/optimizer.rs | 94 +++++- crates/omnigraph-planner/src/physical.rs | 11 +- .../tests/integration/bound_plan.rs | 14 +- .../tests/integration/lower_walk.rs | 1 + .../tests/query_plan/cost.rs | 217 ++++++++++++ crates/omnigraph/src/engine/lower.rs | 17 +- .../src/engine/operators/anti_join.rs | 128 ++++++- crates/omnigraph/src/engine/report/tests.rs | 56 ++++ .../tests/integration/anti_join_cost.rs | 269 +++++++++++++++ .../omnigraph/tests/integration/engine_v2.rs | 175 ++++++++++ crates/omnigraph/tests/integration/main.rs | 1 + docs/dev/execution.md | 36 +- docs/dev/testing.md | 2 +- docs/user/queries/explain.md | 19 +- docs/user/queries/index.md | 10 +- 28 files changed, 2013 insertions(+), 68 deletions(-) create mode 100644 changelog.d/correlated-block-key-set.breaking.md create mode 100644 changelog.d/correlated-block-key-set.performance.md create mode 100644 crates/omnigraph-gqt/cases/v2/planner/anti_join_key_set_from_a_bounded_binding.gqt create mode 100644 crates/omnigraph-gqt/cases/v2/planner/anti_join_key_set_reverses_hops_and_counts.gqt create mode 100644 crates/omnigraph-planner/src/key_set.rs create mode 100644 crates/omnigraph/tests/integration/anti_join_cost.rs diff --git a/changelog.d/correlated-block-key-set.breaking.md b/changelog.d/correlated-block-key-set.breaking.md new file mode 100644 index 000000000..edabbe445 --- /dev/null +++ b/changelog.d/correlated-block-key-set.breaking.md @@ -0,0 +1 @@ +- Saved query plans now use envelope version 8, which records each correlated block's strategy. Regenerate plans saved with earlier binaries; older binaries reject version 8 plans. diff --git a/changelog.d/correlated-block-key-set.performance.md b/changelog.d/correlated-block-key-set.performance.md new file mode 100644 index 000000000..187b99def --- /dev/null +++ b/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 diff --git a/crates/omnigraph-gqt-core/src/lib.rs b/crates/omnigraph-gqt-core/src/lib.rs index 478e3a0ee..ebe740835 100644 --- a/crates/omnigraph-gqt-core/src/lib.rs +++ b/crates/omnigraph-gqt-core/src/lib.rs @@ -2854,14 +2854,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 { .. } diff --git a/crates/omnigraph-gqt-core/src/plan.rs b/crates/omnigraph-gqt-core/src/plan.rs index 6cc2d74d0..a7e2f1f2c 100644 --- a/crates/omnigraph-gqt-core/src/plan.rs +++ b/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 @@ -32,20 +33,21 @@ use omnigraph_compiler::catalog::Catalog; 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 [ as $var]: columns [a, b]`, `scan [ as $var]: not columns [a, b]`, `scan [ as $var]: filter reads [v.a]`, `scan [ as $var]: no filter`, `scan [ as $var]: access id_lookup`, `scan [ as $var]: access sequential`, `scan [ as $var]: access index_probe `, `hash join $var[ ran ]`, `scan [ as $var]: ranked [ fetch ][ nprobes ]`, `scan [ as $var]: runtime filter `, `scan [ as $var]: no runtime filter`, `contains join $h.x contains $n.y`, `cross join $h.x contains $n.y`, `expand $src $dst: mode [ ran ]`, `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 : ([distinct ]) -> `, `block aggregate : () -> `, `result columns [: , ...]`, `type : `, `cast : -> `, `no cast `, `hydrate $var: columns [a, b]`, `pass `, `not pass `"; +const FORMS: &str = "forms: `scan [ as $var]: columns [a, b]`, `scan [ as $var]: not columns [a, b]`, `scan [ as $var]: filter reads [v.a]`, `scan [ as $var]: no filter`, `scan [ as $var]: access id_lookup`, `scan [ as $var]: access sequential`, `scan [ as $var]: access index_probe `, `hash join $var[ ran ]`, `anti join $var: `, `scan [ as $var]: ranked [ fetch ][ nprobes ]`, `scan [ as $var]: runtime filter `, `scan [ as $var]: no runtime filter`, `contains join $h.x contains $n.y`, `cross join $h.x contains $n.y`, `expand $src $dst: mode [ ran ]`, `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 : ([distinct ]) -> `, `block aggregate : () -> `, `result columns [: , ...]`, `type : `, `cast : -> `, `no cast `, `hydrate $var: columns [a, b]`, `pass `, `not pass `"; 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 { @@ -120,6 +122,12 @@ pub(crate) enum PlanLine { binding: String, ran: Option, }, + /// 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. @@ -228,6 +236,7 @@ pub(crate) fn parse_plan_body(body: &[(usize, &str)]) -> Result, 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, ] @@ -434,6 +443,26 @@ pub(crate) fn parse_plan_body(body: &[(usize, &str)]) -> Result, 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: `")); + }; + lines.push(PlanLine::AntiJoin { + outer_var, + strategy, + }); + continue; + } let join = if let Some(rest) = line.strip_prefix("contains join ") { Some((rest, "contains join")) } else { @@ -464,7 +493,7 @@ pub(crate) fn parse_plan_body(body: &[(usize, &str)]) -> Result, 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 { @@ -918,6 +947,8 @@ struct PlannedNodes { modes: Vec, accesses: Vec, joins: Vec, + /// `(outer_var, strategy)` of every physical `AntiJoin`. + anti_joins: Vec<(String, String)>, contains_joins: Vec, cross_joins: Vec>, sorts: Vec, String>>, @@ -1192,24 +1223,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"), @@ -1853,6 +1887,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 = 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, @@ -2735,6 +2795,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"}); diff --git a/crates/omnigraph-gqt-core/src/tests.rs b/crates/omnigraph-gqt-core/src/tests.rs index a31755863..ac661ca7d 100644 --- a/crates/omnigraph-gqt-core/src/tests.rs +++ b/crates/omnigraph-gqt-core/src/tests.rs @@ -1904,6 +1904,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(())); diff --git a/crates/omnigraph-gqt/README.md b/crates/omnigraph-gqt/README.md index c9f2e8d44..1e7670b27 100644 --- a/crates/omnigraph-gqt/README.md +++ b/crates/omnigraph-gqt/README.md @@ -439,6 +439,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] @@ -531,7 +532,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: ` 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 ` (`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 diff --git a/crates/omnigraph-gqt/cases/v2/planner/anti_join_key_set_from_a_bounded_binding.gqt b/crates/omnigraph-gqt/cases/v2/planner/anti_join_key_set_from_a_bounded_binding.gqt new file mode 100644 index 000000000..792d0dfa2 --- /dev/null +++ b/crates/omnigraph-gqt/cases/v2/planner/anti_join_key_set_from_a_bounded_binding.gqt @@ -0,0 +1,289 @@ +# issue: none +# red_on: e5c14da7: `anti join $x: key_set` is no plan line; the inner tree started from every outer row. +# notes: A correlated block that reaches the outer rows only through `$x`'s identity and binds +# notes: one of its own nodes by a key equality runs once from that binding (`key_set`): the +# notes: excluded source's touchpoints, then their identities. Count, sum and bound-edge blocks +# notes: keep each outer row by its key's matches. A block whose bindings carry no bound filter, +# notes: and a block that reads another outer binding, stay `correlated`. Rows agree either way. + +--- runner +timeout_ms: 10000 +environments: + - target: omnigraph-engine + storage: local-filesystem + +--- schema +node Channel { slug: String @key } +node Source { slug: String @key } +node Touchpoint { + slug: String @key + action: String + weight: I64? +} +node ExternalID { slug: String @key } +edge ChannelWatchesSource: Channel -> Source +edge TouchpointOnSource: Touchpoint -> Source { via: String? } +edge TouchpointByIdentity: Touchpoint -> ExternalID + +--- seed +{"type":"Channel","data":{"slug":"c1"}} +{"type":"Source","data":{"slug":"s1"}} +{"type":"Source","data":{"slug":"s2"}} +{"type":"Source","data":{"slug":"s3"}} +{"type":"Source","data":{"slug":"excluded"}} +{"type":"ExternalID","data":{"slug":"x1"}} +{"type":"ExternalID","data":{"slug":"x2"}} +{"type":"ExternalID","data":{"slug":"x3"}} +{"type":"ExternalID","data":{"slug":"x4"}} +{"type":"ExternalID","data":{"slug":"x5"}} +{"type":"Touchpoint","data":{"slug":"t1","action":"star","weight":1}} +{"type":"Touchpoint","data":{"slug":"t2","action":"star","weight":3}} +{"type":"Touchpoint","data":{"slug":"t3","action":"star","weight":1}} +{"type":"Touchpoint","data":{"slug":"t4","action":"star","weight":1}} +{"type":"Touchpoint","data":{"slug":"t5","action":"star","weight":1}} +{"type":"Touchpoint","data":{"slug":"t6","action":"star","weight":2}} +{"type":"Touchpoint","data":{"slug":"t7","action":"fork","weight":4}} +{"type":"Touchpoint","data":{"slug":"t8","action":"star","weight":1}} +{"type":"Touchpoint","data":{"slug":"t9","action":"fork","weight":1}} +{"type":"Touchpoint","data":{"slug":"t10","action":"star","weight":10}} +{"type":"Touchpoint","data":{"slug":"t11","action":"fork","weight":9}} +{"edge":"ChannelWatchesSource","from":"c1","to":"s1"} +{"edge":"ChannelWatchesSource","from":"c1","to":"s2"} +{"edge":"TouchpointOnSource","from":"t1","to":"s1"} +{"edge":"TouchpointOnSource","from":"t2","to":"excluded","data":{"via":"web"}} +{"edge":"TouchpointOnSource","from":"t3","to":"s1"} +{"edge":"TouchpointOnSource","from":"t4","to":"s2"} +{"edge":"TouchpointOnSource","from":"t5","to":"s2"} +{"edge":"TouchpointOnSource","from":"t6","to":"excluded","data":{"via":"api"}} +{"edge":"TouchpointOnSource","from":"t7","to":"excluded","data":{"via":"web"}} +{"edge":"TouchpointOnSource","from":"t8","to":"s1"} +{"edge":"TouchpointOnSource","from":"t9","to":"s3"} +{"edge":"TouchpointOnSource","from":"t10","to":"s1"} +{"edge":"TouchpointOnSource","from":"t11","to":"excluded","data":{"via":"api"}} +{"edge":"TouchpointByIdentity","from":"t1","to":"x1"} +{"edge":"TouchpointByIdentity","from":"t2","to":"x1"} +{"edge":"TouchpointByIdentity","from":"t3","to":"x2"} +{"edge":"TouchpointByIdentity","from":"t4","to":"x2"} +{"edge":"TouchpointByIdentity","from":"t5","to":"x3"} +{"edge":"TouchpointByIdentity","from":"t6","to":"x3"} +{"edge":"TouchpointByIdentity","from":"t7","to":"x3"} +{"edge":"TouchpointByIdentity","from":"t8","to":"x4"} +{"edge":"TouchpointByIdentity","from":"t9","to":"x4"} +{"edge":"TouchpointByIdentity","from":"t10","to":"x5"} +{"edge":"TouchpointByIdentity","from":"t11","to":"x5"} + +--- query +query lookalikes($channel: String, $exclude: String) { + match { + $c: Channel { slug: $channel } + $s: Source + $c channelWatchesSource $s + $t: Touchpoint { action: "star" } + $t touchpointOnSource $s + $x: ExternalID + $t touchpointByIdentity $x + not { + $s0: Source { slug: $exclude } + $t0: Touchpoint + $t0 touchpointOnSource $s0 + $t0 touchpointByIdentity $x + } + } + return { $x.slug, count($t) as repos } +} + +--- params +{"channel": "c1", "exclude": "excluded"} + +--- expect unordered +{"x.slug": "x2", "repos": 2} +{"x.slug": "x4", "repos": 1} + +--- expect shape +x.slug: String +repos: I64 + +--- expect plan +anti join $x: key_set +pass anti_join_strategy + +--- expect same as v1 + +--- query +query twice_on_excluded($channel: String, $exclude: String) { + match { + $c: Channel { slug: $channel } + $s: Source + $c channelWatchesSource $s + $t: Touchpoint { action: "star" } + $t touchpointOnSource $s + $x: ExternalID + $t touchpointByIdentity $x + count { + $s0: Source { slug: $exclude } + $t0: Touchpoint + $t0 touchpointOnSource $s0 + $t0 touchpointByIdentity $x + } >= 2 + } + return { $x.slug, count($t) as repos } +} + +--- params +{"channel": "c1", "exclude": "excluded"} + +--- expect unordered +{"x.slug": "x3", "repos": 1} + +--- expect shape +x.slug: String +repos: I64 + +--- expect plan +anti join $x: key_set + +--- query +query heavy_on_excluded($channel: String, $exclude: String) { + match { + $c: Channel { slug: $channel } + $s: Source + $c channelWatchesSource $s + $t: Touchpoint { action: "star" } + $t touchpointOnSource $s + $x: ExternalID + $t touchpointByIdentity $x + sum($t0.weight) { + $s0: Source { slug: $exclude } + $t0: Touchpoint + $t0 touchpointOnSource $s0 + $t0 touchpointByIdentity $x + } > 3 + } + return { $x.slug, count($t) as repos } +} + +--- params +{"channel": "c1", "exclude": "excluded"} + +--- expect unordered +{"x.slug": "x3", "repos": 1} +{"x.slug": "x5", "repos": 1} + +--- expect shape +x.slug: String +repos: I64 + +--- expect plan +anti join $x: key_set + +--- query +query not_through_the_web($channel: String, $exclude: String) { + match { + $c: Channel { slug: $channel } + $s: Source + $c channelWatchesSource $s + $t: Touchpoint { action: "star" } + $t touchpointOnSource $s + $x: ExternalID + $t touchpointByIdentity $x + not { + $s0: Source { slug: $exclude } + $t0: Touchpoint + $t0 $e:touchpointOnSource $s0 + $e.via = "web" + $t0 touchpointByIdentity $x + } + } + return { $x.slug, count($t) as repos } +} + +--- params +{"channel": "c1", "exclude": "excluded"} + +--- expect unordered +{"x.slug": "x2", "repos": 2} +{"x.slug": "x4", "repos": 1} +{"x.slug": "x5", "repos": 1} + +--- expect shape +x.slug: String +repos: I64 + +--- expect plan +anti join $x: key_set + +--- expect same as v1 + +--- query +query never_forked($channel: String) { + match { + $c: Channel { slug: $channel } + $s: Source + $c channelWatchesSource $s + $t: Touchpoint { action: "star" } + $t touchpointOnSource $s + $x: ExternalID + $t touchpointByIdentity $x + not { + $t0: Touchpoint { action: "fork" } + $t0 touchpointByIdentity $x + } + } + return { $x.slug, count($t) as repos } +} + +--- params +{"channel": "c1"} + +--- expect unordered +{"x.slug": "x1", "repos": 1} +{"x.slug": "x2", "repos": 2} + +--- expect shape +x.slug: String +repos: I64 + +--- expect plan +anti join $x: correlated +not pass anti_join_strategy + +--- expect same as v1 + +--- query +query no_heavier_on_excluded($channel: String, $exclude: String) { + match { + $c: Channel { slug: $channel } + $s: Source + $c channelWatchesSource $s + $t: Touchpoint { action: "star" } + $t touchpointOnSource $s + $x: ExternalID + $t touchpointByIdentity $x + not { + $s0: Source { slug: $exclude } + $t0: Touchpoint + $t0 touchpointOnSource $s0 + $t0 touchpointByIdentity $x + $t0.weight > $t.weight + } + } + return { $x.slug, count($t) as repos } +} + +--- params +{"channel": "c1", "exclude": "excluded"} + +--- expect unordered +{"x.slug": "x2", "repos": 2} +{"x.slug": "x4", "repos": 1} +{"x.slug": "x5", "repos": 1} + +--- expect shape +x.slug: String +repos: I64 + +--- expect plan +anti join $x: correlated +not pass anti_join_strategy + +--- expect same as v1 diff --git a/crates/omnigraph-gqt/cases/v2/planner/anti_join_key_set_reverses_hops_and_counts.gqt b/crates/omnigraph-gqt/cases/v2/planner/anti_join_key_set_reverses_hops_and_counts.gqt new file mode 100644 index 000000000..fbae6525f --- /dev/null +++ b/crates/omnigraph-gqt/cases/v2/planner/anti_join_key_set_reverses_hops_and_counts.gqt @@ -0,0 +1,169 @@ +# issue: none +# notes: A key-set block runs its pattern from the bound binding, so each traversal toward the +# notes: outer binding is reversed. Exact and ranged hop counts and a block written from the +# notes: bound binding keep their rows. An unbound traversal counts +# notes: distinct neighbours and a bound edge counts edge rows in either direction: d's two +# notes: parallel edges to e count once unbound and twice bound. + +--- runner +timeout_ms: 10000 +environments: + - target: omnigraph-engine + storage: local-filesystem + +--- schema +node Person { + name: String @key +} + +edge Knows: Person -> Person + +--- seed +{"type":"Person","data":{"name":"a"}} +{"type":"Person","data":{"name":"b"}} +{"type":"Person","data":{"name":"c"}} +{"type":"Person","data":{"name":"d"}} +{"type":"Person","data":{"name":"e"}} +{"edge":"Knows","from":"a","to":"b"} +{"edge":"Knows","from":"c","to":"d"} +{"edge":"Knows","from":"d","to":"e"} +{"edge":"Knows","from":"d","to":"e"} + +--- query +query no_two_hops_to_e() { + match { + $p: Person + not { + $p knows{2,2} $x + $x: Person { name: "e" } + } + } + return { $p.name } +} + +--- expect unordered +{"p.name": "a"} +{"p.name": "b"} +{"p.name": "d"} +{"p.name": "e"} + +--- expect shape +p.name: String + +--- expect plan +anti join $p: key_set + +--- expect same as v1 + +--- query +query not_within_two_of_e() { + match { + $p: Person + not { + $p knows{1,2} $x + $x: Person { name: "e" } + } + } + return { $p.name } +} + +--- expect unordered +{"p.name": "a"} +{"p.name": "b"} +{"p.name": "e"} + +--- expect shape +p.name: String + +--- expect plan +anti join $p: key_set + +--- expect same as v1 + +--- query +query not_known_by_c() { + match { + $p: Person + not { + $x knows $p + $x: Person { name: "c" } + } + } + return { $p.name } +} + +--- expect unordered +{"p.name": "a"} +{"p.name": "b"} +{"p.name": "c"} +{"p.name": "e"} + +--- expect shape +p.name: String + +--- expect plan +anti join $p: key_set + +--- expect same as v1 + +--- query +query one_neighbour_e() { + match { + $p: Person + count { + $p knows $x + $x: Person { name: "e" } + } = 1 + } + return { $p.name } +} + +--- expect unordered +{"p.name": "d"} + +--- expect shape +p.name: String + +--- expect plan +anti join $p: key_set + +--- query +query one_edge_row_to_e() { + match { + $p: Person + count { + $p $k:knows $x + $x: Person { name: "e" } + } = 1 + } + return { $p.name } +} + +--- expect unordered + +--- expect shape +p.name: String + +--- expect plan +anti join $p: key_set + +--- query +query two_edge_rows_to_e() { + match { + $p: Person + count { + $p $k:knows $x + $x: Person { name: "e" } + } = 2 + } + return { $p.name } +} + +--- expect unordered +{"p.name": "d"} + +--- expect shape +p.name: String + +--- expect plan +anti join $p: key_set diff --git a/crates/omnigraph-planner/src/bound.rs b/crates/omnigraph-planner/src/bound.rs index 00d01c983..4ada7897d 100644 --- a/crates/omnigraph-planner/src/bound.rs +++ b/crates/omnigraph-planner/src/bound.rs @@ -15,7 +15,7 @@ use crate::mirror::BoundPlanMirror; use crate::physical::{NodeId, PhysicalPlan}; /// Saved execution plans use a separate version from rendered explain output. -pub const BOUND_PLAN_VERSION: u32 = 7; +pub const BOUND_PLAN_VERSION: u32 = 8; /// The envelope omits top-level `plan` and `values` so unversioned readers reject it. #[derive(Serialize, Deserialize)] diff --git a/crates/omnigraph-planner/src/cost.rs b/crates/omnigraph-planner/src/cost.rs index 33d53d1e6..4c57d73e2 100644 --- a/crates/omnigraph-planner/src/cost.rs +++ b/crates/omnigraph-planner/src/cost.rs @@ -129,6 +129,31 @@ pub fn choose_access_path( }) } +/// How a correlated block (`not { … }`, `count { … } > 2`) reaches the +/// inner matches of its outer rows. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AntiJoinStrategy { + /// The inner tree runs once over the outer rows, starting from them: work + /// scales with the outer rows times the correlation binding's fanout. + #[default] + Correlated, + /// The inner tree runs once from its most bounded binding, with no outer + /// rows, and ends on the correlation binding; the outer rows are kept by + /// the aggregate of their key's matches. Work scales with the block alone. + KeySet, +} + +/// The strategy of a correlated block from the summed row estimates of its two +/// inner trees (`chain_rows`): the key set only when it produces fewer rows. +pub fn choose_anti_join_strategy(correlated_rows: u64, key_set_rows: u64) -> AntiJoinStrategy { + if key_set_rows < correlated_rows { + AntiJoinStrategy::KeySet + } else { + AntiJoinStrategy::Correlated + } +} + /// Whether the per-hop `key_col IN (...)` scan is served by the BTREE /// (`Indexed`) or falls back to a full scan (`Degraded`). New plans derive this /// from pinned index facts; legacy plans retain the runtime coverage probe. @@ -379,6 +404,43 @@ fn bounded_property<'a>( /// destination type's row count, so a hub can exceed it. `None` where the /// source holds no statistic for a node the estimate depends on. pub fn estimate_rows(plan: &LogicalPlan, node: LogicalId, source: &dyn PlanSource) -> Option { + estimate_rows_over(plan, node, source, None) +} + +/// The rows a correlated block's single-input chain produces, summed from +/// `top` down to its leaf: the `OuterReference` leaf counts `outer_rows`, a +/// key-set tree's root scan its own estimate. The anti-join cost measure: +/// every row a chain node emits is a row the next node reads. `None` when a +/// node has another shape or any estimate is unknown. +pub fn chain_rows( + plan: &LogicalPlan, + top: LogicalId, + source: &dyn PlanSource, + outer_rows: Option, +) -> Option { + let mut total = 0u64; + let mut node = Some(top); + while let Some(id) = node { + total = total.saturating_add(estimate_rows_over(plan, id, source, outer_rows)?); + node = match plan.node(id)? { + LogicalNode::TableScan { input, .. } => *input, + LogicalNode::Expand { input, .. } | LogicalNode::Filter { input, .. } => Some(*input), + LogicalNode::OuterReference { .. } => None, + _ => return None, + }; + } + Some(total) +} + +fn estimate_rows_over( + plan: &LogicalPlan, + node: LogicalId, + source: &dyn PlanSource, + outer_rows: Option, +) -> Option { + let estimate_rows = |plan: &LogicalPlan, node: LogicalId, source: &dyn PlanSource| { + estimate_rows_over(plan, node, source, outer_rows) + }; match plan.node(node)? { LogicalNode::TableScan { input, spec } => { let rows = scan_row_estimate(spec, source)?; @@ -433,7 +495,8 @@ pub fn estimate_rows(plan: &LogicalPlan, node: LogicalId, source: &dyn PlanSourc estimate_rows(plan, *left, source)? .saturating_mul(estimate_rows(plan, *right, source)?), ), - LogicalNode::OuterReference { .. } | LogicalNode::MergeClassify { .. } => None, + LogicalNode::OuterReference { .. } => outer_rows, + LogicalNode::MergeClassify { .. } => None, } } diff --git a/crates/omnigraph-planner/src/key_set.rs b/crates/omnigraph-planner/src/key_set.rs new file mode 100644 index 000000000..e7ff0d01b --- /dev/null +++ b/crates/omnigraph-planner/src/key_set.rs @@ -0,0 +1,316 @@ +//! The key-set form of a correlated block (`not { … }`, `count { … } > 2`). +//! +//! The logical inner tree of an `AntiJoin` starts from the outer rows: its +//! leaf is an `OuterReference` and its first `Expand` leaves the correlation +//! binding. When the block reads nothing of the outer rows but that binding's +//! identity, the same matches come from the block's own pattern, rooted at +//! one of its own bindings and ended on the correlation binding: an `Expand` +//! emits one row per distinct endpoint pair (or per edge row, with an edge +//! binding), and the reversed traversal emits the same pairs, so every key +//! meets the same matches in either orientation. The key-set tree is built +//! here, in its own arena, and the physical selection lowers it only when the +//! cost model prices it below the correlated tree; the logical plan keeps the +//! correlated tree as the block's meaning. + +use std::collections::{HashMap, HashSet, VecDeque}; + +use omnigraph_compiler::ir::{IRExpr, SubqueryPredicate}; +use omnigraph_compiler::traversal::EdgeSelection; + +use crate::cost::scan_row_estimate; +use crate::error::PlanError; +use crate::logical::{ColumnRef, LogicalId, LogicalNode, LogicalPlan, Predicate, ScanSpec}; +use crate::optimizer::reads_of_expr; +use crate::source::PlanSource; + +/// The re-rooted inner tree of one `AntiJoin`, alone in its arena. +#[derive(Debug)] +pub(crate) struct KeySetTree { + pub plan: LogicalPlan, + /// The tree's top: its last `Expand`, which emits the correlation + /// binding's identity, or the `Filter` of the block's residual conjuncts. + pub root: LogicalId, +} + +/// One `Expand` of the block's pattern as the correlated tree wrote it. +struct PatternEdge { + src: String, + dst: String, + edges: EdgeSelection, + src_type: String, + dst_type: String, + min_hops: u32, + max_hops: Option, + edge_binding: Option, +} + +impl PatternEdge { + /// This edge's `Expand` leaving `from`, reversed when it was written + /// toward `from`. + fn leaving(&self, from: &str, input: LogicalId) -> LogicalNode { + let reversed = self.dst == from; + LogicalNode::Expand { + input, + src: if reversed { &self.dst } else { &self.src }.clone(), + dst: if reversed { &self.src } else { &self.dst }.clone(), + edges: if reversed { + self.edges.reversed() + } else { + self.edges.clone() + }, + src_type: if reversed { + &self.dst_type + } else { + &self.src_type + } + .clone(), + dst_type: if reversed { + &self.src_type + } else { + &self.dst_type + } + .clone(), + min_hops: self.min_hops, + max_hops: self.max_hops, + edge_binding: self.edge_binding.clone(), + } + } + + fn other(&self, end: &str) -> &str { + if self.src == end { + &self.dst + } else { + &self.src + } + } +} + +/// The key-set tree of the `AntiJoin` whose correlated tree is `inner`, when +/// the block qualifies: the inner tree is a chain of `Expand`, dependent +/// `TableScan` and `Filter` nodes over the `OuterReference`, every `Expand` +/// introduces a new binding from a bound one (a pattern tree rooted at +/// `outer_var`), every introduced node binding has its scan, nothing reads +/// an outer binding (`outer_var` included: its identity is the key, never a +/// column read), and one binding's scan is bounded by its own filter (a key +/// or unique equality or a list membership) below its table's row count. The +/// root is the binding with the fewest estimated rows, the first in the +/// correlated tree on a tie. `None` when the block does not qualify. +pub(crate) fn key_set_tree( + plan: &LogicalPlan, + inner: LogicalId, + outer_var: &str, + predicate: &SubqueryPredicate, + source: &dyn PlanSource, +) -> Result, PlanError> { + if plan.traversal_work_limit().is_some() { + return Ok(None); + } + let mut chain = Vec::new(); + let mut current = inner; + loop { + match plan.node(current) { + Some(LogicalNode::OuterReference { outer_var: leaf }) if leaf == outer_var => break, + Some( + LogicalNode::TableScan { + input: Some(input), .. + } + | LogicalNode::Expand { input, .. } + | LogicalNode::Filter { input, .. }, + ) => { + chain.push(current); + current = *input; + } + _ => return Ok(None), + } + } + chain.reverse(); + + let mut edges: Vec = Vec::new(); + let mut vertices: Vec = vec![outer_var.to_string()]; + let mut introduced: HashSet = HashSet::new(); + let mut scans: HashMap = HashMap::new(); + let mut residual: Vec = Vec::new(); + let mut reads: Vec = Vec::new(); + let mut last_expand_dst: Option = None; + for &id in &chain { + match plan.node(id) { + Some(LogicalNode::Expand { + src, + dst, + edges: selection, + src_type, + dst_type, + min_hops, + max_hops, + edge_binding, + .. + }) => { + let fresh = |name: &str| name != outer_var && !introduced.contains(name); + if !vertices.contains(src) + || !fresh(dst) + || edge_binding + .as_deref() + .is_some_and(|binding| !fresh(binding)) + { + return Ok(None); + } + vertices.push(dst.clone()); + introduced.insert(dst.clone()); + introduced.extend(edge_binding.iter().cloned()); + edges.push(PatternEdge { + src: src.clone(), + dst: dst.clone(), + edges: selection.clone(), + src_type: src_type.clone(), + dst_type: dst_type.clone(), + min_hops: *min_hops, + max_hops: *max_hops, + edge_binding: edge_binding.clone(), + }); + last_expand_dst = Some(dst.clone()); + } + Some(LogicalNode::TableScan { spec, .. }) => { + let binding = spec.binding.as_deref(); + if binding.is_none() + || binding != last_expand_dst.as_deref() + || spec.fragments.is_some() + || spec.runtime_filter.is_some() + { + return Ok(None); + } + if let Some(filter) = &spec.filter { + predicate_reads(filter, &mut reads); + } + scans.insert(spec.binding.clone().unwrap_or_default(), id); + last_expand_dst = None; + } + Some(LogicalNode::Filter { conjuncts, .. }) => { + for conjunct in conjuncts { + reads_of_expr(conjunct, &mut reads); + } + residual.extend(conjuncts.iter().cloned()); + } + _ => return Ok(None), + } + } + if let Some(argument) = predicate.left.arg() { + reads_of_expr(argument, &mut reads); + } + if edges.is_empty() + || reads.iter().any(|read| !introduced.contains(&read.binding)) + || vertices[1..] + .iter() + .any(|vertex| !scans.contains_key(vertex)) + { + return Ok(None); + } + + let mut root: Option<(u64, &str)> = None; + for vertex in &vertices[1..] { + let Some(LogicalNode::TableScan { spec, .. }) = plan.node(scans[vertex]) else { + return Ok(None); + }; + let Some(bound) = bounded_rows(spec, source) else { + continue; + }; + if root.is_none_or(|(rows, _)| bound < rows) { + root = Some((bound, vertex.as_str())); + } + } + let Some((_, root)) = root else { + return Ok(None); + }; + + let schema_of = |id: LogicalId| { + plan.schema(id) + .cloned() + .ok_or_else(|| PlanError::Internal(format!("logical node {id} has no schema"))) + }; + let mut tree = LogicalPlan::new(); + let scan_of = |vertex: &str| match plan.node(scans[vertex]) { + Some(LogicalNode::TableScan { spec, .. }) => Ok(spec.clone()), + _ => Err(PlanError::Internal(format!( + "the key-set scan of `${vertex}` is not a table scan" + ))), + }; + let mut top = tree.add( + LogicalNode::TableScan { + input: None, + spec: scan_of(root)?, + }, + schema_of(scans[root])?, + ); + let mut used = vec![false; edges.len()]; + let mut reached = HashSet::from([root.to_string()]); + let mut pending = VecDeque::from([root.to_string()]); + while let Some(from) = pending.pop_front() { + for (index, edge) in edges.iter().enumerate() { + if used[index] || (edge.src != from && edge.dst != from) { + continue; + } + used[index] = true; + let to = edge.other(&from).to_string(); + let schema = tree + .schema(top) + .cloned() + .ok_or_else(|| PlanError::Internal("key-set node has no schema".to_string()))?; + top = tree.add(edge.leaving(&from, top), schema.clone()); + if to != outer_var { + top = tree.add( + LogicalNode::TableScan { + input: Some(top), + spec: scan_of(&to)?, + }, + schema, + ); + } + if reached.insert(to.clone()) { + pending.push_back(to); + } + } + } + if used.iter().any(|used| !used) || !reached.contains(outer_var) { + return Err(PlanError::Internal( + "a correlated block's pattern is not a tree over its bindings".to_string(), + )); + } + if !residual.is_empty() { + let schema = tree + .schema(top) + .cloned() + .ok_or_else(|| PlanError::Internal("key-set node has no schema".to_string()))?; + top = tree.add( + LogicalNode::Filter { + input: top, + conjuncts: residual, + }, + schema, + ); + } + tree.set_root(top); + Ok(Some(KeySetTree { + plan: tree, + root: top, + })) +} + +/// The rows `spec` reads as a table scan when its own filter bounds them +/// below the table's row count; `None` when it reads the whole table. +fn bounded_rows(spec: &ScanSpec, source: &dyn PlanSource) -> Option { + let table_rows = source + .node_type(spec.table.node_type_name()?) + .ok()? + .row_count?; + scan_row_estimate(spec, source).filter(|rows| *rows < table_rows) +} + +fn predicate_reads(predicate: &Predicate, out: &mut Vec) { + match predicate { + Predicate::Gq { reads, .. } => out.extend(reads.iter().cloned()), + Predicate::And { left, right } => { + predicate_reads(left, out); + predicate_reads(right, out); + } + Predicate::IdAfter { .. } | Predicate::VersionWindow { .. } => {} + } +} diff --git a/crates/omnigraph-planner/src/lib.rs b/crates/omnigraph-planner/src/lib.rs index d3dd14a32..027f08a54 100644 --- a/crates/omnigraph-planner/src/lib.rs +++ b/crates/omnigraph-planner/src/lib.rs @@ -29,6 +29,7 @@ pub mod cost; pub mod error; pub mod explain; pub mod gate; +mod key_set; pub mod logical; pub mod lower; pub mod mirror; @@ -48,10 +49,11 @@ pub use aggregate::{ }; pub use bound::{BOUND_PLAN_VERSION, BoundPlan, ValueTable}; pub use cost::{ - AccessPath, CSR_BUILD_FACTOR, CoverageProvenance, ExpandCostInputs, ExpandMode, ExpandPolicy, - HASH_JOIN_POOL_DIVISOR, HASH_JOIN_RATIO, HYDRATE_ROW_RATIO, IndexCoverage, choose_access_path, - choose_expand_mode, direction_probe_factor, estimate_rows, executed_hops, hydrate_chunk_bytes, - scan_row_estimate, should_switch_to_csr, + AccessPath, AntiJoinStrategy, CSR_BUILD_FACTOR, CoverageProvenance, ExpandCostInputs, + ExpandMode, ExpandPolicy, HASH_JOIN_POOL_DIVISOR, HASH_JOIN_RATIO, HYDRATE_ROW_RATIO, + IndexCoverage, chain_rows, choose_access_path, choose_anti_join_strategy, choose_expand_mode, + direction_probe_factor, estimate_rows, executed_hops, hydrate_chunk_bytes, scan_row_estimate, + should_switch_to_csr, }; pub use error::PlanError; pub use explain::Explain; diff --git a/crates/omnigraph-planner/src/mirror.rs b/crates/omnigraph-planner/src/mirror.rs index ceacfeddd..7fa3f1a5d 100644 --- a/crates/omnigraph-planner/src/mirror.rs +++ b/crates/omnigraph-planner/src/mirror.rs @@ -21,7 +21,7 @@ use serde::{Deserialize, Serialize}; use crate::aggregate::AggregateSpec; use crate::bound::{BoundPlan, ValueTable}; -use crate::cost::{AccessPath, ExpandMode, ExpandPolicy}; +use crate::cost::{AccessPath, AntiJoinStrategy, ExpandMode, ExpandPolicy}; use crate::error::PlanError; use crate::logical::{ ColumnRef, GqFilter, KeyJoinKind, Predicate, RuntimeFilterSpec, ScanAccess, ScanSpec, @@ -314,6 +314,7 @@ pub enum NodeMirror { outer_var: String, predicate: SubqueryPredicateMirror, aggregate: Option, + strategy: AntiJoinStrategy, }, OuterReference { outer_var: String, @@ -497,12 +498,14 @@ impl From<&PhysicalNode> for NodeMirror { outer_var, predicate, aggregate, + strategy, } => Self::AntiJoin { input: *input, inner: *inner, outer_var: outer_var.clone(), predicate: SubqueryPredicateMirror::from(predicate), aggregate: *aggregate, + strategy: *strategy, }, PhysicalNode::OuterReference { outer_var } => Self::OuterReference { outer_var: outer_var.clone(), @@ -695,12 +698,14 @@ impl TryFrom for PhysicalNode { outer_var, predicate, aggregate, + strategy, } => Self::AntiJoin { input, inner, outer_var, predicate: SubqueryPredicate::from(predicate), aggregate, + strategy, }, NodeMirror::OuterReference { outer_var } => Self::OuterReference { outer_var }, NodeMirror::RankFuse { diff --git a/crates/omnigraph-planner/src/optimizer.rs b/crates/omnigraph-planner/src/optimizer.rs index 6764a49d9..a8a00f775 100644 --- a/crates/omnigraph-planner/src/optimizer.rs +++ b/crates/omnigraph-planner/src/optimizer.rs @@ -13,12 +13,13 @@ use omnigraph_compiler::traversal::{EDGE_TYPE_COLUMN, EdgeSelection}; use omnigraph_compiler::types::Direction; use crate::cost::{ - AccessPath, CoverageProvenance, ExpandCostInputs, ExpandMode, ExpandPolicy, - HASH_JOIN_POOL_DIVISOR, HYDRATE_ROW_RATIO, IndexCoverage, choose_access_path, - choose_expand_mode, direction_probe_factor, estimate_rows, executed_hops, hydrate_chunk_bytes, - scan_row_estimate, + AccessPath, AntiJoinStrategy, CoverageProvenance, ExpandCostInputs, ExpandMode, ExpandPolicy, + HASH_JOIN_POOL_DIVISOR, HYDRATE_ROW_RATIO, IndexCoverage, chain_rows, choose_access_path, + choose_anti_join_strategy, choose_expand_mode, direction_probe_factor, estimate_rows, + executed_hops, hydrate_chunk_bytes, scan_row_estimate, }; use crate::error::PlanError; +use crate::key_set::{KeySetTree, key_set_tree}; use crate::logical::{ ColumnRef, EDGE_TYPE_MEMBER, GqFilter, IDENTITY_MEMBER, KeyJoinKind, LOGICAL_ID, LogicalId, LogicalNode, LogicalPlan, Predicate, RuntimeFilterKind, RuntimeFilterSpec, ScanSpec, SearchArm, @@ -68,6 +69,7 @@ pub const PASS_LATE_MATERIALIZATION: &str = "late_materialization"; pub const PASS_JOIN_ALGORITHM: &str = "join_algorithm"; pub const PASS_EXPAND_MODE: &str = "expand_mode"; pub const PASS_ACCESS_PATH: &str = "access_path"; +pub const PASS_ANTI_JOIN_STRATEGY: &str = "anti_join_strategy"; /// The optimizer's output: the physical plan with every property declared, /// the passes that fired, and every statistic a pass read. @@ -1135,15 +1137,18 @@ pub fn physical_plan( if address_short_circuit(plan, source) { fired.push(PASS_ADDRESS_SHORT_CIRCUIT); } + let key_sets = key_set_trees(plan, source)?; let mut lowering = Lowering { logical: plan, source, bounds, + key_sets: &key_sets, physical: PhysicalPlan::new(), late_materialization: false, join_algorithm: false, expand_mode: false, access_path: false, + anti_join_strategy: false, csr_cached: false, decisions: Vec::new(), ranking: None, @@ -1156,6 +1161,7 @@ pub fn physical_plan( join_algorithm, expand_mode, access_path, + anti_join_strategy, mut decisions, .. } = lowering; @@ -1177,6 +1183,9 @@ pub fn physical_plan( if access_path { fired.push(PASS_ACCESS_PATH); } + if anti_join_strategy { + fired.push(PASS_ANTI_JOIN_STRATEGY); + } let mut statistics = decisions; statistics.extend(derive_properties(&mut physical, source, bounds)?); let after = physical @@ -1193,6 +1202,29 @@ pub fn physical_plan( }) } +/// The key-set tree of every `AntiJoin` whose block qualifies (`key_set.rs`), +/// built before physical selection, which lowers it in place of the +/// correlated tree when pass 10 prices it lower. +fn key_set_trees( + plan: &LogicalPlan, + source: &dyn PlanSource, +) -> Result, PlanError> { + let mut trees = HashMap::new(); + for (id, node) in plan.live() { + if let LogicalNode::AntiJoin { + inner, + outer_var, + predicate, + .. + } = node + && let Some(tree) = key_set_tree(plan, *inner, outer_var, predicate, source)? + { + trees.insert(id, tree); + } + } + Ok(trees) +} + /// The nodes of the pipeline under `root` that constrain its rows: every /// node reachable from `root` without entering an `AntiJoin`'s inner tree. fn top_level(plan: &PhysicalPlan, root: NodeId) -> Vec { @@ -2229,9 +2261,13 @@ fn address_short_circuit(plan: &mut LogicalPlan, source: &dyn PlanSource) -> boo /// materialization (pass 6) at every `Join` and the join algorithm rule /// (pass 7): a fragment-scoped right input builds, everything else streams. struct Lowering<'a> { + /// The arena `lower` reads: the plan's, or a key-set tree's while one is + /// lowered in place of its block's correlated tree. logical: &'a LogicalPlan, source: &'a dyn PlanSource, bounds: &'a Bounds, + /// The key-set tree of each qualifying `AntiJoin`, by its logical id. + key_sets: &'a HashMap, physical: PhysicalPlan, late_materialization: bool, join_algorithm: bool, @@ -2239,6 +2275,8 @@ struct Lowering<'a> { expand_mode: bool, /// Pass 9 fired: a dependent scan's access path came from the cost model. access_path: bool, + /// Pass 10 fired: a correlated block's strategy came from the cost model. + anti_join_strategy: bool, /// An earlier lowered node of this plan realizes the query's CSR (an /// `Expand` in `Csr` mode or an `AntiJoin`), so a later `Expand` reuses /// it for free. Lowering order is execution order (inputs first). @@ -2744,14 +2782,31 @@ impl Lowering<'_> { predicate, } => { let lowered = self.lower(*input)?; - let inner_lowered = self.lower(*inner)?; - self.csr_cached = true; + let key_set = self.key_sets.get(&id); + let strategy = match key_set { + Some(tree) => self.anti_join_strategy(*input, *inner, outer_var, tree), + None => AntiJoinStrategy::Correlated, + }; + let inner_lowered = match (strategy, key_set) { + (AntiJoinStrategy::KeySet, Some(tree)) => { + let logical = std::mem::replace(&mut self.logical, &tree.plan); + let lowered = self.lower(tree.root); + self.logical = logical; + lowered? + } + _ => { + let lowered = self.lower(*inner)?; + self.csr_cached = true; + lowered + } + }; Ok(self.physical.add(PhysicalNode::AntiJoin { input: lowered, inner: inner_lowered, outer_var: outer_var.clone(), predicate: predicate.clone(), aggregate: crate::plan_block_aggregate(&predicate.left)?, + strategy, })) } LogicalNode::OuterReference { outer_var } => { @@ -3147,6 +3202,33 @@ impl Lowering<'_> { Ok(access) } + /// Pass 10, the strategy of a correlated block with a key-set tree: the + /// cost model over the rows each inner tree produces, the correlated one + /// from the outer input's row-count estimate (`chain_rows`); `Correlated` + /// when an estimate is unknown. + fn anti_join_strategy( + &mut self, + input: LogicalId, + inner: LogicalId, + outer_var: &str, + tree: &KeySetTree, + ) -> AntiJoinStrategy { + let outer_rows = estimate_rows(self.logical, input, self.source); + let (Some(correlated), Some(key_set)) = ( + chain_rows(self.logical, inner, self.source, outer_rows), + chain_rows(&tree.plan, tree.root, self.source, None), + ) else { + return AntiJoinStrategy::Correlated; + }; + self.anti_join_strategy = true; + self.decisions.push(StatisticSource { + statistic: format!("anti_join_rows(${outer_var})"), + value: format!("correlated {correlated} estimated, key_set {key_set} estimated"), + origin: "row-count estimates of the block's two inner trees".to_string(), + }); + choose_anti_join_strategy(correlated, key_set) + } + fn set_ordered(&mut self, id: NodeId, value: bool) { if let Some(PhysicalNode::Scan { ordered, .. }) = self.physical.node_mut(id) { *ordered = value; diff --git a/crates/omnigraph-planner/src/physical.rs b/crates/omnigraph-planner/src/physical.rs index 11d7ed65b..3047b8ce1 100644 --- a/crates/omnigraph-planner/src/physical.rs +++ b/crates/omnigraph-planner/src/physical.rs @@ -10,7 +10,7 @@ use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use crate::aggregate::AggregateSpec; -use crate::cost::{AccessPath, ExpandMode, ExpandPolicy}; +use crate::cost::{AccessPath, AntiJoinStrategy, ExpandMode, ExpandPolicy}; use crate::error::PlanError; use crate::logical::{ ColumnRef, KeyJoinKind, ScanSpec, filters_json, metadata_count_json, ordering_text, scan_json, @@ -526,14 +526,19 @@ pub enum PhysicalNode { policy: ExpandPolicy, versions: BTreeMap>, }, + /// A correlated block over `input`. Under `Correlated` the `inner` tree's + /// leaf is an `OuterReference` to the input rows; under `KeySet` the inner + /// tree reads no outer rows and ends on `outer_var`'s identity, the key + /// the input rows are kept by. AntiJoin { input: NodeId, inner: NodeId, outer_var: String, predicate: SubqueryPredicate, aggregate: Option, + strategy: AntiJoinStrategy, }, - /// The enclosing rows, the leaf of an `AntiJoin` inner tree. + /// The enclosing rows, the leaf of a correlated `AntiJoin` inner tree. OuterReference { outer_var: String, }, @@ -994,10 +999,12 @@ impl PhysicalPlan { outer_var, predicate, aggregate, + strategy, .. } => json!({ "node": node.name(), "outer_var": outer_var, + "strategy": strategy, "predicate": predicate.to_string(), "aggregate": crate::typed::block_aggregate(&predicate.left, *aggregate), "typed_left": crate::typed::block(&predicate.left), diff --git a/crates/omnigraph-planner/tests/integration/bound_plan.rs b/crates/omnigraph-planner/tests/integration/bound_plan.rs index 3928f302a..0f4465463 100644 --- a/crates/omnigraph-planner/tests/integration/bound_plan.rs +++ b/crates/omnigraph-planner/tests/integration/bound_plan.rs @@ -145,12 +145,12 @@ fn saved_plan_version_refuses_both_legacy_and_future_readers() { values: Default::default(), }; let encoded = serde_json::to_value(&bound).unwrap(); - assert_eq!(encoded["bound_plan_version"], 7); + assert_eq!(encoded["bound_plan_version"], 8); assert!(encoded.get("plan").is_none()); assert!(encoded["body"]["plan"].is_object()); assert_eq!(round_trip(&bound), bound); - for version in [1, 2, 3, 4, 5, 6] { + for version in [1, 2, 3, 4, 5, 6, 7] { let mut older = encoded.clone(); older["bound_plan_version"] = serde_json::json!(version); assert!( @@ -164,12 +164,12 @@ fn saved_plan_version_refuses_both_legacy_and_future_readers() { let error = serde_json::from_value::(legacy).unwrap_err(); assert!(error.to_string().contains("regenerate"), "{error}"); let mut future = encoded.clone(); - future["bound_plan_version"] = serde_json::json!(8); + future["bound_plan_version"] = serde_json::json!(9); let error = serde_json::from_value::(future).unwrap_err(); assert!( error .to_string() - .contains("unsupported bound plan version 8"), + .contains("unsupported bound plan version 9"), "{error}" ); @@ -796,6 +796,7 @@ fn block_bound(left: omnigraph_compiler::ir::BlockAggregateExpr, right: IRExpr) op: CompOp::Gt, right, }, + strategy: omnigraph_planner::AntiJoinStrategy::KeySet, }); plan.set_root(root); BoundPlan { @@ -1060,7 +1061,12 @@ fn block_mirrors_and_explain_preserve_every_nested_cast_in_order() { panic!("block"); }; assert_eq!(predicate, back); + assert_eq!( + restored, bound, + "the block's strategy survives the saved plan" + ); let explain = restored.plan.to_json(); + assert_eq!(explain["strategy"], "key_set"); assert_eq!(explain["predicate"], "count > $bound"); assert_eq!(explain["typed_left"]["type"], "F64"); assert_eq!(explain["typed_left"]["args"][0]["type"], "exact_integer"); diff --git a/crates/omnigraph-planner/tests/integration/lower_walk.rs b/crates/omnigraph-planner/tests/integration/lower_walk.rs index 11eddaca0..d976a7269 100644 --- a/crates/omnigraph-planner/tests/integration/lower_walk.rs +++ b/crates/omnigraph-planner/tests/integration/lower_walk.rs @@ -349,6 +349,7 @@ fn an_anti_join_hands_over_its_outer_input_before_the_inner_tree_is_lowered() { outer_var: "o".to_string(), predicate: omnigraph_compiler::ir::SubqueryPredicate::not_exists(), aggregate: None, + strategy: omnigraph_planner::AntiJoinStrategy::Correlated, }); plan.set_root(anti); diff --git a/crates/omnigraph-planner/tests/query_plan/cost.rs b/crates/omnigraph-planner/tests/query_plan/cost.rs index 310f6f377..6c6288aba 100644 --- a/crates/omnigraph-planner/tests/query_plan/cost.rs +++ b/crates/omnigraph-planner/tests/query_plan/cost.rs @@ -1078,3 +1078,220 @@ fn hydration_keeps_every_column_something_besides_the_output_reads() { ]) ); } + +/// `$x.prop = "value"`, a bound equality on one binding. +fn equals(variable: &str, property: &str, value: &str) -> IRExpr { + IRExpr::comparison( + prop(variable, property), + CompOp::Eq, + IRExpr::Literal( + Literal::String(value.to_string()), + value_type(ScalarType::String, false), + ), + ) +} + +/// `not { $a knows $m $m knows $k }` over every `T` as `$a`, +/// `$k`'s own filters `k_filters`. +fn negated_two_hops( + outer_filters: Vec, + k_filters: Vec, + inner_filters: Vec, +) -> Operation { + let mut inner = vec![expand("a", "m", vec![]), expand("m", "k", k_filters)]; + inner.extend(inner_filters); + let outer = IROp::NodeScan { + variable: "a".to_string(), + type_name: "T".to_string(), + filters: outer_filters, + }; + ir( + vec![ + outer, + IROp::AntiJoin { + outer_var: "a".to_string(), + inner, + predicate: omnigraph_compiler::ir::SubqueryPredicate::not_exists(), + }, + ], + vec![prop("a", "slug")], + vec![], + ) +} + +/// 1,000 `T` rows; `knows` fans out ten per source going out and `in_fanout` +/// per destination coming in. +fn anti_join_source(in_fanout: u64) -> MemorySource { + MemorySource::default() + .with_node_type("T", node_type("T", Some(1_000))) + .with_expand_statistics("knows", Direction::Out, knows_statistics(1_000)) + .with_expand_statistics( + "knows", + Direction::In, + ExpandStatistics { + edge_count: 10_000, + src_node_count: 10_000 / in_fanout, + dst_node_count: 1_000, + max_frontier_cap: 1024, + max_hops_cap: 6, + }, + ) +} + +/// The block's strategy and its inner tree's nodes, in pre-order. +fn anti_join_of(plan: &PhysicalPlan) -> (omnigraph_planner::AntiJoinStrategy, Vec<&PhysicalNode>) { + let (strategy, inner) = plan + .live() + .find_map(|(_, node)| match node { + PhysicalNode::AntiJoin { + strategy, inner, .. + } => Some((*strategy, *inner)), + _ => None, + }) + .expect("the plan holds the block's AntiJoin"); + let mut nodes = Vec::new(); + let mut pending = vec![inner]; + while let Some(id) = pending.pop() { + let node = plan.node(id).unwrap(); + nodes.push(node); + pending.extend(node.inputs().into_iter().rev()); + } + (strategy, nodes) +} + +/// A block bounded by a key equality on its own binding runs from it when its +/// chain produces fewer rows: one `$k`, ten `$m`, a hundred `$a` keys against +/// the correlated 1,000 outer rows expanded twice. The key-set tree is a +/// table scan of `$k` under the pushed equality, then both hops reversed, +/// ending on `$a` with no scan of it and no `OuterReference`. +#[test] +fn bounded_block_runs_from_its_key_set_when_its_rows_are_fewer() { + use omnigraph_planner::AntiJoinStrategy; + + let source = anti_join_source(10); + let op = negated_two_hops(vec![], vec![equals("k", "slug", "k1")], vec![]); + let mut plan = resolve(&op, &source).expect("resolve"); + let fired = rewrite(&mut plan, &source).expect("rewrite"); + let optimized = + omnigraph_planner::physical_plan(&mut plan, &source, &bounds(), fired).expect("lower"); + assert!(optimized.fired.contains(&"anti_join_strategy")); + assert!( + optimized.statistics.iter().any(|statistic| { + statistic.statistic == "anti_join_rows($a)" + && statistic.value == "correlated 4001 estimated, key_set 121 estimated" + }), + "{:?}", + optimized.statistics + ); + let (strategy, inner) = anti_join_of(&optimized.physical); + assert_eq!(strategy, AntiJoinStrategy::KeySet); + let shape: Vec = inner + .iter() + .map(|node| match node { + PhysicalNode::Scan { + source: omnigraph_planner::ScanInput::Table, + spec, + .. + } => format!( + "table scan ${} filtered {}", + spec.binding.as_deref().unwrap_or_default(), + spec.filter.is_some() + ), + PhysicalNode::Scan { spec, .. } => format!( + "dependent scan ${}", + spec.binding.as_deref().unwrap_or_default() + ), + PhysicalNode::Expand { + src, dst, edges, .. + } => format!( + "expand ${src} {:?} ${dst}", + edges.named().map(|member| member.direction) + ), + other => other.name().to_string(), + }) + .collect(); + assert_eq!( + shape, + [ + "expand $m Some(In) $a", + "dependent scan $m", + "expand $k Some(In) $m", + "table scan $k filtered true", + ] + ); + let json = optimized.physical.to_json(); + assert_eq!( + json["inputs"][0]["inputs"][0]["strategy"], "key_set", + "{json}" + ); +} + +/// The same block over one outer row stays correlated: its 122 rows beat the +/// 1,201 the key set reaches when each destination has a hundred sources. +#[test] +fn bounded_block_stays_correlated_when_the_outer_side_is_smaller() { + use omnigraph_planner::AntiJoinStrategy; + + let source = anti_join_source(100); + let op = negated_two_hops( + vec![equals("a", "slug", "a1")], + vec![equals("k", "slug", "k1")], + vec![], + ); + let mut plan = resolve(&op, &source).expect("resolve"); + let fired = rewrite(&mut plan, &source).expect("rewrite"); + let optimized = + omnigraph_planner::physical_plan(&mut plan, &source, &bounds(), fired).expect("lower"); + assert!(optimized.fired.contains(&"anti_join_strategy")); + assert!( + optimized.statistics.iter().any(|statistic| { + statistic.statistic == "anti_join_rows($a)" + && statistic.value == "correlated 122 estimated, key_set 1201 estimated" + }), + "{:?}", + optimized.statistics + ); + let (strategy, inner) = anti_join_of(&optimized.physical); + assert_eq!(strategy, AntiJoinStrategy::Correlated); + assert!( + inner + .iter() + .any(|node| matches!(node, PhysicalNode::OuterReference { .. })) + ); +} + +/// No key set without a bounded binding, or when the block reads the outer +/// rows beyond the correlation binding's identity: the planner does not +/// price it, and the block keeps its correlated tree. +#[test] +fn unbounded_or_outer_reading_blocks_stay_correlated() { + use omnigraph_planner::AntiJoinStrategy; + + let reads_outer = IROp::Filter(IRExpr::comparison( + prop("m", "state"), + CompOp::Eq, + prop("a", "state"), + )); + for (what, op) in [ + ( + "a non-key filter bounds nothing", + negated_two_hops(vec![], vec![equals("k", "state", "open")], vec![]), + ), + ( + "a filter reads the outer row", + negated_two_hops(vec![], vec![equals("k", "slug", "k1")], vec![reads_outer]), + ), + ] { + let source = anti_join_source(10); + let (plan, fired) = physical(&op, &source); + assert!(!fired.contains(&"anti_join_strategy"), "{what}: {fired:?}"); + let (strategy, inner) = anti_join_of(&plan); + assert_eq!(strategy, AntiJoinStrategy::Correlated, "{what}"); + assert!( + inner + .iter() + .any(|node| matches!(node, PhysicalNode::OuterReference { .. })), + "{what}" + ); + } +} diff --git a/crates/omnigraph/src/engine/lower.rs b/crates/omnigraph/src/engine/lower.rs index 12edb9b4f..f9a78ade1 100644 --- a/crates/omnigraph/src/engine/lower.rs +++ b/crates/omnigraph/src/engine/lower.rs @@ -24,9 +24,9 @@ use omnigraph_compiler::ir::SubqueryPredicate; use omnigraph_compiler::traversal::EDGE_TYPE_COLUMN; use omnigraph_planner::logical::{EDGE_TYPE_MEMBER, IDENTITY_MEMBER}; use omnigraph_planner::{ - Accumulator, AggregateFields, BoundPlan, ColumnRef, ContainsJoinFields, ExpandFields, - HashJoinFields, HydratedBinding, Lower, NodeId, PhysicalNode, PhysicalPlan, PlanError, - Predicate, ROW_ADDRESS_PREFIX, RankArm, RankFuseFields, RankKind, RankedAccess, + Accumulator, AggregateFields, AntiJoinStrategy, BoundPlan, ColumnRef, ContainsJoinFields, + ExpandFields, HashJoinFields, HydratedBinding, Lower, NodeId, PhysicalNode, PhysicalPlan, + PlanError, Predicate, ROW_ADDRESS_PREFIX, RankArm, RankFuseFields, RankKind, RankedAccess, RuntimeFilterKind, RuntimeFilterSpec, ScanInput, ScanSpec, SideId, SortMergeJoinFields, ValueTable, }; @@ -720,6 +720,7 @@ impl Lower for Walk<'_, '_> { inner: inner_id, predicate, aggregate, + strategy, .. } = self.lowering.node(id)? else { @@ -734,9 +735,12 @@ impl Lower for Walk<'_, '_> { ) .into()); } - let bulk = self - .lowering - .bulk_row_count(*inner_id, outer_var, predicate)?; + let bulk = match strategy { + AntiJoinStrategy::Correlated => self + .lowering + .bulk_row_count(*inner_id, outer_var, predicate)?, + AntiJoinStrategy::KeySet => None, + }; let mask = AntiJoinMaskExec::new( outer, inner, @@ -746,6 +750,7 @@ impl Lower for Walk<'_, '_> { Arc::clone(self.lowering.params()), scope.tag_column, scope.slot, + *strategy, bulk, Arc::clone(&self.scope.env), ); diff --git a/crates/omnigraph/src/engine/operators/anti_join.rs b/crates/omnigraph/src/engine/operators/anti_join.rs index 5574afbe1..cb786eedf 100644 --- a/crates/omnigraph/src/engine/operators/anti_join.rs +++ b/crates/omnigraph/src/engine/operators/anti_join.rs @@ -1,27 +1,35 @@ //! `AntiJoinMaskExec`: a correlated block (`not { … }`, `count { … } > 2`, //! `sum($m.size) { … } > 100`) as a pipeline breaker over the outer wide -//! batch. The inner tree runs once as a lowered plan whose leaf, -//! `OuterReferenceExec`, reads the tagged outer batch from a slot this -//! operator fills before executing it; `SubqueryAggregate` then decides per -//! outer row. The shapes `Lowering::bulk_row_count` selects are answered by -//! the bulk CSR degree check instead (`bulk_anti_join_mask`). +//! batch. Under the `Correlated` strategy the inner tree runs once as a +//! lowered plan whose leaf, `OuterReferenceExec`, reads the tagged outer batch +//! from a slot this operator fills before executing it; `SubqueryAggregate` +//! then decides per outer row. The shapes `Lowering::bulk_row_count` selects +//! are answered by the bulk CSR degree check instead (`bulk_anti_join_mask`). +//! Under `KeySet` the inner tree reads no outer rows and ends on the +//! correlation binding: its rows fold into one aggregate per outer key as +//! they stream, and each outer row takes its key's verdict (`key_set_mask`). use datafusion::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricsSet}; +use std::collections::HashMap; use std::fmt; use std::sync::{Arc, Mutex}; -use arrow_array::{ArrayRef, RecordBatch, UInt32Array}; +use arrow_array::{Array, ArrayRef, BooleanArray, RecordBatch, StringArray, UInt32Array}; use arrow_schema::{DataType, Field, Schema, SchemaRef}; use datafusion::common::Result as DfResult; use datafusion::execution::TaskContext; use datafusion::physical_plan::{ DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, SendableRecordBatchStream, }; +use futures::StreamExt; use omnigraph_compiler::ir::{ParamMap, SubqueryPredicate}; use omnigraph_compiler::types::Direction; -use omnigraph_planner::AggregateSpec; +use omnigraph_planner::{AggregateSpec, AntiJoinStrategy}; -use super::subquery_aggregate::{RowCountPredicate, SubqueryAggregate, absorb_inner_batches}; +use super::memory::WorkMemory; +use super::subquery_aggregate::{ + RowCountPredicate, SubqueryAggregate, ValueAccumulator, absorb_inner_batches, +}; use super::{GraphEnv, breaker_properties, breaker_stream, drain, drain_one, external, polled}; use crate::engine::graph::bulk_anti_join_mask; use crate::error::{OmniError, Result}; @@ -140,6 +148,7 @@ pub(crate) struct AntiJoinMaskExec { params: Arc, tag_column: String, slot: Arc, + strategy: AntiJoinStrategy, /// The edge the bulk CSR check answers (`Lowering::bulk_row_count`). bulk: Option<(String, Direction)>, env: Arc, @@ -158,6 +167,7 @@ impl AntiJoinMaskExec { params: Arc, tag_column: String, slot: Arc, + strategy: AntiJoinStrategy, bulk: Option<(String, Direction)>, env: Arc, ) -> Self { @@ -171,6 +181,7 @@ impl AntiJoinMaskExec { params, tag_column, slot, + strategy, bulk, env, properties: breaker_properties(schema), @@ -212,6 +223,7 @@ impl fmt::Debug for AntiJoinMaskExec { f.debug_struct("AntiJoinMaskExec") .field("outer_var", &self.outer_var) .field("predicate", &self.predicate.to_string()) + .field("strategy", &self.strategy) .field("bulk", &self.bulk) .finish_non_exhaustive() } @@ -224,6 +236,9 @@ impl DisplayAs for AntiJoinMaskExec { "AntiJoinMaskExec: ${} {}", self.outer_var, self.predicate )?; + if self.strategy == AntiJoinStrategy::KeySet { + write!(f, ", key_set")?; + } if let Some((edge_type, direction)) = &self.bulk { write!(f, ", bulk={edge_type} {direction:?}")?; } @@ -268,6 +283,7 @@ impl ExecutionPlan for AntiJoinMaskExec { Arc::clone(&self.params), self.tag_column.clone(), Arc::clone(&self.slot), + self.strategy, self.bulk.clone(), Arc::clone(&self.env), ))) @@ -290,6 +306,7 @@ impl ExecutionPlan for AntiJoinMaskExec { let tag_column = self.tag_column.clone(); let tagged = tagged_schema(&schema, &tag_column); let slot = Arc::clone(&self.slot); + let strategy = self.strategy; let bulk = self.bulk.clone(); let env = Arc::clone(&self.env); let declared = Arc::clone(&schema); @@ -307,6 +324,21 @@ impl ExecutionPlan for AntiJoinMaskExec { if num_rows == 0 { return Ok(wide); } + if strategy == AntiJoinStrategy::KeySet { + let key_column = + format!("{outer_var}.{}", env.catalog.system_columns.id); + let keep = key_set_mask( + &wide, + inner.execute(0, inner_ctx)?, + &key_column, + &predicate, + aggregate_spec, + ¶ms, + &reservation, + ) + .await?; + return reservation.filter(&wide, &keep); + } if let Some((edge_type, direction)) = &bulk && let Some(row_count) = RowCountPredicate::resolve(&predicate, ¶ms).map_err(external)? @@ -360,3 +392,83 @@ impl ExecutionPlan for AntiJoinMaskExec { Ok(polled(&self.metrics, stream)) } } + +/// The keep mask of a `KeySet` block over the outer `wide` batch: each outer +/// key gets one aggregate, the inner rows fold into their key's as they +/// stream (an inner key no outer row holds goes to a discarded group), and +/// every outer row takes its key's verdict. A null outer key takes the +/// verdict of no matches, as its correlated expansion matches nothing. +async fn key_set_mask( + wide: &RecordBatch, + mut inner: SendableRecordBatchStream, + key_column: &str, + predicate: &SubqueryPredicate, + aggregate_spec: Option, + params: &ParamMap, + reservation: &WorkMemory, +) -> DfResult { + let outer_keys = key_array(wide, key_column).map_err(external)?; + let num_rows = wide.num_rows(); + reservation.entries::<(&str, u32)>(num_rows)?; + reservation.entries::(num_rows)?; + let mut groups: HashMap<&str, u32> = HashMap::new(); + let mut row_groups = Vec::with_capacity(num_rows); + for row in 0..num_rows { + row_groups.push(outer_keys.is_valid(row).then(|| { + let next = groups.len() as u32; + *groups.entry(outer_keys.value(row)).or_insert(next) + })); + } + let unmatched = groups.len() as u32; + let discarded = unmatched + 1; + let group_count = groups.len() + 2; + reservation.entries::(group_count)?; + if SubqueryAggregate::tracks_values(aggregate_spec) { + reservation.entries::(group_count)?; + } + let mut aggregate = + SubqueryAggregate::new(predicate, aggregate_spec, params, group_count).map_err(external)?; + while let Some(batch) = inner.next().await { + let batch = batch?; + reservation.metric("input_rows", batch.num_rows()); + if batch.num_rows() == 0 { + continue; + } + let inner_keys = key_array(&batch, key_column).map_err(external)?; + let tags = UInt32Array::from_iter_values((0..batch.num_rows()).map(|row| { + inner_keys + .is_valid(row) + .then(|| groups.get(inner_keys.value(row)).copied()) + .flatten() + .unwrap_or(discarded) + })); + let values = predicate + .left + .arg() + .map(|arg| crate::engine::expr::evaluate_expr(&batch, arg, params)) + .transpose() + .map_err(external)?; + aggregate.absorb(&tags, values.as_ref()).map_err(external)?; + } + let verdicts = aggregate.keep_mask().map_err(external)?; + Ok(row_groups + .into_iter() + .map(|group| Some(verdicts.value(group.unwrap_or(unmatched) as usize))) + .collect()) +} + +/// The correlation binding's identity column of a key-set batch. +fn key_array<'a>(batch: &'a RecordBatch, key_column: &str) -> Result<&'a StringArray> { + batch + .column_by_name(key_column) + .ok_or_else(|| { + OmniError::manifest_internal(format!( + "key-set anti-join batch carries no '{key_column}' column" + )) + })? + .as_any() + .downcast_ref::() + .ok_or_else(|| { + OmniError::manifest_internal(format!("'{key_column}' key column is not Utf8")) + }) +} diff --git a/crates/omnigraph/src/engine/report/tests.rs b/crates/omnigraph/src/engine/report/tests.rs index 619e71a72..3abde3e2a 100644 --- a/crates/omnigraph/src/engine/report/tests.rs +++ b/crates/omnigraph/src/engine/report/tests.rs @@ -184,6 +184,16 @@ query nobody_older() { } return { $p.name } } +query skips_d1() { + match { + $p: Person + not { + $p likes $d + $d: Doc { title: "d1" } + } + } + return { $p.name } +} query none_ordered() { match { $p: Person } return { $p.name } @@ -795,6 +805,52 @@ async fn anti_join_inner_tree_is_skipped_under_the_bulk_check_and_executed_witho } } +/// A key-set block's inner tree reads no outer rows: no `OuterReference` +/// node, its table scan of the bounded binding and its reversed expand both +/// executed, and the `AntiJoinMaskExec` keeps the persons d1 never met. +#[tokio::test] +async fn key_set_anti_join_executes_its_inner_tree_without_an_outer_leaf() { + let dir = tempfile::tempdir().unwrap(); + let db = graph(&dir, PEOPLE_SCHEMA, PEOPLE_SEED).await; + let (rows, run) = run(&db, PEOPLE_QUERIES, "skips_d1", &ParamMap::new()).await; + assert_eq!(rows, 2); + assert_covers(&run); + let anti = node_id(&run.plan, |node| { + matches!( + node, + PhysicalNode::AntiJoin { + strategy: omnigraph_planner::AntiJoinStrategy::KeySet, + .. + } + ) + }); + assert_eq!(operator(&run, anti), "AntiJoinMaskExec"); + assert_eq!(row(&run, anti).status, RowStatus::Executed); + assert!( + !run.plan + .live() + .any(|(_, node)| matches!(node, PhysicalNode::OuterReference { .. })) + ); + let root = node_id(&run.plan, |node| { + matches!( + node, + PhysicalNode::Scan { + source: omnigraph_planner::ScanInput::Table, + spec, + .. + } if spec.binding.as_deref() == Some("d") + ) + }); + assert_eq!(operator(&run, root), "ScanExec"); + assert_eq!(row(&run, root).status, RowStatus::Executed); + let reversed = node_id( + &run.plan, + |node| matches!(node, PhysicalNode::Expand { src, dst, .. } if src == "d" && dst == "p"), + ); + assert_eq!(operator(&run, reversed), "ExpandExec"); + assert_eq!(row(&run, reversed).status, RowStatus::Executed); +} + #[tokio::test] async fn zero_limit_executes_the_root_alone_and_skips_every_other_node() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/omnigraph/tests/integration/anti_join_cost.rs b/crates/omnigraph/tests/integration/anti_join_cost.rs new file mode 100644 index 000000000..17c06bcd8 --- /dev/null +++ b/crates/omnigraph/tests/integration/anti_join_cost.rs @@ -0,0 +1,269 @@ +//! Decision instrument for a correlated block's strategy (`key_set` against +//! `correlated`): the rows every operator emits, those of the block's inner +//! tree, and the wall time of one channel lookalike read with and without its +//! `not { }` exclusion. The block is +//! forced correlated by a conjunct that reads `$x` and always holds, so one +//! binary measures both strategies over the same snapshot. Each read opens +//! the graph cold. +//! +//! By default the graph is synthetic and shaped like a CRM channel: a channel +//! watching 140 sources, 2,000 identities that each star two of them and +//! touch 15 other sources, and an excluded source starred by 1,600 of them, +//! loaded in 48 commits with no indexes, so every table has about 48 +//! fragments and every dependent scan reads them all. With +//! `OMNIGRAPH_ANTI_JOIN_STORE` set to a graph URI of the same types, the +//! three reads run there instead, with `OMNIGRAPH_ANTI_JOIN_CHANNEL`, +//! `OMNIGRAPH_ANTI_JOIN_EXCLUDE` and `OMNIGRAPH_ANTI_JOIN_SINCE`. + +use std::time::Instant; + +use omnigraph::db::Omnigraph; +use omnigraph::loader::LoadMode; +use omnigraph_compiler::ir::ParamMap; +use omnigraph_compiler::query::ast::Literal; +use omnigraph_planner::{AntiJoinStrategy, PhysicalNode}; + +use crate::helpers::*; + +const SCHEMA: &str = r#" +node Channel { slug: String @key } +node Source { slug: String @key } +node Touchpoint { + slug: String @key + action: String + observed_at: DateTime +} +node ExternalID { slug: String @key } +edge ChannelWatchesSource: Channel -> Source +edge TouchpointOnSource: Touchpoint -> Source +edge TouchpointByIdentity: Touchpoint -> ExternalID +"#; + +const OUTER: &str = r#" + $c: Channel { slug: $channel } + $s: Source + $c channelWatchesSource $s + $t: Touchpoint { action: "star" } + $t touchpointOnSource $s + $t.observed_at >= $since + $x: ExternalID + $t touchpointByIdentity $x +"#; + +const BLOCK: &str = r#" + not { + $s0: Source { slug: $exclude } + $t0: Touchpoint + $t0 touchpointOnSource $s0 + $t0 touchpointByIdentity $x + FORCE + } +"#; + +const RETURN: &str = r#" + return { $x.slug, count($t) as repos, max($t.observed_at) as latest } + order { repos desc } + limit 500 +"#; + +/// The read with no block, with the block, and with the block forced +/// correlated (`$x.slug = $x.slug` reads the outer row). +fn reads() -> [(&'static str, String); 3] { + let query = |block: &str| { + format!( + "query lookalikes($channel: String, $since: DateTime, $exclude: String) {{\n match {{{OUTER}{block} }}{RETURN}}}" + ) + }; + [ + ("no block", query("").replace(", $exclude: String", "")), + ("key_set", query(&BLOCK.replace("FORCE", ""))), + ( + "correlated", + query(&BLOCK.replace("FORCE", "$x.slug = $x.slug")), + ), + ] +} + +/// A small deterministic generator: the fixture is the same on every run. +struct Lcg(u64); + +impl Lcg { + fn below(&mut self, bound: usize) -> usize { + self.0 = self + .0 + .wrapping_mul(6_364_136_223_846_793_005) + .wrapping_add(1_442_695_040_888_963_407); + ((self.0 >> 33) as usize) % bound + } +} + +/// The synthetic graph as JSONL batches, one commit each. Every node is +/// declared just before its first use, so a sequential split never names an +/// endpoint a later commit adds, and every table gathers about one fragment +/// per commit. +fn synthetic_batches(commits: usize) -> Vec { + const WATCHED: usize = 140; + const OTHER: usize = 2_500; + const IDENTITIES: usize = 2_000; + const STARS: usize = 2; + const ELSEWHERE: usize = 15; + const EXCLUDED: usize = 1_600; + let mut random = Lcg(7); + let mut lines = vec![ + r#"{"type":"Channel","data":{"slug":"channel"}}"#.to_string(), + r#"{"type":"Source","data":{"slug":"excluded"}}"#.to_string(), + ]; + let mut declared = vec![false; OTHER]; + let mut touchpoints = 0usize; + let mut touch = |lines: &mut Vec, source: &str, identity: usize, day: usize| { + let slug = format!("t{touchpoints}"); + touchpoints += 1; + lines.push(format!( + r#"{{"type":"Touchpoint","data":{{"slug":"{slug}","action":"star","observed_at":"2026-{:02}-{:02}T00:00:00Z"}}}}"#, + 1 + day % 9, + 1 + day % 28 + )); + lines.push(format!( + r#"{{"edge":"TouchpointOnSource","from":"{slug}","to":"{source}"}}"# + )); + lines.push(format!( + r#"{{"edge":"TouchpointByIdentity","from":"{slug}","to":"x{identity}"}}"# + )); + }; + for identity in 0..IDENTITIES { + lines.push(format!( + r#"{{"type":"ExternalID","data":{{"slug":"x{identity}"}}}}"# + )); + if identity < WATCHED { + lines.push(format!( + r#"{{"type":"Source","data":{{"slug":"w{identity}"}}}}"# + )); + lines.push(format!( + r#"{{"edge":"ChannelWatchesSource","from":"channel","to":"w{identity}"}}"# + )); + } + for _ in 0..STARS { + let source = format!("w{}", random.below(WATCHED.min(identity + 1))); + let day = random.below(250); + touch(&mut lines, &source, identity, day); + } + for _ in 0..ELSEWHERE { + let other = random.below(OTHER); + if !std::mem::replace(&mut declared[other], true) { + lines.push(format!( + r#"{{"type":"Source","data":{{"slug":"o{other}"}}}}"# + )); + } + let day = random.below(250); + touch(&mut lines, &format!("o{other}"), identity, day); + } + } + for identity in 0..EXCLUDED { + let day = random.below(250); + touch(&mut lines, "excluded", identity, day); + } + lines + .chunks(lines.len().div_ceil(commits)) + .map(|chunk| chunk.join("\n")) + .collect() +} + +fn env_or(name: &str, default: &str) -> String { + std::env::var(name).unwrap_or_else(|_| default.to_string()) +} + +/// The rows every operator of a run's `plan` emitted by its `report`, and +/// those of the operators under the plan's `AntiJoin` inner tree. +fn rows_through(plan: &omnigraph_planner::PhysicalPlan, report: &serde_json::Value) -> (u64, u64) { + let rows_of = |id: omnigraph_planner::NodeId| { + report["rows"] + .as_array() + .unwrap() + .iter() + .find(|row| row["id"] == serde_json::json!(id)) + .and_then(|row| row["attempts"].as_array()?.last()?["actual_rows"].as_u64()) + .unwrap_or(0) + }; + let all = plan.live().map(|(id, _)| rows_of(id)).sum(); + let mut inner = 0; + let mut pending: Vec<_> = plan + .live() + .filter_map(|(_, node)| match node { + PhysicalNode::AntiJoin { inner, .. } => Some(*inner), + _ => None, + }) + .collect(); + while let Some(id) = pending.pop() { + inner += rows_of(id); + pending.extend(plan.node(id).unwrap().inputs()); + } + (all, inner) +} + +#[tokio::test] +#[ignore = "instrument: rows and wall time of a correlated block's two strategies over a fragmented, unindexed CRM-shaped graph (or OMNIGRAPH_ANTI_JOIN_STORE)"] +async fn correlated_block_strategy_rows_and_time() { + let dir = tempfile::tempdir().unwrap(); + let uri = match std::env::var("OMNIGRAPH_ANTI_JOIN_STORE") { + Ok(uri) => uri, + Err(_) => { + let uri = dir.path().to_str().unwrap().to_string(); + let db = session(Omnigraph::init(&uri, SCHEMA).await.unwrap()); + for batch in synthetic_batches(48) { + db.load_jsonl(&batch, LoadMode::Append).await.unwrap(); + } + uri + } + }; + let mut params = ParamMap::new(); + params.insert( + "channel".to_string(), + Literal::String(env_or("OMNIGRAPH_ANTI_JOIN_CHANNEL", "channel")), + ); + params.insert( + "exclude".to_string(), + Literal::String(env_or("OMNIGRAPH_ANTI_JOIN_EXCLUDE", "excluded")), + ); + params.insert( + "since".to_string(), + Literal::DateTime(env_or("OMNIGRAPH_ANTI_JOIN_SINCE", "2025-11-01T00:00:00Z")), + ); + let mut answers = Vec::new(); + for (label, query) in reads() { + let mut query_params = params.clone(); + if label == "no block" { + query_params.remove("exclude"); + } + let db = session(Omnigraph::open(&uri).await.unwrap()); + let started = Instant::now(); + let executed = db + .query_inspected("main", &query, "lookalikes", &query_params) + .await + .unwrap(); + let elapsed = started.elapsed(); + let strategy = executed.plan.plan.live().find_map(|(_, node)| match node { + PhysicalNode::AntiJoin { strategy, .. } => Some(*strategy), + _ => None, + }); + let report = serde_json::to_value(&executed.report).unwrap(); + let (all, inner) = rows_through(&executed.plan.plan, &report); + println!( + "{label:>10}: strategy {strategy:?}, {} result rows, {all} rows through operators, \ + {inner} through the block's inner tree, {elapsed:.2?}", + executed.result.num_rows(), + ); + match label { + "key_set" => assert_eq!(strategy, Some(AntiJoinStrategy::KeySet)), + "correlated" => assert_eq!(strategy, Some(AntiJoinStrategy::Correlated)), + _ => assert_eq!(strategy, None), + } + answers.push((executed.result.to_rust_json().unwrap(), inner)); + } + let [_, (key_set, key_set_rows), (correlated, correlated_rows)] = + <[_; 3]>::try_from(answers).unwrap(); + assert_eq!(key_set, correlated, "both strategies keep the same rows"); + assert!( + key_set_rows < correlated_rows, + "the key set's inner tree emits fewer rows: {key_set_rows} against {correlated_rows}" + ); +} diff --git a/crates/omnigraph/tests/integration/engine_v2.rs b/crates/omnigraph/tests/integration/engine_v2.rs index ac74d31b1..d9e09bafc 100644 --- a/crates/omnigraph/tests/integration/engine_v2.rs +++ b/crates/omnigraph/tests/integration/engine_v2.rs @@ -947,3 +947,178 @@ query total($q: Vector(4)) { ); } } + +const KEY_SET_SCHEMA: &str = r#" +node Channel { slug: String @key } +node Source { slug: String @key } +node Touchpoint { slug: String @key action: String } +node ExternalID { slug: String @key } +edge ChannelWatchesSource: Channel -> Source +edge TouchpointOnSource: Touchpoint -> Source +edge TouchpointByIdentity: Touchpoint -> ExternalID +"#; + +const KEY_SET_QUERY: &str = r#" +query lookalikes($channel: String, $exclude: String) { + match { + $c: Channel { slug: $channel } + $s: Source + $c channelWatchesSource $s + $t: Touchpoint { action: "star" } + $t touchpointOnSource $s + $x: ExternalID + $t touchpointByIdentity $x + not { + $s0: Source { slug: $exclude } + $t0: Touchpoint + $t0 touchpointOnSource $s0 + $t0 touchpointByIdentity $x + } + } + return { $x.slug, count($t) as repos } +} +"#; + +/// A channel watching `watched` sources, `identities` identities that star +/// each of them and touch `elsewhere` other touchpoints, and the first +/// `excluded` identities starring the excluded source once. +fn key_set_seed(watched: usize, identities: usize, elsewhere: usize, excluded: usize) -> String { + let mut lines = vec![ + r#"{"type":"Channel","data":{"slug":"c"}}"#.to_string(), + r#"{"type":"Source","data":{"slug":"excluded"}}"#.to_string(), + r#"{"type":"Source","data":{"slug":"elsewhere"}}"#.to_string(), + ]; + for source in 0..watched { + lines.push(format!( + r#"{{"type":"Source","data":{{"slug":"s{source}"}}}}"# + )); + lines.push(format!( + r#"{{"edge":"ChannelWatchesSource","from":"c","to":"s{source}"}}"# + )); + } + let mut touch = |slug: String, source: &str, identity: usize| { + lines.push(format!( + r#"{{"type":"Touchpoint","data":{{"slug":"{slug}","action":"star"}}}}"# + )); + lines.push(format!( + r#"{{"edge":"TouchpointOnSource","from":"{slug}","to":"{source}"}}"# + )); + lines.push(format!( + r#"{{"edge":"TouchpointByIdentity","from":"{slug}","to":"x{identity}"}}"# + )); + }; + for identity in 0..identities { + for source in 0..watched { + touch( + format!("t{identity}-{source}"), + &format!("s{source}"), + identity, + ); + } + for other in 0..elsewhere { + touch(format!("e{identity}-{other}"), "elsewhere", identity); + } + } + for identity in 0..excluded { + touch(format!("x{identity}"), "excluded", identity); + } + for identity in 0..identities { + lines.push(format!( + r#"{{"type":"ExternalID","data":{{"slug":"x{identity}"}}}}"# + )); + } + lines.join("\n") +} + +/// A correlated block bound by a key equality on one of its own bindings runs +/// once from that binding: its inner rows follow the excluded source's three +/// touchpoints, not the 400 outer rows times each identity's 31 touchpoints +/// (the correlated tree expands 12,030 rows here). The anti-join's input +/// still carries every outer row, and the kept rows are the identities the +/// excluded source never saw. +#[tokio::test] +async fn bounded_correlated_block_runs_once_from_its_key_set() { + use omnigraph_planner::{AntiJoinStrategy, PhysicalNode}; + + const WATCHED: usize = 10; + const IDENTITIES: usize = 40; + const ELSEWHERE: usize = 20; + const EXCLUDED: usize = 3; + let dir = tempfile::tempdir().unwrap(); + let db = session( + Omnigraph::init(dir.path().to_str().unwrap(), KEY_SET_SCHEMA) + .await + .unwrap(), + ); + db.load_jsonl( + &key_set_seed(WATCHED, IDENTITIES, ELSEWHERE, EXCLUDED), + LoadMode::Overwrite, + ) + .await + .unwrap(); + let executed = db + .query_inspected( + "main", + KEY_SET_QUERY, + "lookalikes", + ¶ms(&[("channel", "c"), ("exclude", "excluded")]), + ) + .await + .unwrap(); + + let serde_json::Value::Array(mut kept) = executed.result.to_rust_json().unwrap() else { + panic!("rows"); + }; + kept.sort_by_key(|row| row["x.slug"].as_str().unwrap().to_string()); + let expected: Vec = { + let mut slugs: Vec = (EXCLUDED..IDENTITIES).map(|i| format!("x{i}")).collect(); + slugs.sort(); + slugs + .into_iter() + .map(|slug| serde_json::json!({"x.slug": slug, "repos": WATCHED})) + .collect() + }; + assert_eq!(kept, expected); + + let plan = &executed.plan.plan; + let (outer, inner) = plan + .live() + .find_map(|(_, node)| match node { + PhysicalNode::AntiJoin { + input, + inner, + strategy, + .. + } => { + assert_eq!(*strategy, AntiJoinStrategy::KeySet, "{node:?}"); + Some((*input, *inner)) + } + _ => None, + }) + .expect("the plan holds the block's AntiJoin"); + let report = serde_json::to_value(&executed.report).unwrap(); + let rows_of_node = |id| { + report["rows"] + .as_array() + .unwrap() + .iter() + .find(|row| row["id"] == serde_json::json!(id)) + .and_then(|row| row["attempts"].as_array()?.last()?["actual_rows"].as_u64()) + .unwrap_or_else(|| panic!("node {id} has no report row: {report}")) + }; + assert_eq!(rows_of_node(outer), (WATCHED * IDENTITIES) as u64); + let mut pending = vec![inner]; + let mut inner_expands = Vec::new(); + while let Some(id) = pending.pop() { + let node = plan.node(id).unwrap(); + assert!( + !matches!(node, PhysicalNode::OuterReference { .. }), + "a key-set inner tree reads no outer rows" + ); + if matches!(node, PhysicalNode::Expand { .. }) { + inner_expands.push(rows_of_node(id)); + } + pending.extend(node.inputs()); + } + assert_eq!(inner_expands, vec![EXCLUDED as u64; 2], "{report}"); +} diff --git a/crates/omnigraph/tests/integration/main.rs b/crates/omnigraph/tests/integration/main.rs index 241cc0e10..8779dbc34 100644 --- a/crates/omnigraph/tests/integration/main.rs +++ b/crates/omnigraph/tests/integration/main.rs @@ -6,6 +6,7 @@ #![recursion_limit = "512"] mod aggregation; +mod anti_join_cost; mod branch_control_cost; mod branching; mod changes; diff --git a/docs/dev/execution.md b/docs/dev/execution.md index bbf233093..12ece0860 100644 --- a/docs/dev/execution.md +++ b/docs/dev/execution.md @@ -141,7 +141,8 @@ wildcard expansions. A selected dataset absent from the captured snapshot fails when opened; empty input need not open it. Bound plans use a versioned envelope and require regeneration when the stored version is unsupported. Envelope version 7 requires aggregate expressions to carry their `distinct` -flag. Plans saved with earlier binaries must be regenerated before replay. +flag, and version 8 every `AntiJoin` its `strategy`. Plans saved with earlier +binaries must be regenerated before replay. ### Expand path selection @@ -200,6 +201,37 @@ scoped requests with per-request freshness checks. Both execution paths have identical query semantics, so the mode override is an operational escape hatch and test seam only. +### Correlated block strategy + +A correlated block's logical inner tree starts from the outer rows: its leaf is +an `OuterReference` and its first `Expand` leaves the correlation binding. That +`correlated` tree runs once over the whole outer batch, so its work is the +outer rows times the correlation binding's fanout, and every destination it +reaches is looked up once per slice of those rows. When the block's pattern +touches the outer rows only through the correlation binding's identity, the +planner also builds a `key_set` tree (`omnigraph_planner` `key_set.rs`): the +same pattern rooted at the block's most bounded binding (a key or unique +equality or a list membership on its own scan, below the table's row count), +each `Expand` written toward that root reversed, ending on the correlation +binding's identity column with no scan of it. An unbound `Expand` emits one row +per distinct endpoint pair and a bound one one row per edge row, in either +direction, so every key meets the same matches. Pass `anti_join_strategy` +prices both from the summed row estimates of each chain (`chain_rows`, the +correlated one from the outer input's estimate), records +`anti_join_rows($x)`, and lowers the `key_set` tree only when it produces fewer +rows; an unknown estimate keeps `correlated`. The logical plan keeps the +correlated tree as the block's meaning; the strategy is the physical +`AntiJoin`'s `strategy` and the GQT line `anti join $x: key_set`. + +A block stays `correlated` when its inner tree holds anything but `Expand`, +dependent `TableScan` and `Filter` nodes (a nested block, a second +component), when an `Expand` leaves a binding the pattern has not reached or +closes a cycle, when any filter or the block aggregate's argument reads an +outer binding (the correlation binding included: its identity is the key, +never a read column), when no binding is bounded, and under a +`traversal_work_limit`. The bulk CSR degree check needs a filter-free +destination, so its shapes never have a bounded binding and stay correlated. + ## Filters and pushdown A read query runs on engine v2, from a plan. The session setting `engine` @@ -345,7 +377,7 @@ by `engine/scan.rs`, `engine/graph.rs`, `engine/expr.rs` and | `ContainsJoin` | `ContainsJoinExec`: the planner's join for a `Filter` over a `CrossJoin` holding a `$r.x contains $l.y` conjunct whose `$r` is the unranked table `Scan` on the right and whose two properties are text (`optimizer.rs` `filtered_cross_join`, reported as the `join_algorithm` pass; a search order that later ranks that scan takes the marker off again and turns the join back into the `CrossJoin` of its conjuncts, `Lowering::unmark`, since a ranked scan runs under a search mode and no runtime filter); the filter's other conjuncts are the node's `residual`, and the right scan's `ScanSpec` carries `runtime_filter` (`column`, `needle`, `kind` `text_contains_any`), which the engine's lowering turns into the `RuntimeFilterSlot` both operators share; the lowering refuses a plan (a replayed or edited one) whose marked scan's column or needle differs from its `ContainsJoin`'s, or whose marked plain table scan no `ContainsJoin` fills (a marker on any other scan is not read, and that scan reads unfiltered). The left input is collected under the query pool (`contains join left`); the join admits its `PairBuffer`, then fills that slot (a fill first clears any earlier execution's filter) with one Aho-Corasick automaton over the distinct non-empty `y` values, charged once as `runtime filter needles` and shared with the join (the scan is left unfiltered when a value is empty, since that value is in every text, though the join still pairs through the automaton; none, and the join tests every pair, when the pool refuses it; no right execution when every value is null); both operators print `runtime_filter=$r.x contains any($l.y)`, and `ScanExec` sieves each Lance batch through it as the `Scan` row describes (`runtime_filter_rows_read`, `runtime_filter_rows_dropped`, `runtime_filter_inert`; the join's `runtime_filter_needles`). At its first right batch the join builds its needle rows over that same automaton, which the streaming scan may still be sieving through (`contains_join=aho_corasick`; only the row tables are charged, as `contains join needles`): each non-null text pairs only with the left rows of the needles it holds, the pass over a text stopping once every needle is found and checking for cancellation every 4,096 occurrences, and each empty needle's row pairs with each non-null text; when the pool refuses the needle rows it pairs every row. Every pair goes through the same `PairBuffer` as `CrossJoinExec`, which tests only the `residual` conjuncts on the needle-rows path, whose every pair holds the `contains` conjunct by construction, and the `contains` conjunct with them on the every-row path; with the right scan streaming, the join's memory is the left side plus the shared automaton, the needle rows, one right batch and one output batch, beside the producer queue's two batches in flight. `contains_join_matcher` is 1 when the run paired through the needle rows, 0 when the pool refused them, and absent when no right row reached the join; `contains_join_pairs` counts the pairs found for the non-empty needles before the residual conjuncts, beside a matcher of 1 | omnigraph | | `Filter` | `FilterExec`: every `IRFilter` of the node as one conjunction over the wide batch (`evaluate_filter`), so a filter over two bindings reads two columns | omnigraph | | `Expand` | `ExpandExec`: an unbudgeted named unbound single hop uses one vectorized walk per input batch (`operators/single_hop.rs`); unbudgeted named multi-hop drains its input into the BFS breaker `execute_expand`. Budgeted expansions, including selected multi-hop, consume fixed source windows through `SourceWindows`. Bound edges use the bounded pair producer, spillable pair ordering and incremental hydration within each input batch or budgeted source window. Output chunks contain at most 256 rows (`expand_pairs` counts pairs handed on); dropping the stream at a downstream `Limit` cancels the producer | omnigraph | -| `AntiJoin` | `AntiJoinMaskExec` over the outer plan and the lowered inner plan: the bulk CSR degree mask when the predicate counts rows and the inner is one single-hop, filter-free, unbound expand over the `OuterReference`; else the outer rows are tagged, the inner plan runs over `OuterReferenceExec` under the same `TaskContext`, and `SubqueryAggregate` folds the tagged inner rows per outer row and applies the predicate | omnigraph | +| `AntiJoin` | `AntiJoinMaskExec` over the outer plan and the lowered inner plan. `correlated`: the bulk CSR degree mask when the predicate counts rows and the inner is one single-hop, filter-free, unbound expand over the `OuterReference`; else the outer rows are tagged, the inner plan runs over `OuterReferenceExec` under the same `TaskContext`, and `SubqueryAggregate` folds the tagged inner rows per outer row and applies the predicate. `key_set`: the outer rows are drained, each distinct `.` gets one `SubqueryAggregate` group, the inner plan (no outer leaf) streams and each batch folds into its key's group (an inner key no outer row holds into a discarded group) without being held, and each outer row takes its key's verdict, a null key that of no matches | omnigraph | | `Projection` | `ProjectionExec` over `GqProjectionExpr` per return expression, output charged as `projection output`; when a `Sort` consumes it, every column the sort reads and every declared tie-break metadata column follow the return columns under the hidden prefix `~`, which the sort drops; under a `HydrateColumns` it skips the deferred return items and carries each deferred binding's `_rowaddr` as `^binding` | omnigraph | | `HydrateColumns` | `HydrateExec` (`operators/hydrate.rs`): per input batch, in chunks planned from the costliest output row seen (its decoded row and its copy; four rows first), the distinct `^binding` addresses of each binding, sorted and grouped by fragment, read from its pinned table through `FileFragment::open` and `FragmentReader::take` with the deferred columns' projection and one row per decode task. Each decoded row is copied at its own size through `WorkMemory::take_once` (charged as `hydrate fetch`; Lance's row can slice a buffer of the whole read) before the fetch waits for the next, at most `ROWS_IN_FLIGHT` (8) rows decoding at once and `FRAGMENTS_AHEAD` (4) fragments' reads scheduled ahead. A fragment whose data files are in the table's own base is read in one coalesced request through an engine `ScanScheduler`, whose I/O buffer, a quarter of the bound, is reserved for the operator's life as `hydrate read-ahead`; a fragment holding a file Lance reaches through a `base_id`, for which Lance 11 opens its own scheduler, is read one row per request and counted as `single_row_reads` (only a table fork created before v11 holds one; a graph branch's writes stay in the table's own base). An address the table does not hold is an integrity error. The rows are copied to every output row that names them through `WorkMemory::interleave_once`, which admits each copy (one per output row, so a row a join repeats is copied per repetition; a null address costs a null row) as `hydrate output` before Arrow builds it. The bound is `hydrate_chunk_bytes(pool)` (an eighth of the query pool, the node's declared `retained_limit`): the read-ahead reservation, then the decoded rows and their copies within the rest. A fetch stops once its decoded rows pass half of that rest, and the window is planned again from the widest row decoded; a chunk whose copies would exceed it keeps a prefix halved until they fit, the rest starting the next chunk. So the operator holds at most the bound and the row that crosses it, with at most `ROWS_IN_FLIGHT` rows decoded and not yet charged, and Lance's read-ahead past its buffer is at most the request at the head of its queue, which Lance may coalesce from many rows' ranges (up to its maximum I/O size, 16 MiB by default), so the read-ahead share is soft. A chunk is charged as `hydrate chunk`, keeps only its output once built, and is sent as it is built; `hydrated_rows` counts the rows and the `peak_chunk_bytes` gauge records the most one chunk held, every decoded row included | omnigraph | | `Sort` | `SortExec`: with a `fetch`, a streaming top-k (every input batch merged into the retained best `fetch` rows and released, so the pool holds one batch and `fetch` rows at a time); without one, the whole input held under the query pool (no spill) and sorted once. Keys are `lexsort_to_indices` over the node's `order_by`, `nulls_first = !descending`, then the metadata columns of the node's declared `tiebreak`, ascending; the `~` columns are dropped on the way out. The planner (`optimizer::sort_tiebreak`) declares identity columns in binding-name order, with selected bound-edge type before its ID. Each metadata key already explicitly ordered is omitted independently; group rows and a `return` whose every expression is an order key need no hidden ties; `projection_pushdown` reads an id only for a declared tie-break, a traversal, a dependent scan, an anti-join, a ranked scan or an expression naming `@id`. The planner writes a search order's score key (`$d._score desc`, `$d._distance asc`) first and the query's plain keys after it, with the limit as `fetch`; a fusion plans no `Sort`, and an aggregate under a search order plans none | omnigraph | diff --git a/docs/dev/testing.md b/docs/dev/testing.md index b09d6edcc..d02b28198 100644 --- a/docs/dev/testing.md +++ b/docs/dev/testing.md @@ -70,7 +70,7 @@ The engine integration suite is grouped by behavior, not implementation module: | Export and lineage | `integration/export.rs`, `integration/lineage_projection.rs` | | Legacy-vintage graphs (`id`/`src`/`dst` spellings, born at the current stamp) | `integration/legacy_columns.rs` — load, query, export round trip, evolution; needs `--features failpoints` | | System-column upgrade (RFC 0040 step 3: respelling in place on a supported standalone graph; vintage is independent of the storage stamp) | `system_column_upgrade.rs`: check and execute, preflight refusals, every window before the manifest commit leaving no residue, a complete contract and table state after a post-commit failure, same-handle retry, the control-object cost; needs `--features failpoints` | -| Cost and benchmark contracts | `integration/write_cost.rs`, `integration/write_cost_s3.rs`, `integration/warm_read_cost.rs`, `integration/branch_control_cost.rs`, `integration/merge_cost.rs`, `integration/changes_cost.rs`, the checkpoint/head lookup instruments, and the GQT benchmark adapter tests in `omnigraph-bench` | +| Cost and benchmark contracts | `integration/write_cost.rs`, `integration/write_cost_s3.rs`, `integration/warm_read_cost.rs`, `integration/branch_control_cost.rs`, `integration/merge_cost.rs`, `integration/changes_cost.rs`, the checkpoint/head lookup instruments, the correlated-block strategy instrument `integration/anti_join_cost.rs`, and the GQT benchmark adapter tests in `omnigraph-bench` | Use `tests/helpers/mod.rs` for the standard graph, snapshots, row reads, Blob selectors, and bounded Blob collection. Recovery helpers belong in `tests/helpers/recovery.rs`; object-store counters belong in `tests/helpers/cost.rs`; graphs whose rows trip the ordered-scan sorter cap belong in `tests/helpers/wide_rows.rs`. diff --git a/docs/user/queries/explain.md b/docs/user/queries/explain.md index 8070fb188..ff022a7d4 100644 --- a/docs/user/queries/explain.md +++ b/docs/user/queries/explain.md @@ -88,7 +88,17 @@ physical `RankFuse` row has two inputs, one subtree per arm, each with its own ranked `Scan`; it carries `arms` (binding and kind per arm), `k`, `limit` (the query's, which sizes the nearest arm's candidate window; the fused rows are -cut by the `Limit` above the `Sort`) and `row_tiebreak`. These downstream +cut by the `Limit` above the `Sort`) and `row_tiebreak`. A physical `AntiJoin` +row (a correlated block: `not`, `exists`, `count` and the column aggregates) +carries `strategy`: `correlated` when its second input starts from the outer +rows through an `OuterReference`, `key_set` when the block's pattern runs once +from its own most bounded binding (a key or unique equality or a list +membership) and ends on the correlation binding, whose identity keeps or drops +each outer row. The planner chooses `key_set` only for a block that reads +nothing of the outer rows but that identity, and only when its row estimate is +lower; it then records the pass `anti_join_strategy` and the statistic +`anti_join_rows($x)` with both estimates. Either strategy returns the same +rows. These downstream row keys follow the arm's score and fused node ID; selected edges use `$e.@type` before `$e.@id`. Fusion still scores the ranked node by its ID and appends the fused score as `._rrf`. A @@ -155,7 +165,7 @@ row per operator with `depth` from its nesting and `detail` the operator's own text (a filter's predicate, a sort's keys, a scan's columns; `AntiJoinMaskExec` serves every correlated block, `not`, `exists`, `count` and the column aggregates, and prints its -predicate). It is the tree of the query's first pass; a +predicate, then `key_set` under that strategy). It is the tree of the query's first pass; a search that widens its scan after a short answer lowers it again. An `explain` runs nothing, so when the lowering would need the embedding client (a `nearest()` whose query is a string), the tree is left out and one `plan` @@ -184,7 +194,7 @@ on the `ExpandExec` of an `Expand`, the mode the traversal ended on. Explain its profile is returned beside the rows by the run that produced them (`Session::query_inspected`, the v2 inspection door, through `Executed::profile`). The row schema is `explain_version` 4. Saved physical -plans use a `bound_plan_version: 6` envelope whose `body` contains `plan` and +plans use a `bound_plan_version: 8` envelope whose `body` contains `plan` and `values`. Older unversioned plans and unsupported versions are refused with a regeneration instruction, including plans without traversal nodes. @@ -199,7 +209,8 @@ such as `["total: F64?", "person: Person"]`. Names match executed result columns `?` records declared nullability, so `count` is `I64?` even when execution returns no nulls. A node type names its complete projected object, excluding Blob and Vector properties. Saved version 1 through 5 plans must be regenerated -for the current typed expression declarations. +for the current typed expression declarations, and version 6 and 7 plans for +the distinct-count flag and the `AntiJoin` strategy. Expression-bearing plan fields have additive `typed_*` counterparts. Each expression tree records `op`, the existing GQ text in `gq`, its stored `type`, diff --git a/docs/user/queries/index.md b/docs/user/queries/index.md index b6ba5a3e2..9a337af22 100644 --- a/docs/user/queries/index.md +++ b/docs/user/queries/index.md @@ -128,11 +128,11 @@ skips null values. The block narrows the rows before `order` and `limit`, so a paged listing filtered by a relationship count is exact. A binding the block's traversal -connects to the outer row is reached through that traversal, never scanned as -a whole table, however many rows the outer pattern has; a binding correlated -only through a filter, or read by a text search, is scanned. A -single-hop, filter-free `count { ... }` over a directed, unbound edge is -answered from the graph index's degree. A bare aggregate in `match`, +connects to the outer row is reached through it, or the block runs once from a +node it binds by key if that is cheaper (`strategy` in [explain](explain.md)); +a binding correlated only through a filter, or read by a text search, is +scanned. A single-hop, filter-free `count { ... }` over a directed, unbound +edge is answered from the graph index's degree. A bare aggregate in `match`, `count($d) > 2` without a block, is refused: it names no row to group by. ``` From 0d5f2dffce05d3ce6c06cd35ce4564831b8a0076 Mon Sep 17 00:00:00 2001 From: Ragnor Comerford Date: Wed, 14 Oct 2026 11:55:07 -0700 Subject: [PATCH 2/2] perf(planner): refuse a block strategy its inner tree lacks; pin the key-set root and its equivalence Plan acceptance refuses an AntiJoin whose inner tree does not have the shape its strategy declares (check `block strategy`). The key-set root choice and its tie rule have planner tests; a differential property test compares every key-set block shape with its forced-correlated twin. The arena swap refuses a key-set tree holding a correlated block, and the execution guide notes that a hub root is under-priced. --- crates/omnigraph-planner/src/optimizer.rs | 13 +- .../src/validate/invariants.rs | 80 ++++++++++-- .../omnigraph-planner/src/validate/tests.rs | 67 ++++++++++ .../tests/query_plan/cost.rs | 89 +++++++++++++ .../tests/integration/proptest_equivalence.rs | 122 ++++++++++++++++++ docs/dev/execution.md | 15 ++- docs/dev/testing.md | 2 +- 7 files changed, 368 insertions(+), 20 deletions(-) diff --git a/crates/omnigraph-planner/src/optimizer.rs b/crates/omnigraph-planner/src/optimizer.rs index 2995913a3..e40d5eeff 100644 --- a/crates/omnigraph-planner/src/optimizer.rs +++ b/crates/omnigraph-planner/src/optimizer.rs @@ -2469,7 +2469,9 @@ struct Lowering<'a> { logical: &'a LogicalPlan, source: &'a dyn PlanSource, bounds: &'a Bounds, - /// The key-set tree of each qualifying `AntiJoin`, by its logical id. + /// The key-set tree of each qualifying `AntiJoin`, by its id in the + /// plan's arena. A tree's own ids overlap these, so a tree must hold no + /// `AntiJoin` to look up while it is lowered (checked at the swap). key_sets: &'a HashMap, physical: PhysicalPlan, late_materialization: bool, @@ -3047,6 +3049,15 @@ impl Lowering<'_> { }; let inner_lowered = match (strategy, key_set) { (AntiJoinStrategy::KeySet, Some(tree)) => { + if tree + .plan + .live() + .any(|(_, node)| matches!(node, LogicalNode::AntiJoin { .. })) + { + return Err(PlanError::Internal(format!( + "the key-set tree of block `${outer_var}` holds a correlated block, whose id would name another arena's tree" + ))); + } let logical = std::mem::replace(&mut self.logical, &tree.plan); let lowered = self.lower(tree.root); self.logical = logical; diff --git a/crates/omnigraph-planner/src/validate/invariants.rs b/crates/omnigraph-planner/src/validate/invariants.rs index a63026883..e91990dc8 100644 --- a/crates/omnigraph-planner/src/validate/invariants.rs +++ b/crates/omnigraph-planner/src/validate/invariants.rs @@ -24,6 +24,7 @@ use omnigraph_compiler::types::{ExprType, ScalarType}; use super::budget::Budget; use super::requirements::{Requirements, Retrieval, Search, constant_true}; use super::{AcceptInput, ValidationError}; +use crate::cost::AntiJoinStrategy; use crate::logical::{ColumnRef, EDGE_TYPE_MEMBER, IDENTITY_MEMBER}; use crate::optimizer::{RRF_NEAREST_ARM_K, contains_aggregate, derived_order}; use crate::physical::{ @@ -55,6 +56,51 @@ fn pipeline( Ok(nodes) } +/// A block's inner tree has the shape its strategy declares +/// (`PhysicalNode::AntiJoin`): under `correlated` its pipeline reaches the +/// `OuterReference` to the block's binding; under `key_set` it reaches no +/// `OuterReference` and an `Expand` introduces the binding, whose identity +/// keys the outer rows. Refused here, a mismatched saved or edited plan never +/// reaches the operator, where it would fail on a missing outer batch or +/// correlation column. +fn check_block_strategy( + plan: &PhysicalPlan, + id: NodeId, + outer_var: &str, + inner: NodeId, + strategy: AntiJoinStrategy, + budget: &mut Budget, +) -> Result<(), ValidationError> { + let nodes = pipeline(plan, inner, budget)?; + let reaches = |want: &dyn Fn(&PhysicalNode) -> bool| { + nodes.iter().any(|node| plan.node(*node).is_some_and(want)) + }; + let holds = match strategy { + AntiJoinStrategy::Correlated => reaches( + &|node| matches!(node, PhysicalNode::OuterReference { outer_var: leaf } if leaf == outer_var), + ), + AntiJoinStrategy::KeySet => { + !reaches(&|node| matches!(node, PhysicalNode::OuterReference { .. })) + && reaches( + &|node| matches!(node, PhysicalNode::Expand { dst, .. } if dst == outer_var), + ) + } + }; + if holds { + return Ok(()); + } + Err(ValidationError::violated( + "block strategy", + format!( + "node {id} declares the `{}` strategy for its block over `${outer_var}`, which its inner tree does not have", + match strategy { + AntiJoinStrategy::Correlated => "correlated", + AntiJoinStrategy::KeySet => "key_set", + } + ), + )) +} + /// One pipeline a query's eligibility constrains: `own`, the nodes that /// produce its rows, and `after`, the nodes those rows pass through above a /// fusion (empty without one). @@ -194,7 +240,7 @@ impl Requirements { let reached = pipeline.reached(); self.check_bindings(plan, &reached)?; self.check_eligibility(plan, &pipeline.own, &reached, input, budget)?; - self.check_blocks(plan, &pipeline.own)?; + self.check_blocks(plan, &pipeline.own, budget)?; if let Some(id) = pipeline .after .iter() @@ -404,18 +450,26 @@ impl Requirements { /// The correlated blocks of each pipeline the query's rows come from are /// exactly the written ones, each over its outer binding with the /// predicate the query states. - fn check_blocks(&self, plan: &PhysicalPlan, nodes: &[NodeId]) -> Result<(), ValidationError> { - let mut planned: Vec<(&String, &omnigraph_compiler::ir::SubqueryPredicate)> = nodes - .iter() - .filter_map(|id| match plan.node(*id) { - Some(PhysicalNode::AntiJoin { - outer_var, - predicate, - .. - }) => Some((outer_var, predicate)), - _ => None, - }) - .collect(); + fn check_blocks( + &self, + plan: &PhysicalPlan, + nodes: &[NodeId], + budget: &mut Budget, + ) -> Result<(), ValidationError> { + let mut planned: Vec<(&String, &omnigraph_compiler::ir::SubqueryPredicate)> = Vec::new(); + for id in nodes { + if let Some(PhysicalNode::AntiJoin { + outer_var, + predicate, + inner, + strategy, + .. + }) = plan.node(*id) + { + check_block_strategy(plan, *id, outer_var, *inner, *strategy, budget)?; + planned.push((outer_var, predicate)); + } + } if planned.len() != self.blocks.len() { return Err(ValidationError::violated( "predicate retention", diff --git a/crates/omnigraph-planner/src/validate/tests.rs b/crates/omnigraph-planner/src/validate/tests.rs index aa11d88dd..c2e0e6a6a 100644 --- a/crates/omnigraph-planner/src/validate/tests.rs +++ b/crates/omnigraph-planner/src/validate/tests.rs @@ -334,6 +334,73 @@ fn a_dropped_conjunct_fails_predicate_retention() { assert_eq!(check, "predicate retention", "{detail}"); } +/// The strategy of every block of `plan` set to `strategy`. +fn relabel_blocks(plan: &mut PhysicalPlan, strategy: crate::AntiJoinStrategy) { + for id in node_ids(plan, |node| matches!(node, PhysicalNode::AntiJoin { .. })) { + if let Some(PhysicalNode::AntiJoin { + strategy: declared, .. + }) = plan.node_mut(id) + { + *declared = strategy; + } + } +} + +#[test] +fn a_strategy_its_inner_tree_lacks_fails_block_strategy() { + use crate::AntiJoinStrategy::{Correlated, KeySet}; + use crate::source::PlanSource; + use omnigraph_compiler::types::Direction; + let mut fixture = Fixture::new( + "query q($s: String) { match { $d: Doc not { $d cites $e $e: Doc { slug: $s } } } return { $d.slug } }", + &[("s", Literal::String("x".to_string()))], + ); + let strategies = |plan: &PhysicalPlan| -> Vec { + plan.live() + .filter_map(|(_, node)| match node { + PhysicalNode::AntiJoin { strategy, .. } => Some(*strategy), + _ => None, + }) + .collect() + }; + // No row counts: the block is correlated, its inner tree over the outer rows. + let correlated = fixture.plan(); + assert_eq!(strategies(&correlated), [Correlated]); + fixture + .accept(correlated.clone()) + .expect("the correlated plan is accepted"); + let mut relabelled = correlated; + relabel_blocks(&mut relabelled, KeySet); + let (check, detail) = fixture.refused(relabelled); + assert_eq!(check, "block strategy", "{detail}"); + + // With row counts and edge statistics the bound `$e` makes the key set cheaper. + let mut doc = fixture.source.node_type("Doc").unwrap(); + doc.row_count = Some(100); + let statistics = crate::ExpandStatistics { + edge_count: 1_000, + src_node_count: 100, + dst_node_count: 100, + max_frontier_cap: 1024, + max_hops_cap: 6, + }; + fixture.source = fixture + .source + .clone() + .with_node_type("Doc", doc) + .with_expand_statistics("Cites", Direction::Out, statistics) + .with_expand_statistics("Cites", Direction::In, statistics); + let key_set = fixture.plan(); + assert_eq!(strategies(&key_set), [KeySet]); + fixture + .accept(key_set.clone()) + .expect("the key-set plan is accepted"); + let mut relabelled = key_set; + relabel_blocks(&mut relabelled, Correlated); + let (check, detail) = fixture.refused(relabelled); + assert_eq!(check, "block strategy", "{detail}"); +} + fn value_type( scalar: omnigraph_compiler::types::ScalarType, ) -> omnigraph_compiler::types::ExprType { diff --git a/crates/omnigraph-planner/tests/query_plan/cost.rs b/crates/omnigraph-planner/tests/query_plan/cost.rs index 6c6288aba..c94c05d27 100644 --- a/crates/omnigraph-planner/tests/query_plan/cost.rs +++ b/crates/omnigraph-planner/tests/query_plan/cost.rs @@ -1295,3 +1295,92 @@ fn unbounded_or_outer_reading_blocks_stay_correlated() { ); } } + +/// `$x.prop` in a list of `values`, a binding bounded to that many rows. +fn member_of(variable: &str, property: &str, values: &[&str]) -> IRExpr { + IRExpr::comparison( + IRExpr::Literal( + Literal::List( + values + .iter() + .map(|value| Literal::String(value.to_string())) + .collect(), + ), + ExprType::Value { + scalar: ScalarType::String, + list: true, + nullable: false, + }, + ), + CompOp::Contains, + prop(variable, property), + ) +} + +/// The binding of the table scan a key-set tree starts from. +fn key_set_root(plan: &PhysicalPlan) -> String { + let (strategy, inner) = anti_join_of(plan); + assert_eq!(strategy, omnigraph_planner::AntiJoinStrategy::KeySet); + inner + .iter() + .find_map(|node| match node { + PhysicalNode::Scan { + source: omnigraph_planner::ScanInput::Table, + spec, + .. + } => spec.binding.clone(), + _ => None, + }) + .expect("a key-set tree starts from a table scan") +} + +/// The key set starts from the binding with the fewest estimated rows: `$k` +/// under a key equality (one row) over `$m` in a three-value list; between +/// two equal bounds, from the binding the correlated tree reaches first. +#[test] +fn key_set_starts_from_the_most_bounded_binding_and_the_first_on_a_tie() { + let block = |m_filters: Vec, k_filters: Vec| { + ir( + vec![ + scan("a"), + IROp::AntiJoin { + outer_var: "a".to_string(), + inner: vec![expand("a", "m", m_filters), expand("m", "k", k_filters)], + predicate: omnigraph_compiler::ir::SubqueryPredicate::not_exists(), + }, + ], + vec![prop("a", "slug")], + vec![], + ) + }; + let source = anti_join_source(10); + for (what, op, root) in [ + ( + "a key equality over a list", + block( + vec![member_of("m", "slug", &["m1", "m2", "m3"])], + vec![equals("k", "slug", "k1")], + ), + "k", + ), + ( + "a list under a key equality", + block( + vec![equals("m", "slug", "m1")], + vec![member_of("k", "slug", &["k1", "k2", "k3"])], + ), + "m", + ), + ( + "two key equalities", + block( + vec![equals("m", "slug", "m1")], + vec![equals("k", "slug", "k1")], + ), + "m", + ), + ] { + let (plan, _) = physical(&op, &source); + assert_eq!(key_set_root(&plan), root, "{what}"); + } +} diff --git a/crates/omnigraph/tests/integration/proptest_equivalence.rs b/crates/omnigraph/tests/integration/proptest_equivalence.rs index 919dcd101..76b10e3db 100644 --- a/crates/omnigraph/tests/integration/proptest_equivalence.rs +++ b/crates/omnigraph/tests/integration/proptest_equivalence.rs @@ -288,3 +288,125 @@ fn prop_antijoin_partitions_persons() { }) .unwrap(); } + +/// Correlated blocks bound by a key on one of their own bindings, as +/// `(outer pattern, block)`. Each runs as written and again with `FORCE` set +/// to a conjunct that reads the outer row (`$p.name = $p.name`, always true), +/// which keeps the twin `correlated`; written, the planner may run the block as +/// a key set from the bound binding. The last shape repeats each `$p` once per +/// outer `knows` edge. +const KEY_SET_SHAPES: &[(&str, &str)] = &[ + ( + "$p: Person", + "not { $p knows{1,3} $x $x: Person { name: $name } FORCE }", + ), + ( + "$p: Person", + "not { $x knows{1,2} $p $x: Person { name: $name } FORCE }", + ), + ( + "$p: Person", + "count { $p {1,2} $x $x: Person { name: $name } FORCE } = 1", + ), + ( + "$p: Person", + "not { $p knows{2,2} $x $x: Person { name: $name } FORCE }", + ), + ( + "$p: Person", + "count { $p $k:knows $x $x: Person { name: $name } FORCE } = 1", + ), + ( + "$p: Person", + "not { $p knows $m $p worksAt $c $c: Company { name: $name } FORCE }", + ), + ( + "$p: Person $p knows $q", + "not { $p worksAt $c $c: Company { name: $name } FORCE }", + ), +]; + +/// The run of `block` over `outer` with `force` in its place: its first +/// column as a sorted multiset, and the block's strategy. +async fn block_run( + db: &Session, + outer: &str, + block: &str, + force: &str, + params: &ParamMap, +) -> (Vec, omnigraph_planner::AntiJoinStrategy) { + let query = format!( + "query q($name: String) {{ match {{ {outer} {} }} return {{ $p.name }} }}", + block.replace("FORCE", force) + ); + let executed = db + .query_inspected(ReadTarget::branch("main"), &query, "q", params) + .await + .unwrap(); + let strategy = executed + .plan + .plan + .live() + .find_map(|(_, node)| match node { + omnigraph_planner::PhysicalNode::AntiJoin { strategy, .. } => Some(*strategy), + _ => None, + }) + .expect("the block plans an AntiJoin"); + let mut names = Vec::new(); + if executed.result.num_rows() > 0 { + let batch = executed.result.concat_batches().unwrap(); + let column = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + names = (0..column.len()) + .map(|row| column.value(row).to_string()) + .collect(); + names.sort(); + } + (names, strategy) +} + +// INVARIANT 4: a key-set block keeps the rows its correlated form keeps. The +// key set runs the pattern from the bound binding, each traversal toward the +// outer binding reversed, so this holds only while every `Expand` emits the +// same pairs (or edge rows) in both orientations. Over collision-prone keys, +// cycles, self-loops and duplicated outer rows, each shape's row multiset as +// written equals its forced-correlated twin's, for every key. +#[test] +fn prop_key_set_block_eq_correlated() { + use omnigraph_planner::AntiJoinStrategy; + let rt = tokio::runtime::Runtime::new().unwrap(); + let mut runner = TestRunner::new(config()); + let key_set_runs = std::sync::atomic::AtomicUsize::new(0); + runner + .run(&arb_graph(), |graph| { + let mismatch = rt.block_on(async { + let (_dir, db) = load_graph(&graph).await; + for (outer, block) in KEY_SET_SHAPES { + for key in KEYS { + let params = one_param(key); + let (written, strategy) = block_run(&db, outer, block, "", ¶ms).await; + let (forced, twin) = + block_run(&db, outer, block, "$p.name = $p.name", ¶ms).await; + assert_eq!(twin, AntiJoinStrategy::Correlated, "{block}"); + if strategy == AntiJoinStrategy::KeySet { + key_set_runs.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } + if written != forced { + return Some((*block, *key, strategy, written, forced)); + } + } + } + None + }); + prop_assert!(mismatch.is_none(), "key-set divergence: {:?}", mismatch); + Ok(()) + }) + .unwrap(); + assert!( + key_set_runs.load(std::sync::atomic::Ordering::Relaxed) > 0, + "no written block ran as a key set, so the property compared nothing" + ); +} diff --git a/docs/dev/execution.md b/docs/dev/execution.md index 4bd85f433..d3d59a602 100644 --- a/docs/dev/execution.md +++ b/docs/dev/execution.md @@ -222,11 +222,16 @@ binding's identity column with no scan of it. An unbound `Expand` emits one row per distinct endpoint pair and a bound one one row per edge row, in either direction, so every key meets the same matches. Pass `anti_join_strategy` prices both from the summed row estimates of each chain (`chain_rows`, the -correlated one from the outer input's estimate), records -`anti_join_rows($x)`, and lowers the `key_set` tree only when it produces fewer -rows; an unknown estimate keeps `correlated`. The logical plan keeps the -correlated tree as the block's meaning; the strategy is the physical -`AntiJoin`'s `strategy` and the GQT line `anti join $x: key_set`. +correlated one from the outer input's estimate), records `anti_join_rows($x)`, +and lowers the `key_set` tree only when it produces fewer rows; an unknown +estimate keeps `correlated`. Each hop is priced at its edge type's average +degree, so a hub root is under-priced: with a tiny outer side, `key_set` can +read every edge of the hub, bounded by that root's neighbourhood and never by +the outer rows. The recorded estimates make such a choice visible. The logical +plan keeps the correlated tree as the block's meaning; the strategy is the +physical `AntiJoin`'s `strategy` and the GQT line `anti join $x: key_set`. Plan +acceptance refuses a strategy its inner tree does not have (check `block +strategy`), so an edited saved plan fails before it runs. A block stays `correlated` when its inner tree holds anything but `Expand`, dependent `TableScan` and `Filter` nodes (a nested block, a second diff --git a/docs/dev/testing.md b/docs/dev/testing.md index 867621e7b..caed2633a 100644 --- a/docs/dev/testing.md +++ b/docs/dev/testing.md @@ -60,7 +60,7 @@ The engine integration suite is grouped by behavior, not implementation module: | Frozen engine v1, the reference | `crates/omnigraph-reference-engine/tests/frozen.rs` pins every source file's bytes; `forbidden_apis.rs` owns that `omnigraph-gqt` is the crate's only dependent and that its source names neither the engine nor the planner; the crate has no other tests, and a `.gqt` step's `--- expect same as v1` compares v2's rows with its answer | | V2 plan nodes' operators and the execution report | In-source `engine/report/tests.rs` owns which operator each node builds and what its row says (`ran`, `actual_rows`, `drained`); `crates/omnigraph-planner/tests/integration/lower_walk.rs` owns the walk's order and refusals; the `ran` lines of `--- expect plan` read the report of the case's own run | | V2 plan sufficiency: the bound plan alone reproduces a run | `crates/omnigraph/tests/integration/engine_v2_plan_replay.rs` (see [Plan replay](#plan-replay)): one query per node kind with a switch or a ladder, replayed twice through `Session::replay_bound_plan` from its serialized replay envelope, equal rows, the same trace and the same search decisions required; `crates/omnigraph-planner/tests/integration/bound_plan.rs` owns the serialized form; in-source `engine/search.rs` tests own the declared ladder's stepping; `engine/plan_source.rs` tests own that the assumed memory limit sizes the plan, the lowering and the pool | -| V2 plan acceptance: every read plan checked against its checked query | In-source `crates/omnigraph-planner/src/validate/tests.rs` owns one broken plan per check (binding, traversal and search identity, predicate retention and origin, row cut, order and tie-break including fused and grouped searches, projection and score origin, hydration, declared policies, full-text prerequisites), the key-narrowing conjunct admitted only with its re-derived id, exact-fragment membership over admitted and excluded forms, the derivation's rules and rejections (hydrated members included), entry parity under an exhausted limit, and the `instrument:` derivation cost curve; `engine/plan_source/key_tests.rs` owns the narrowed key on a real snapshot and its refusal when edited; `integration/engine_v2_plan_replay.rs` owns that a saved plan failing well-formedness or a requirement is a bad request at the replay door; `validate/replay.rs` and `validate/budget.rs` tests own decoding limits, version refusals and budget charges; `integration/engine_v2.rs::every_read_door_shares_one_acceptance` owns that the run, both explain doors and the inspected run accept one plan and refuse alike; GQT cases claim the scope with `validation scope ` in `--- expect plan` | +| V2 plan acceptance: every read plan checked against its checked query | In-source `crates/omnigraph-planner/src/validate/tests.rs` owns one broken plan per check (binding, traversal and search identity, predicate retention and origin, block strategy, row cut, order and tie-break including fused and grouped searches, projection and score origin, hydration, declared policies, full-text prerequisites), the key-narrowing conjunct admitted only with its re-derived id, exact-fragment membership over admitted and excluded forms, the derivation's rules and rejections (hydrated members included), entry parity under an exhausted limit, and the `instrument:` derivation cost curve; `engine/plan_source/key_tests.rs` owns the narrowed key on a real snapshot and its refusal when edited; `integration/engine_v2_plan_replay.rs` owns that a saved plan failing well-formedness or a requirement is a bad request at the replay door; `validate/replay.rs` and `validate/budget.rs` tests own decoding limits, version refusals and budget charges; `integration/engine_v2.rs::every_read_door_shares_one_acceptance` owns that the run, both explain doors and the inspected run accept one plan and refuse alike; GQT cases claim the scope with `validation scope ` in `--- expect plan` | | V2 scrubbed replay: no input beside the bound plan and the snapshot reaches execution | `engine_v2_scrubbed_replay.rs` owns the mechanism (the `std::env` grep over `crates/omnigraph/src/engine/`, the replay under an ambient memory limit and an ambient `OMNIGRAPH_EXPAND_INDEXED_MAX_FRONTIER` the plan did not capture, `ann_nprobes` as a plan field, the `profile` rows); `integration/engine_v2_plan_replay.rs` owns the snapshot pins (an edge write or an insert after planning refuses the replay); the inventory cases `cases/v2/planner/input_*.gqt` own one input class each (`engine`, `ann_nprobes`, index build state), and the clock class is owned by `integration/engine_v2_plan_replay.rs`; `integration/engine_v2_memory.rs` owns `ran id_lookup` under a refused build | | V2 declared switches: the report names the side a run took | `cases/v2/planner/switch_ran_on_the_report.gqt` (the `ran` line of `--- expect plan`); in-source `engine/report/tests.rs` owns the switch's row and a scan whose plan declares no fallback | | Search and physical indexes | `search.rs`, `integration/scalar_indexes.rs`, `lance_surface_guards.rs`, `integration/rrf_prefilter_gate.rs` (the rrf plan gate's differential oracle and fences), `integration/repro_issue_563.rs` (`#[ignore]`d overflow-scale symptom tier) |