feat(engine): execute Direction::Both on both Expand arms

CSR arm: an undirected hop chains csr(edge).neighbors with
csc(edge).neighbors under the existing visited/seen_dst gates, so
both-direction pairs and self-loops dedup for free (set semantics),
including per-hop in variable-length BFS. Indexed arm: endpoint_probes
returns both (src,dst) and (dst,src) orientations for Both; the per-hop
scan runs once per orientation into one neighbor_map, deduped by the
existing per-source seen_dst sets. Bulk anti-join: "no edge in EITHER
direction" (csr || csc has_neighbors).

Also fixes a lowering gap the anti-join test caught red: negation
inners are typechecked into a discarded context clone, so the
ResolvedTraversal direction lookup silently fell back to Out inside
not{} — undirectedness now travels on the AST node itself (the syntax
is the source of truth), immune to context-propagation gaps.

Tests: traversal.rs (out ∪ in vs the directional miss, both-ways dedup,
var-length from an incoming-only node, undirected anti-join vs
directional), traversal_indexed.rs both_modes equivalence, and the
proptest arm-equivalence property widened with <knows>{1,3}.
This commit is contained in:
aaltshuler 2026-07-05 22:57:11 +03:00 committed by Andrew Altshuler
parent d4b21ce4eb
commit 9460f097d9
5 changed files with 254 additions and 40 deletions

View file

@ -267,14 +267,24 @@ fn lower_clauses(
))
})?;
let direction = type_ctx
.traversals
.iter()
.find(|rt| {
rt.src == traversal.src && rt.dst == traversal.dst && rt.edge_type == edge.name
})
.map(|rt| rt.direction)
.unwrap_or(Direction::Out);
// Undirected is carried on the AST node itself — negation inners
// are typechecked into a discarded context clone, so the
// ResolvedTraversal lookup below cannot see their direction; the
// syntax is the source of truth for Both.
let direction = if traversal.undirected {
Direction::Both
} else {
type_ctx
.traversals
.iter()
.find(|rt| {
rt.src == traversal.src
&& rt.dst == traversal.dst
&& rt.edge_type == edge.name
})
.map(|rt| rt.direction)
.unwrap_or(Direction::Out)
};
let dst_type = match direction {
Direction::Out => edge.to_type.clone(),

View file

@ -1050,6 +1050,8 @@ fn gather_cost_inputs(
let src_type = match direction {
Direction::Out => &edge_def.from_type,
Direction::In => &edge_def.to_type,
// Both requires from_type == to_type (typecheck T22).
Direction::Both => &edge_def.from_type,
};
let src_entry = snapshot.entry(&format!("node:{}", src_type))?;
Some(ExpandCostInputs {
@ -1109,7 +1111,20 @@ fn warn_on_degraded_coverage(
fn endpoint_columns(direction: Direction) -> (&'static str, &'static str) {
match direction {
Direction::Out => ("src", "dst"),
// Both: the primary orientation (used by the cost probe; the indexed
// execution loop adds the reverse probe itself via endpoint_probes).
Direction::In => ("dst", "src"),
Direction::Both => ("src", "dst"),
}
}
/// All (key, opposite) probes a direction requires: one for Out/In, both
/// orientations for an undirected traversal.
fn endpoint_probes(direction: Direction) -> &'static [(&'static str, &'static str)] {
match direction {
Direction::Out => &[("src", "dst")],
Direction::In => &[("dst", "src")],
Direction::Both => &[("src", "dst"), ("dst", "src")],
}
}
@ -1281,7 +1296,7 @@ async fn execute_expand_indexed(
// The keyed/opposite endpoint columns for this direction. The edge dataset
// and the C6 coverage warn are owned by the caller (`execute_expand`), which
// opens the dataset once and threads it in.
let (key_col, opp_col) = endpoint_columns(direction);
let probes = endpoint_probes(direction);
let max = max_hops.unwrap_or(min_hops.max(1));
// Cross-type edges cannot chain (a Company is not a `WorksAt` source), so a
@ -1344,30 +1359,38 @@ async fn execute_expand_indexed(
})
.collect();
let batches = crate::table_store::TableStore::scan_edges_by_endpoint(
&edge_ds, key_col, opp_col, &union_keys,
)
.await?;
// dense key -> dense neighbors (scan order; duplicates preserved, like CSR multi-edges).
// dense key -> dense neighbors (scan order; duplicates preserved, like
// CSR multi-edges). One probe per orientation: Out/In do one pass;
// Both merges the src-keyed and dst-keyed scans into one map (the
// per-source `seen_dst` gate below dedups pairs present both ways).
let mut neighbor_map: HashMap<u32, Vec<u32>> = HashMap::new();
for batch in &batches {
let keys = batch
.column_by_name(key_col)
.ok_or_else(|| OmniError::manifest(format!("edge batch missing '{}'", key_col)))?
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| OmniError::manifest(format!("edge '{}' is not Utf8", key_col)))?;
let opps = batch
.column_by_name(opp_col)
.ok_or_else(|| OmniError::manifest(format!("edge batch missing '{}'", opp_col)))?
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| OmniError::manifest(format!("edge '{}' is not Utf8", opp_col)))?;
for r in 0..batch.num_rows() {
let k = interner.get_or_insert(keys.value(r));
let o = interner.get_or_insert(opps.value(r));
neighbor_map.entry(k).or_default().push(o);
for &(key_col, opp_col) in probes {
let batches = crate::table_store::TableStore::scan_edges_by_endpoint(
&edge_ds, key_col, opp_col, &union_keys,
)
.await?;
for batch in &batches {
let keys = batch
.column_by_name(key_col)
.ok_or_else(|| {
OmniError::manifest(format!("edge batch missing '{}'", key_col))
})?
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| OmniError::manifest(format!("edge '{}' is not Utf8", key_col)))?;
let opps = batch
.column_by_name(opp_col)
.ok_or_else(|| {
OmniError::manifest(format!("edge batch missing '{}'", opp_col))
})?
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| OmniError::manifest(format!("edge '{}' is not Utf8", opp_col)))?;
for r in 0..batch.num_rows() {
let k = interner.get_or_insert(keys.value(r));
let o = interner.get_or_insert(opps.value(r));
neighbor_map.entry(k).or_default().push(o);
}
}
}
@ -1520,6 +1543,8 @@ async fn execute_expand_csr(
let (src_type_name, dst_type_name) = match direction {
Direction::Out => (&edge_def.from_type, &edge_def.to_type),
Direction::In => (&edge_def.to_type, &edge_def.from_type),
// Both requires from_type == to_type (typecheck T22).
Direction::Both => (&edge_def.from_type, &edge_def.from_type),
};
let src_type_idx = graph_index
@ -1530,10 +1555,20 @@ async fn execute_expand_csr(
.ok_or_else(|| OmniError::manifest(format!("no type index for '{}'", dst_type_name)))?;
let adj = match direction {
Direction::Out => graph_index.csr(edge_type),
Direction::Out | Direction::Both => graph_index.csr(edge_type),
Direction::In => graph_index.csc(edge_type),
}
.ok_or_else(|| OmniError::manifest(format!("no adjacency index for edge '{}'", edge_type)))?;
// Undirected: additionally walk incoming edges (CSC); the BFS gates below
// dedup pairs that exist in both directions and self-loops.
let adj_rev = match direction {
Direction::Both => Some(
graph_index
.csc(edge_type)
.ok_or_else(|| OmniError::manifest(format!("no adjacency index for edge '{}'", edge_type)))?,
),
_ => None,
};
let max = max_hops.unwrap_or(min_hops.max(1));
@ -1567,7 +1602,8 @@ async fn execute_expand_csr(
for hop in 1..=max {
let mut next_frontier = Vec::new();
for &node in &frontier {
for &neighbor in adj.neighbors(node) {
let rev: &[u32] = adj_rev.map(|a| a.neighbors(node)).unwrap_or(&[]);
for &neighbor in adj.neighbors(node).iter().chain(rev) {
if !same_type || visited.insert(neighbor) {
next_frontier.push(neighbor);
if hop >= min_hops && seen_dst_dense.insert(neighbor) {
@ -1735,12 +1771,17 @@ fn try_bulk_anti_join_mask(
let src_type_name = match direction {
Direction::Out => &edge_def.from_type,
Direction::In => &edge_def.to_type,
Direction::In | Direction::Both => &edge_def.to_type,
};
let adj = match direction {
Direction::Out => gi.csr(edge_type),
Direction::Out | Direction::Both => gi.csr(edge_type),
Direction::In => gi.csc(edge_type),
}?;
// Undirected anti-join: "no edge in EITHER direction".
let adj_rev = match direction {
Direction::Both => Some(gi.csc(edge_type)?),
_ => None,
};
let type_idx = gi.type_index(src_type_name)?;
let id_col_name = format!("{}.id", outer_var);
@ -1753,7 +1794,10 @@ fn try_bulk_anti_join_mask(
.map(|i| {
let id = outer_ids.value(i);
match type_idx.to_dense(id) {
Some(dense) => !adj.has_neighbors(dense),
Some(dense) => {
!adj.has_neighbors(dense)
&& !adj_rev.map(|a| a.has_neighbors(dense)).unwrap_or(false)
}
None => true, // not in graph index = no edges = keep
}
})

View file

@ -42,6 +42,13 @@ query friends($name: String) {
}
return { $f.name }
}
query related($name: String) {
match {
$p: Person { name: $name }
$p <knows>{1,3} $f
}
return { $f.name }
}
query employers($name: String) {
match {
$p: Person { name: $name }
@ -178,8 +185,9 @@ async fn col0_set(db: &mut Omnigraph, name: &str, params: &ParamMap) -> HashSet<
// INVARIANT 1: mode equivalence. For any generated graph and start key, the
// CSR, indexed, and auto paths return identical result multisets — over both a
// same-type traversal (knows{1,3}, exercises cycles/self-loops) and a cross-type
// one (worksAt{1,2}, collision-prone). This is the search-over-the-class version
// same-type traversal (knows{1,3}, exercises cycles/self-loops), its undirected
// form (<knows>{1,3} — Direction::Both must agree across arms, incl. dedup of
// pairs present both ways), and a cross-type one (worksAt{1,2}, collision-prone). This is the search-over-the-class version
// of the hand-built cross-type-collision fixture.
#[test]
fn prop_expand_indexed_eq_csr() {
@ -191,7 +199,7 @@ fn prop_expand_indexed_eq_csr() {
let (_dir, mut db) = load_graph(&graph).await;
for start in graph.persons.clone() {
let p = one_param(&start);
for q in ["friends", "employers"] {
for q in ["friends", "related", "employers"] {
// The seam is scope-bound: the forced mode is gone when the
// wrapped future resolves, so it never leaks across runs.
let csr = with_traversal_mode("csr", col0_sorted(&mut db, q, &p)).await;

View file

@ -8,6 +8,140 @@ use omnigraph_compiler::ir::ParamMap;
use helpers::*;
// ─── Undirected traversal (`$a <edge> $b`, Direction::Both) ─────────────────
//
// iss-gq-undirected-traversal: the CSR arm unions csr+csc under the existing
// per-source dedup gates; pairs present in both directions and self-loops
// appear once. Fixture Knows edges: Alice->Bob, Alice->Charlie, Bob->Diana.
#[tokio::test]
async fn undirected_one_hop_unions_out_and_in_neighbors() {
let dir = tempfile::tempdir().unwrap();
let mut db = init_and_load(&dir).await;
let queries = r#"
query connected($name: String) {
match {
$p: Person { name: $name }
$p <knows> $f
}
return { $f.name }
}
query connected_directional($name: String) {
match {
$p: Person { name: $name }
$p knows $f
}
return { $f.name }
}
"#;
// Directional from Bob misses the incoming Alice->Bob edge — the
// motivating dashboard bug.
let directional = query_main(
&mut db,
queries,
"connected_directional",
&params(&[("$name", "Bob")]),
)
.await
.unwrap();
assert_eq!(
first_column_sorted(&directional),
vec!["Diana"],
"directional sees only outgoing"
);
let undirected = query_main(&mut db, queries, "connected", &params(&[("$name", "Bob")]))
.await
.unwrap();
assert_eq!(
first_column_sorted(&undirected),
vec!["Alice", "Diana"],
"undirected sees out in"
);
// Dedup: add the reverse edge Diana->Bob so (Bob, Diana) exists both
// ways; Diana must still appear exactly once.
load_jsonl(
&mut db,
r#"{"edge": "Knows", "from": "Diana", "to": "Bob"}"#,
LoadMode::Merge,
)
.await
.unwrap();
let deduped = query_main(&mut db, queries, "connected", &params(&[("$name", "Bob")]))
.await
.unwrap();
assert_eq!(
first_column_sorted(&deduped),
vec!["Alice", "Diana"],
"a pair connected in both directions appears once (set semantics)"
);
}
#[tokio::test]
async fn undirected_variable_hops() {
let dir = tempfile::tempdir().unwrap();
let mut db = init_and_load(&dir).await;
let queries = r#"
query reach_both($name: String) {
match {
$p: Person { name: $name }
$p <knows>{1,2} $f
}
return { $f.name }
}
"#;
// Charlie's only edge is INCOMING (Alice->Charlie). Undirected: hop 1
// reaches Alice; hop 2 from Alice reaches Bob (out) — Charlie itself is
// the visited source, never re-emitted.
let result = query_main(&mut db, queries, "reach_both", &params(&[("$name", "Charlie")]))
.await
.unwrap();
assert_eq!(
first_column_sorted(&result),
vec!["Alice", "Bob"],
"undirected 2-hop frontier from a node with only incoming edges"
);
}
#[tokio::test]
async fn undirected_anti_join_excludes_both_directions() {
let dir = tempfile::tempdir().unwrap();
let mut db = init_and_load(&dir).await;
let queries = r#"
query isolated() {
match {
$p: Person
not { $p <knows> $x }
}
return { $p.name }
}
query no_outgoing() {
match {
$p: Person
not { $p knows $x }
}
return { $p.name }
}
"#;
// Directional `not`: keeps people with no OUTGOING edge — Charlie and
// Diana (both have only incoming).
let directional = query_main(&mut db, queries, "no_outgoing", &ParamMap::new())
.await
.unwrap();
assert_eq!(first_column_sorted(&directional), vec!["Charlie", "Diana"]);
// Undirected `not`: no edge in EITHER direction — every fixture person
// has at least one, so the result is empty.
let undirected = query_main(&mut db, queries, "isolated", &ParamMap::new())
.await
.unwrap();
assert_eq!(undirected.num_rows(), 0, "everyone touches a Knows edge");
}
// ─── Anti-join slow path (predicated negation) ──────────────────────────────
#[tokio::test]

View file

@ -123,6 +123,24 @@ async fn indexed_matches_csr_one_hop_same_type() {
assert_eq!(got, vec!["Bob", "Charlie"], "Alice knows Bob and Charlie");
}
#[tokio::test]
async fn indexed_matches_csr_undirected_one_hop() {
let dir = tempfile::tempdir().unwrap();
let mut db = init_and_load(&dir).await;
let queries = r#"
query connected($name: String) {
match {
$p: Person { name: $name }
$p <knows> $f
}
return { $f.name }
}
"#;
// Bob: outgoing Bob->Diana, incoming Alice->Bob — undirected sees both.
let got = both_modes(&mut db, queries, "connected", &params(&[("$name", "Bob")])).await;
assert_eq!(got, vec!["Alice", "Diana"], "out in neighbors of Bob");
}
#[tokio::test]
async fn indexed_matches_csr_multi_hop_same_type() {
let dir = tempfile::tempdir().unwrap();