diff --git a/crates/omnigraph-compiler/src/ir/lower.rs b/crates/omnigraph-compiler/src/ir/lower.rs index 1f00f96c..1d5ca5de 100644 --- a/crates/omnigraph-compiler/src/ir/lower.rs +++ b/crates/omnigraph-compiler/src/ir/lower.rs @@ -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(), diff --git a/crates/omnigraph/src/exec/query.rs b/crates/omnigraph/src/exec/query.rs index a8ddf22e..106ee9f3 100644 --- a/crates/omnigraph/src/exec/query.rs +++ b/crates/omnigraph/src/exec/query.rs @@ -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> = 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::() - .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::() - .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::() + .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::() + .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 } }) diff --git a/crates/omnigraph/tests/proptest_equivalence.rs b/crates/omnigraph/tests/proptest_equivalence.rs index 3b4011fa..823a4124 100644 --- a/crates/omnigraph/tests/proptest_equivalence.rs +++ b/crates/omnigraph/tests/proptest_equivalence.rs @@ -42,6 +42,13 @@ query friends($name: String) { } return { $f.name } } +query related($name: String) { + match { + $p: Person { name: $name } + $p {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 ({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; diff --git a/crates/omnigraph/tests/traversal.rs b/crates/omnigraph/tests/traversal.rs index 2f518fd7..20af41c3 100644 --- a/crates/omnigraph/tests/traversal.rs +++ b/crates/omnigraph/tests/traversal.rs @@ -8,6 +8,140 @@ use omnigraph_compiler::ir::ParamMap; use helpers::*; +// ─── Undirected traversal (`$a $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 $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", + ¶ms(&[("$name", "Bob")]), + ) + .await + .unwrap(); + assert_eq!( + first_column_sorted(&directional), + vec!["Diana"], + "directional sees only outgoing" + ); + + let undirected = query_main(&mut db, queries, "connected", ¶ms(&[("$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", ¶ms(&[("$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 {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", ¶ms(&[("$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 $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] diff --git a/crates/omnigraph/tests/traversal_indexed.rs b/crates/omnigraph/tests/traversal_indexed.rs index cadd3d23..805c295b 100644 --- a/crates/omnigraph/tests/traversal_indexed.rs +++ b/crates/omnigraph/tests/traversal_indexed.rs @@ -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 $f + } + return { $f.name } +} +"#; + // Bob: outgoing Bob->Diana, incoming Alice->Bob — undirected sees both. + let got = both_modes(&mut db, queries, "connected", ¶ms(&[("$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();