# Issues #202 + #203 + #204 Hash-Join Cluster — Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Eliminate three O(N²) inner loops in the query engine — negation pre-filters (#202), disjunction branch evaluation (#203), and the general pattern join (#204) — by replacing nested fact scans with hash-based lookups.

**Architecture:** Three surgical fixes, one per site:
- **#202** (`executor.rs` + `evaluator.rs`): pre-compute not/not-join exclusion sets once, probe per outer binding.
- **#203** (`executor.rs`): evaluate or/or-join branches from an empty seed, hash-join results back onto incoming bindings.
- **#204** (`matcher.rs`): detect shared join variables between patterns and build a hash map from candidate facts, replacing the nested-loop scan in `join_with_pattern`.

Each fix is independent; implement in order (#202 → #203 → #204) since they touch different functions.

**Tech Stack:** Rust stable, `std::collections::HashMap` / `HashSet`, existing `PatternMatcher` and `evaluate_not_join` APIs.

---

## File Map

| Action | File | What changes |
|--------|------|--------------|
| Modify | `src/query/datalog/executor.rs:438-470` | #202 not-filter: pre-compute exclusion set before filter loop |
| Modify | `src/query/datalog/evaluator.rs:477-529` | #202 evaluate_not_join: pre-compute before outer binding loop |
| Modify | `src/query/datalog/executor.rs:1370-1448` | #203 apply_or_clauses: empty-seed branch eval + hash-join back |
| Modify | `src/query/datalog/matcher.rs:396-410` | #204 join_with_pattern: hash-join when shared variable detected |

---

### Task 1: #202 — Pre-compute not/not-join exclusion sets in `execute_query`

**Files:**
- Modify: `src/query/datalog/executor.rs:438-470`

**Background:** The current code at lines 438–470 calls `not_body_matches(not_body, binding, …)` for each outer `binding`. `not_body_matches` creates a fresh `PatternMatcher` and scans all facts every call → O(outer × inner).

The fix: before the filter loop, evaluate each not-body patterns once against `filtered_facts`, collect the set of "excluded" join-variable tuples, then check the set per binding.

- [ ] **Step 1: Write a failing test that would distinguish O(N) from O(N²)**

Add to the `#[cfg(test)]` module in `src/query/datalog/executor.rs`:

```rust
#[cfg(test)]
mod not_hash_join_tests {
    use crate::graph::FactStorage;
    use crate::query::datalog::executor::DatalogExecutor;
    

    fn make_not_db(n: usize, excluded: usize) -> DatalogExecutor {
        let storage = FactStorage::new();
        let exec = DatalogExecutor::new(storage);
        for batch_start in (0..n).step_by(100) {
            let batch_end = (batch_start + 100).min(n);
            let mut cmd = String::from("(transact [");
            for i in batch_start..batch_end {
                cmd.push_str(&format!("[:e{i} :val {i}]", i = i));
            }
            cmd.push_str("])");
            exec.execute(crate::query::datalog::parser::parse_datalog_command(&cmd).unwrap()).unwrap();
        }
        for batch_start in (0..excluded).step_by(100) {
            let batch_end = (batch_start + 100).min(excluded);
            let mut cmd = String::from("(transact [");
            for i in batch_start..batch_end {
                cmd.push_str(&format!("[:e{i} :banned true]", i = i));
            }
            cmd.push_str("])");
            exec.execute(crate::query::datalog::parser::parse_datalog_command(&cmd).unwrap()).unwrap();
        }
        exec
    }

    #[test]
    fn not_filter_returns_correct_count() {
        let n = 1_000;
        let excluded = n / 10; // 100 banned
        let exec = make_not_db(n, excluded);
        let result = exec
            .execute(
                crate::query::datalog::parser::parse_datalog_command(
                    "(query [:find ?e :where [?e :val ?v] (not [?e :banned true])])",
                )
                .unwrap(),
            )
            .unwrap();
        if let crate::query::datalog::executor::QueryResult::QueryResults { results, .. } = result
        {
            assert_eq!(
                results.len(),
                n - excluded,
                "expected {} results after not-filter",
                n - excluded
            );
        } else {
            panic!("expected QueryResults");
        }
    }

    #[test]
    fn not_join_filter_returns_correct_count() {
        let n = 1_000;
        let excluded = n / 10;
        let storage = FactStorage::new();
        let exec = DatalogExecutor::new(storage);
        for batch_start in (0..n).step_by(100) {
            let batch_end = (batch_start + 100).min(n);
            let mut cmd = String::from("(transact [");
            for i in batch_start..batch_end {
                cmd.push_str(&format!("[:e{i} :val {i}]", i = i));
            }
            cmd.push_str("])");
            exec.execute(crate::query::datalog::parser::parse_datalog_command(&cmd).unwrap()).unwrap();
        }
        exec.execute(crate::query::datalog::parser::parse_datalog_command("(transact [[:d-bad :status :bad]])").unwrap())
            .unwrap();
        for batch_start in (0..excluded).step_by(100) {
            let batch_end = (batch_start + 100).min(excluded);
            let mut cmd = String::from("(transact [");
            for i in batch_start..batch_end {
                cmd.push_str(&format!("[:e{i} :dep :d-bad]", i = i));
            }
            cmd.push_str("])");
            exec.execute(crate::query::datalog::parser::parse_datalog_command(&cmd).unwrap()).unwrap();
        }
        let result = exec
            .execute(
                crate::query::datalog::parser::parse_datalog_command(
                    "(query [:find ?e :where [?e :val ?v] \
                     (not-join [?e] [?e :dep ?d] [?d :status :bad])])",
                )
                .unwrap(),
            )
            .unwrap();
        if let crate::query::datalog::executor::QueryResult::QueryResults { results, .. } = result
        {
            assert_eq!(
                results.len(),
                n - excluded,
                "expected {} results after not-join-filter",
                n - excluded
            );
        } else {
            panic!("expected QueryResults");
        }
    }
}
```

- [ ] **Step 2: Run the tests (expect pass — correctness baseline)**

```bash
cargo test not_hash_join_tests 2>&1 | tail -8
```

Expected: both tests pass.

- [ ] **Step 3: Replace the not-filter loop with pre-computed exclusion sets**

In `execute_query` (around lines 438–470), the current code is:

```rust
let not_filtered: Vec<_> = if not_clauses.is_empty() && not_join_clauses.is_empty() {
    bindings
} else {
    bindings
        .into_iter()
        .filter(|binding| {
            for not_body in &not_clauses {
                if not_body_matches(
                    not_body,
                    binding,
                    filtered_facts.clone(),
                    valid_at_value.clone(),
                    &registry,
                ) {
                    return false;
                }
            }
            for (join_vars, nj_clauses) in &not_join_clauses {
                if evaluate_not_join(
                    join_vars,
                    nj_clauses,
                    binding,
                    filtered_facts.clone(),
                    &registry,
                ) {
                    return false;
                }
            }
            true
        })
        .collect()
};
```

Replace with:

```rust
let not_filtered: Vec<_> = if not_clauses.is_empty() && not_join_clauses.is_empty() {
    bindings
} else {
    // Pre-compute exclusion sets — one evaluation per not-body, not per outer binding.
    //
    // For each not-body, run pattern matching once against `filtered_facts` to get
    // all bindings where the body is satisfiable. Then collect the "join keys" (the
    // subset of variables that appear in the outer bindings) into a HashSet.
    // The filter loop below does one O(1) probe per outer binding instead of a full
    // pattern match.
    //
    // Edge case: expr-only bodies (no patterns) produce no pre-computed set and fall
    // through to `not_body_matches` as before (rare, already fast).
    use crate::query::datalog::evaluator::substitute_pattern;
    use std::collections::HashSet;

    // --- Not bodies ---
    // Each element: either Some(HashSet of excluded join-key tuples) or None (expr-only,
    // use slow path).
    let not_exclusion_sets: Vec<Option<HashSet<Vec<(String, Value)>>>> = not_clauses
        .iter()
        .map(|not_body| {
            let patterns: Vec<_> = not_body
                .iter()
                .filter_map(|c| match c {
                    WhereClause::Pattern(p) => Some(p.clone()),
                    _ => None,
                })
                .collect();
            if patterns.is_empty() {
                // Expr-only body: no pre-computation possible.
                return None;
            }
            let matcher = crate::query::datalog::matcher::PatternMatcher::from_slice_with_valid_at(
                filtered_facts.clone(),
                valid_at_value.clone(),
            );
            let body_bindings = matcher.match_patterns(&patterns);
            // Determine which variable names are join variables: those that appear in
            // the not-body patterns AND in the incoming bindings. We discover these
            // dynamically from the body binding keys.
            // We store all body bindings as sorted (key, value) vecs for probing.
            let exclusion_set: HashSet<Vec<(String, Value)>> = body_bindings
                .into_iter()
                .map(|mut b| {
                    // Drop hidden metadata keys (prefixed `__f`)
                    b.retain(|k, _| !k.starts_with("__f"));
                    let mut kv: Vec<(String, Value)> =
                        b.into_iter().collect();
                    kv.sort_unstable_by(|a, b| a.0.cmp(&b.0));
                    kv
                })
                .collect();
            Some(exclusion_set)
        })
        .collect();

    // --- Not-join bodies ---
    let not_join_exclusion_sets: Vec<Option<HashSet<Vec<(String, Value)>>>> =
        not_join_clauses
            .iter()
            .map(|(join_vars, nj_clauses)| {
                let patterns: Vec<_> = nj_clauses
                    .iter()
                    .filter_map(|c| match c {
                        WhereClause::Pattern(p) => Some(p.clone()),
                        _ => None,
                    })
                    .collect();
                if patterns.is_empty() {
                    return None;
                }
                let matcher =
                    crate::query::datalog::matcher::PatternMatcher::from_slice_with_valid_at(
                        filtered_facts.clone(),
                        valid_at_value.clone(),
                    );
                let body_bindings = matcher.match_patterns(&patterns);
                // Project to join_vars only.
                let exclusion_set: HashSet<Vec<(String, Value)>> = body_bindings
                    .into_iter()
                    .filter_map(|b| {
                        let kv: Vec<(String, Value)> = join_vars
                            .iter()
                            .filter_map(|v| b.get(v).map(|val| (v.clone(), val.clone())))
                            .collect();
                        if kv.len() == join_vars.len() {
                            Some(kv)
                        } else {
                            None
                        }
                    })
                    .collect();
                Some(exclusion_set)
            })
            .collect();

    bindings
        .into_iter()
        .filter(|binding| {
            // Check not-bodies via pre-computed exclusion sets (fast path) or
            // via not_body_matches (slow path for expr-only bodies).
            for (i, not_body) in not_clauses.iter().enumerate() {
                match &not_exclusion_sets[i] {
                    Some(exclusion_set) => {
                        // Build this binding's key (variables present in the exclusion set).
                        // Project binding to the variables in the not-body patterns.
                        let has_expr = not_body.iter().any(|c| matches!(c, WhereClause::Expr { .. }));
                        let patterns: Vec<_> = not_body
                            .iter()
                            .filter_map(|c| match c {
                                WhereClause::Pattern(p) => Some(substitute_pattern(p, binding)),
                                _ => None,
                            })
                            .collect();
                        // Collect the variable names from the pattern (post-substitution
                        // variables that are still unbound — shouldn't be any for bound
                        // outer bindings, but we handle the general case).
                        // Simple approach: intersect binding with the not-body pattern vars,
                        // form a sorted key, and probe the set.
                        // Get the variable names directly from the exclusion_set entries.
                        if exclusion_set.is_empty() {
                            // No excluding bindings → this outer binding is safe.
                            // But only if there are no expr clauses making it complex.
                            if !has_expr {
                                continue;
                            }
                            // Fall through to slow path if there are expr clauses.
                        }
                        if !has_expr && !patterns.is_empty() {
                            // Fast path: probe exclusion set using the outer binding's
                            // values for the join variables.
                            // Determine join variable names from a sample set entry.
                            if let Some(sample) = exclusion_set.iter().next() {
                                let key: Vec<(String, Value)> = sample
                                    .iter()
                                    .filter_map(|(var, _)| {
                                        binding
                                            .get(var)
                                            .map(|val| (var.clone(), val.clone()))
                                    })
                                    .collect();
                                // Only probe if all join vars are bound in the outer binding.
                                if key.len() == sample.len()
                                    && exclusion_set.contains(&key)
                                {
                                    return false;
                                }
                                continue;
                            }
                        }
                        // Slow path fallback (expr clauses or empty exclusion set with exprs)
                        if not_body_matches(
                            not_body,
                            binding,
                            filtered_facts.clone(),
                            valid_at_value.clone(),
                            &registry,
                        ) {
                            return false;
                        }
                    }
                    None => {
                        // Expr-only body: slow path.
                        if not_body_matches(
                            not_body,
                            binding,
                            filtered_facts.clone(),
                            valid_at_value.clone(),
                            &registry,
                        ) {
                            return false;
                        }
                    }
                }
            }

            // Check not-join-bodies.
            for (i, (join_vars, nj_clauses)) in not_join_clauses.iter().enumerate() {
                match &not_join_exclusion_sets[i] {
                    Some(exclusion_set) => {
                        let has_expr = nj_clauses
                            .iter()
                            .any(|c| matches!(c, WhereClause::Expr { .. }));
                        if !has_expr {
                            let key: Vec<(String, Value)> = join_vars
                                .iter()
                                .filter_map(|v| {
                                    binding.get(v).map(|val| (v.clone(), val.clone()))
                                })
                                .collect();
                            if key.len() == join_vars.len()
                                && exclusion_set.contains(&key)
                            {
                                return false;
                            }
                            continue;
                        }
                        // Fall through to slow path if expr clauses present.
                        if evaluate_not_join(
                            join_vars,
                            nj_clauses,
                            binding,
                            filtered_facts.clone(),
                            &registry,
                        ) {
                            return false;
                        }
                    }
                    None => {
                        if evaluate_not_join(
                            join_vars,
                            nj_clauses,
                            binding,
                            filtered_facts.clone(),
                            &registry,
                        ) {
                            return false;
                        }
                    }
                }
            }
            true
        })
        .collect()
};
```

- [ ] **Step 4: Build**

```bash
cargo build 2>&1 | head -20
```

Expected: no errors. Fix any `use` import issues (add `use crate::query::datalog::evaluator::substitute_pattern;` at the top of the function if needed — it's already imported at line 821 inside `not_body_matches`, but you may need it earlier in the file).

- [ ] **Step 5: Run not-hash-join tests**

```bash
cargo test not_hash_join_tests 2>&1 | tail -8
```

Expected: both tests pass.

- [ ] **Step 6: Run full test suite**

```bash
cargo test 2>&1 | tail -10
```

Expected: all tests pass.

- [ ] **Step 7: Commit**

```bash
git add src/query/datalog/executor.rs
git commit -m "perf: pre-compute not/not-join exclusion sets — O(N²) → O(N) (#202)"
```

---

### Task 2: #203 — Empty-seed branch evaluation in `apply_or_clauses`

**Files:**
- Modify: `src/query/datalog/executor.rs:1370-1448`

**Background:** `apply_or_clauses` calls `evaluate_branch(branch, bindings.clone(), …)` for each branch. `evaluate_branch` uses `match_patterns_seeded` which iterates every incoming binding × every fact → O(seeds × facts). The fix: evaluate branches from an empty seed, then hash-join results back onto incoming bindings.

- [ ] **Step 1: Write failing tests**

Add to the `#[cfg(test)]` module in `executor.rs`:

```rust
#[cfg(test)]
mod or_hash_join_tests {
    use crate::graph::FactStorage;
    use crate::query::datalog::executor::DatalogExecutor;
    

    fn make_or_db(n: usize, a_count: usize, b_count: usize) -> DatalogExecutor {
        let storage = FactStorage::new();
        let exec = DatalogExecutor::new(storage);
        for batch_start in (0..n).step_by(100) {
            let batch_end = (batch_start + 100).min(n);
            let mut cmd = String::from("(transact [");
            for i in batch_start..batch_end {
                cmd.push_str(&format!("[:e{i} :val {i}]", i = i));
            }
            cmd.push_str("])");
            exec.execute(crate::query::datalog::parser::parse_datalog_command(&cmd).unwrap()).unwrap();
        }
        for batch_start in (0..a_count).step_by(100) {
            let batch_end = (batch_start + 100).min(a_count);
            let mut cmd = String::from("(transact [");
            for i in batch_start..batch_end {
                cmd.push_str(&format!("[:e{i} :tag-a true]", i = i));
            }
            cmd.push_str("])");
            exec.execute(crate::query::datalog::parser::parse_datalog_command(&cmd).unwrap()).unwrap();
        }
        let b_start = n.saturating_sub(b_count);
        for batch_start in (b_start..n).step_by(100) {
            let batch_end = (batch_start + 100).min(n);
            let mut cmd = String::from("(transact [");
            for i in batch_start..batch_end {
                cmd.push_str(&format!("[:e{i} :tag-b true]", i = i));
            }
            cmd.push_str("])");
            exec.execute(crate::query::datalog::parser::parse_datalog_command(&cmd).unwrap()).unwrap();
        }
        exec
    }

    #[test]
    fn or_clause_returns_correct_count() {
        // 1000 entities, first 250 have :tag-a, last 250 have :tag-b, no overlap
        let n = 1_000;
        let a = n / 4;
        let b = n / 4;
        let exec = make_or_db(n, a, b);
        let result = exec
            .execute(
                crate::query::datalog::parser::parse_datalog_command(
                    "(query [:find ?e :where [?e :val ?v] \
                     (or [?e :tag-a true] [?e :tag-b true])])",
                )
                .unwrap(),
            )
            .unwrap();
        if let crate::query::datalog::executor::QueryResult::QueryResults { results, .. } = result
        {
            assert_eq!(
                results.len(),
                a + b,
                "expected {} results from or (a={} b={} no overlap)",
                a + b,
                a,
                b
            );
        } else {
            panic!("expected QueryResults");
        }
    }

    #[test]
    fn or_join_clause_returns_correct_count() {
        let n = 1_000;
        let a = n / 4;
        let b = n / 4;
        let exec = make_or_db(n, a, b);
        let result = exec
            .execute(
                crate::query::datalog::parser::parse_datalog_command(
                    "(query [:find ?e :where [?e :val ?v] \
                     (or-join [?e] [?e :tag-a true] [?e :tag-b true])])",
                )
                .unwrap(),
            )
            .unwrap();
        if let crate::query::datalog::executor::QueryResult::QueryResults { results, .. } = result
        {
            assert_eq!(
                results.len(),
                a + b,
                "expected {} results from or-join",
                a + b
            );
        } else {
            panic!("expected QueryResults");
        }
    }

    #[test]
    fn or_clause_with_overlap_deduplicates() {
        // 100 entities, all have both :tag-a and :tag-b → result should be 100, not 200
        let n = 100;
        let exec = make_or_db(n, n, n);
        let result = exec
            .execute(
                crate::query::datalog::parser::parse_datalog_command(
                    "(query [:find ?e :where [?e :val ?v] \
                     (or [?e :tag-a true] [?e :tag-b true])])",
                )
                .unwrap(),
            )
            .unwrap();
        if let crate::query::datalog::executor::QueryResult::QueryResults { results, .. } = result
        {
            assert_eq!(results.len(), n, "expected {} deduplicated results", n);
        } else {
            panic!("expected QueryResults");
        }
    }
}
```

- [ ] **Step 2: Run tests (expect pass — correctness baseline)**

```bash
cargo test or_hash_join_tests 2>&1 | tail -8
```

Expected: all 3 pass.

- [ ] **Step 3: Replace `apply_or_clauses` body**

Find `apply_or_clauses` at around line 1370. Replace the entire function body (keep the signature unchanged):

```rust
pub(crate) fn apply_or_clauses(
    clauses: &[WhereClause],
    mut bindings: Vec<Binding>,
    storage: Arc<[Fact]>,
    rules: &crate::query::datalog::rules::RuleRegistry,
    as_of: Option<AsOf>,
    valid_at: Option<ValidAt>,
    registry: &FunctionRegistry,
) -> anyhow::Result<Vec<Binding>> {
    for clause in clauses {
        match clause {
            WhereClause::Or(branches) => {
                // Evaluate every branch from an empty seed (independent of incoming bindings).
                // Then hash-join the union back onto incoming bindings on shared variables.
                let empty_seed: Vec<Binding> = vec![HashMap::new()];
                let mut union_bindings: Vec<Binding> = Vec::new();
                let mut seen_keys: std::collections::HashSet<Vec<(String, Value)>> =
                    std::collections::HashSet::new();

                for branch in branches {
                    let branch_result = evaluate_branch(
                        branch,
                        empty_seed.clone(),
                        storage.clone(),
                        rules,
                        as_of.clone(),
                        valid_at.clone(),
                        registry,
                    )?;
                    for b in branch_result {
                        let mut key: Vec<_> =
                            b.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
                        key.sort_unstable_by(|a, b| a.0.cmp(&b.0));
                        if seen_keys.insert(key) {
                            union_bindings.push(b);
                        }
                    }
                }

                // Determine shared variable names: variables present in both
                // incoming bindings and branch results.
                let branch_var_names: std::collections::HashSet<&str> = union_bindings
                    .iter()
                    .flat_map(|b| b.keys().map(|k| k.as_str()))
                    .collect();
                let shared_vars: Vec<String> = bindings
                    .iter()
                    .flat_map(|b| b.keys().cloned())
                    .collect::<std::collections::HashSet<_>>()
                    .into_iter()
                    .filter(|v| branch_var_names.contains(v.as_str()))
                    .collect();

                if shared_vars.is_empty() {
                    // No shared variables → replace bindings with branch union entirely.
                    bindings = union_bindings;
                    continue;
                }

                // Build HashMap: shared-key tuple → Vec<branch Binding>
                let mut branch_map: HashMap<Vec<(String, Value)>, Vec<Binding>> = HashMap::new();
                for b in union_bindings {
                    let key: Vec<(String, Value)> = shared_vars
                        .iter()
                        .filter_map(|v| b.get(v).map(|val| (v.clone(), val.clone())))
                        .collect();
                    branch_map.entry(key).or_default().push(b);
                }

                // For each incoming binding, look up matching branch results and merge.
                let mut result: Vec<Binding> = Vec::new();
                let mut seen_result: std::collections::HashSet<Vec<(String, Value)>> =
                    std::collections::HashSet::new();
                for incoming in &bindings {
                    let key: Vec<(String, Value)> = shared_vars
                        .iter()
                        .filter_map(|v| incoming.get(v).map(|val| (v.clone(), val.clone())))
                        .collect();
                    if let Some(matches) = branch_map.get(&key) {
                        for branch_binding in matches {
                            // Merge: start with incoming, extend with branch-introduced vars.
                            let mut merged = incoming.clone();
                            for (k, v) in branch_binding {
                                merged.entry(k.clone()).or_insert_with(|| v.clone());
                            }
                            let mut dedup_key: Vec<_> =
                                merged.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
                            dedup_key.sort_unstable_by(|a, b| a.0.cmp(&b.0));
                            if seen_result.insert(dedup_key) {
                                result.push(merged);
                            }
                        }
                    }
                }
                bindings = result;
            }

            WhereClause::OrJoin {
                join_vars,
                branches,
            } => {
                let outer_keys: std::collections::HashSet<String> =
                    bindings.iter().flat_map(|b| b.keys().cloned()).collect();

                let empty_seed: Vec<Binding> = vec![HashMap::new()];
                let mut projected: Vec<Binding> = Vec::new();
                let mut seen_proj: std::collections::HashSet<Vec<(String, Value)>> =
                    std::collections::HashSet::new();

                for branch in branches {
                    let branch_result = evaluate_branch(
                        branch,
                        empty_seed.clone(),
                        storage.clone(),
                        rules,
                        as_of.clone(),
                        valid_at.clone(),
                        registry,
                    )?;
                    for mut b in branch_result {
                        if !join_vars.iter().all(|v| b.contains_key(v)) {
                            continue;
                        }
                        // Project to outer_keys (preserves join_vars since join_vars ⊆ outer_keys)
                        b.retain(|k, _| outer_keys.contains(k));
                        let mut key: Vec<_> =
                            b.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
                        key.sort_unstable_by(|a, b| a.0.cmp(&b.0));
                        if seen_proj.insert(key) {
                            projected.push(b);
                        }
                    }
                }

                // Build HashMap keyed on join_vars tuple.
                let mut branch_map: HashMap<Vec<(String, Value)>, Vec<Binding>> = HashMap::new();
                for b in projected {
                    let key: Vec<(String, Value)> = join_vars
                        .iter()
                        .filter_map(|v| b.get(v).map(|val| (v.clone(), val.clone())))
                        .collect();
                    branch_map.entry(key).or_default().push(b);
                }

                let mut result: Vec<Binding> = Vec::new();
                let mut seen_result: std::collections::HashSet<Vec<(String, Value)>> =
                    std::collections::HashSet::new();
                for incoming in &bindings {
                    let key: Vec<(String, Value)> = join_vars
                        .iter()
                        .filter_map(|v| incoming.get(v).map(|val| (v.clone(), val.clone())))
                        .collect();
                    if let Some(matches) = branch_map.get(&key) {
                        for branch_binding in matches {
                            let mut merged = incoming.clone();
                            for (k, v) in branch_binding {
                                merged.entry(k.clone()).or_insert_with(|| v.clone());
                            }
                            let mut dedup_key: Vec<_> =
                                merged.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
                            dedup_key.sort_unstable_by(|a, b| a.0.cmp(&b.0));
                            if seen_result.insert(dedup_key) {
                                result.push(merged);
                            }
                        }
                    }
                }
                bindings = result;
            }

            _ => {}
        }
    }
    Ok(bindings)
}
```

- [ ] **Step 4: Build**

```bash
cargo build 2>&1 | head -20
```

Expected: no errors.

- [ ] **Step 5: Run or-hash-join tests**

```bash
cargo test or_hash_join_tests 2>&1 | tail -8
```

Expected: all 3 pass.

- [ ] **Step 6: Run full test suite**

```bash
cargo test 2>&1 | tail -10
```

Expected: all tests pass.

- [ ] **Step 7: Commit**

```bash
git add src/query/datalog/executor.rs
git commit -m "perf: empty-seed branch eval in apply_or_clauses — O(N²) → O(N) (#203)"
```

---

### Task 3: #204 — Hash-join in `join_with_pattern`

**Files:**
- Modify: `src/query/datalog/matcher.rs:396-410`

**Background:** `join_with_pattern` calls `match_pattern_with_bindings(pattern, existing)` for every `existing` binding — for each call, `match_pattern_with_bindings_scan` iterates all facts. This is O(existing × facts). When the pattern shares a variable (e.g. `?e`) with existing bindings, we can build a HashMap<join_key, Vec<new_bindings>> from the facts once, then probe per existing binding.

- [ ] **Step 1: Write failing tests**

Add to the `#[cfg(test)]` module in `src/query/datalog/matcher.rs`:

```rust
#[cfg(test)]
mod hash_join_tests {
    use crate::graph::FactStorage;
    use crate::query::datalog::executor::DatalogExecutor;
    

    fn make_join_db(n: usize, dept_count: usize) -> DatalogExecutor {
        let storage = FactStorage::new();
        let exec = DatalogExecutor::new(storage);
        for batch_start in (0..n).step_by(50) {
            let batch_end = (batch_start + 50).min(n);
            let mut cmd = String::from("(transact [");
            for i in batch_start..batch_end {
                cmd.push_str(&format!("[:e{i} :val {i}]", i = i));
                cmd.push_str(&format!("[:e{i} :dept :d{d}]", i = i, d = i % dept_count));
            }
            cmd.push_str("])");
            exec.execute(crate::query::datalog::parser::parse_datalog_command(&cmd).unwrap()).unwrap();
        }
        exec
    }

    #[test]
    fn two_pattern_join_returns_correct_count() {
        let n = 1_000;
        let exec = make_join_db(n, 10);
        let result = exec
            .execute(
                crate::query::datalog::parser::parse_datalog_command(
                    "(query [:find ?dept (sum ?v) :with ?e \
                     :where [?e :dept ?dept] [?e :val ?v]])",
                )
                .unwrap(),
            )
            .unwrap();
        if let crate::query::datalog::executor::QueryResult::QueryResults { results, .. } = result
        {
            // 10 departments → 10 result rows
            assert_eq!(results.len(), 10, "expected 10 department groups");
        } else {
            panic!("expected QueryResults");
        }
    }

    #[test]
    fn two_pattern_join_values_are_correct() {
        // 10 entities, 2 depts. Dept :d0 gets e0,e2,e4,e6,e8 (vals 0,2,4,6,8 → sum=20)
        // Dept :d1 gets e1,e3,e5,e7,e9 (vals 1,3,5,7,9 → sum=25)
        let n = 10;
        let exec = make_join_db(n, 2);
        let result = exec
            .execute(
                crate::query::datalog::parser::parse_datalog_command(
                    "(query [:find ?dept (sum ?v) :with ?e \
                     :where [?e :dept ?dept] [?e :val ?v]])",
                )
                .unwrap(),
            )
            .unwrap();
        if let crate::query::datalog::executor::QueryResult::QueryResults {
            vars, results, ..
        } = result
        {
            assert_eq!(results.len(), 2, "expected 2 department groups");
            // Find the sum for :d0
            let dept_idx = vars.iter().position(|v| v == "?dept").unwrap_or(0);
            let sum_idx = 1 - dept_idx; // the other column
            let mut sums: std::collections::HashMap<String, i64> = std::collections::HashMap::new();
            for row in &results {
                if let (
                    crate::graph::types::Value::Keyword(dept),
                    crate::graph::types::Value::Integer(sum),
                ) = (&row[dept_idx], &row[sum_idx])
                {
                    sums.insert(dept.clone(), *sum);
                }
            }
            assert_eq!(sums.get(":d0"), Some(&20), "sum for :d0 should be 20");
            assert_eq!(sums.get(":d1"), Some(&25), "sum for :d1 should be 25");
        } else {
            panic!("expected QueryResults");
        }
    }
}
```

- [ ] **Step 2: Run tests (expect pass — baseline)**

```bash
cargo test hash_join_tests 2>&1 | tail -8
```

Expected: both pass.

- [ ] **Step 3: Replace `join_with_pattern` with hash-join implementation**

Find `join_with_pattern` at around line 396. Replace it (keep `match_pattern_with_bindings` and all other methods unchanged):

```rust
/// Join existing bindings with a new pattern.
///
/// Uses a hash-join when the pattern shares a variable with existing bindings
/// (the common `?e`-join shape). Falls back to the nested-loop scan when no
/// shared join variable is found (unrelated patterns, fully-literal patterns,
/// or pseudo-attribute patterns which have their own fast path in
/// `match_pattern_with_bindings`).
fn join_with_pattern(
    &self,
    existing_bindings: Vec<Bindings>,
    pattern: &Pattern,
) -> Vec<Bindings> {
    if existing_bindings.is_empty() {
        return vec![];
    }

    // Detect the join variable: the first variable in entity, then value position
    // of the new pattern that also appears as a bound key in the existing bindings.
    // We skip pseudo-attributes (they have a dedicated fast path in
    // match_pattern_with_bindings) and fully-literal patterns (nested loop is
    // already O(1) per binding for those).
    let is_pseudo = matches!(&pattern.attribute, AttributeSpec::Pseudo(_));

    let join_var: Option<String> = if is_pseudo {
        None
    } else {
        // Check entity position first.
        let entity_var = match &pattern.entity {
            EdnValue::Symbol(s) if s.starts_with('?') => {
                // It's a variable. Is it bound in existing bindings?
                if existing_bindings
                    .first()
                    .map_or(false, |b| b.contains_key(s))
                {
                    Some(s.clone())
                } else {
                    None
                }
            }
            _ => None,
        };
        if entity_var.is_some() {
            entity_var
        } else {
            // Check value position.
            match &pattern.value {
                EdnValue::Symbol(s) if s.starts_with('?') => {
                    if existing_bindings
                        .first()
                        .map_or(false, |b| b.contains_key(s))
                    {
                        Some(s.clone())
                    } else {
                        None
                    }
                }
                _ => None,
            }
        }
    };

    let Some(jvar) = join_var else {
        // No join variable detected: fall back to nested-loop scan.
        let mut results = Vec::new();
        for existing in existing_bindings {
            let new_matches = self.match_pattern_with_bindings(pattern, &existing);
            results.extend(new_matches);
        }
        return results;
    };

    // Hash-join path:
    // 1. Scan all candidate facts for this pattern from an empty binding (no outer context).
    //    This gives us every binding the pattern can produce, keyed by the join variable.
    let candidate_bindings = self.match_pattern(&pattern.clone());
    // Note: match_pattern uses an empty binding — it produces all possible bindings.

    // 2. Build HashMap: join_var_value → Vec<Bindings from pattern>
    let mut build_side: HashMap<Value, Vec<Bindings>> = HashMap::new();
    for b in candidate_bindings {
        if let Some(val) = b.get(&jvar) {
            build_side.entry(val.clone()).or_default().push(b);
        }
    }

    // 3. Probe: for each existing binding, look up join_var value.
    let mut results = Vec::new();
    for existing in existing_bindings {
        let Some(probe_val) = existing.get(&jvar) else {
            // Outer binding doesn't have the join var yet — fall back to scan for this row.
            let new_matches = self.match_pattern_with_bindings(pattern, &existing);
            results.extend(new_matches);
            continue;
        };
        if let Some(matches) = build_side.get(probe_val) {
            for candidate in matches {
                // Merge and consistency-check (reuse existing logic via a combined binding).
                let mut merged = existing.clone();
                let mut consistent = true;
                for (var, val) in candidate {
                    if var.starts_with("__f") {
                        merged.insert(var.clone(), val.clone());
                        continue;
                    }
                    if let Some(existing_val) = existing.get(var) {
                        if existing_val != val {
                            consistent = false;
                            break;
                        }
                    } else {
                        merged.insert(var.clone(), val.clone());
                    }
                }
                if consistent {
                    results.push(merged);
                }
            }
        }
        // No match in build side → this outer binding is dropped (inner join semantics).
    }
    results
}
```

- [ ] **Step 4: Build**

```bash
cargo build 2>&1 | head -20
```

Expected: no errors. Note: `match_pattern` is already a public method on `PatternMatcher` (line 71 of matcher.rs).

- [ ] **Step 5: Run hash-join tests**

```bash
cargo test hash_join_tests 2>&1 | tail -8
```

Expected: both tests pass.

- [ ] **Step 6: Run full test suite**

```bash
cargo test 2>&1 | tail -10
```

Expected: all tests pass. This is the most impactful change — if any join test fails, check that the consistency-check logic handles all `Value` variants and hidden `__f` keys correctly (compare with the original `match_pattern_with_bindings_scan`).

- [ ] **Step 7: Commit**

```bash
git add src/query/datalog/matcher.rs
git commit -m "perf: hash-join in join_with_pattern — O(N²) → O(N) (#204)"
```

---

### Task 4: Final verification and benchmark smoke-check

**Files:** None changed.

- [ ] **Step 1: Run full test suite**

```bash
cargo test 2>&1 | tail -5
```

Expected: all tests pass.

- [ ] **Step 2: Smoke-check the relevant benchmarks (quick, not a full run)**

```bash
cargo bench negation -- --sample-size 3 2>&1 | grep "not_scale"
cargo bench disjunction -- --sample-size 3 2>&1 | grep "or_scale"
cargo bench aggregation -- --sample-size 3 2>&1 | grep "with_grouped_sum"
```

Expected: all three groups complete without panics. Numbers will vary; what matters is they finish without errors.

- [ ] **Step 3: Verify no `{:?}` debug-format of sensitive types in new test code**

```bash
grep -n '{:?}' src/query/datalog/executor.rs src/query/datalog/matcher.rs src/query/datalog/evaluator.rs | grep -v "//.*{:?}" | head -20
```

Expected: any matches should be in existing code, not in the tests added in this plan. New test assertions in this plan use `.len()`, direct `Value` comparisons, and `.is_empty()` — not `{:?}` on Result/Fact/Value.

- [ ] **Step 4: Final commit summary**

```bash
git log --oneline -6
```

Expected output (roughly):
```
<hash> perf: hash-join in join_with_pattern — O(N²) → O(N) (#204)
<hash> perf: empty-seed branch eval in apply_or_clauses — O(N²) → O(N) (#203)
<hash> perf: pre-compute not/not-join exclusion sets — O(N²) → O(N) (#202)
<hash> test: add selective lookup correctness tests (#208)   ← from Plan 1 PR
…
```
