From b5326d0a52a33218691a66e867873c72fd08cbb9 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 22 Sep 2026 21:15:34 -0400 Subject: [PATCH] Corgi join: one path for every projection The corgi join routed projections that `compilable` rejected (a `case`, list intro, a data-driven tag) through a second path: join with the identity projection, then apply the terms as a rebased row-environment `Project`. The premise was that join projections are compiled before any container is in hand. They are not: `CorgiJoinBackend` compiles the projection per block against the matched columns' shapes, so every term it can lower as a `Project` it can lower in the join. Removes `compilable`, `rebase_join_term`, and the fallback branch; the `join_fallback` test program stays, with its header rewritten. Measured (ddir_server, corgi backend, joins of two random 1M-edge relations on 200k nodes, then 50 ticks of 1000 changes; two runs each, master-next -> this): a `case` projection loads in 1.43-1.46s -> 1.48s with churn unchanged (437-438ms -> 439-440ms for the 50 ticks); a plain projection is unchanged (1.44-1.45s, 436-438ms). The small load cost is the per-block compile of the `case` term, which the fallback compiled once; caching the graph by shape in the join would remove it if it matters. Co-Authored-By: Claude Opus 5.5 (1M context) --- interactive/src/backend/corgi.rs | 62 ++------------------ interactive/src/corgi/logic.rs | 30 ---------- interactive/tests/programs/join_fallback.ddp | 13 +--- 3 files changed, 7 insertions(+), 98 deletions(-) diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index 636d58105..243fcbba9 100644 --- a/interactive/src/backend/corgi.rs +++ b/interactive/src/backend/corgi.rs @@ -30,9 +30,9 @@ use crate::corgi::exchange::CorgiPact; use crate::corgi::join::CorgiJoinBackend; use crate::corgi::reduce::CorgiReduceBackend; use differential_dataflow::operators::int_proxy::{ProxyJoinTactic, ProxyReduceTactic}; -use crate::corgi::logic::{compilable, compile_flatmap, compile_predicate, compile_projection, compile_scalar, shape_of_row}; +use crate::corgi::logic::{compile_flatmap, compile_predicate, compile_projection, compile_scalar, shape_of_row}; use corgi::{Graph, NumOp, Shape}; -use crate::ir::{Diff, LinearOp, Projection, Reducer, Term, Time, Value as DValue}; +use crate::ir::{Diff, LinearOp, Projection, Reducer, Time, Value as DValue}; use crate::scope_ir as st; /// A DDIR row, an update, the corgi container on dataflow edges, and the columnar trace — @@ -41,45 +41,6 @@ type Row = DValue; type CC = CorgiContainer; type CTrace = differential_dataflow::trace::chunk::ChunkSpine>; -/// Rebase a join-projection term from the join environment (`$0`=key, `$1`=left val, -/// `$2`=right val) onto the row environment of the identity join's output -/// (`$0`=key, `$1`=(left val, right val)): `$1 -> $1[0]`, `$2 -> $1[1]`, structurally -/// everywhere. `Bound` binders are scope-relative and pass through untouched. -fn rebase_join_term(t: &Term) -> Term { - use crate::ir::Term::*; - match t { - Var(0) => Var(0), - Var(1) => Proj(Box::new(Var(1)), 0), - Var(2) => Proj(Box::new(Var(1)), 1), - Var(n) => panic!("join projection references ${n}"), - Bound(k) => Bound(*k), - Int(n) => Int(*n), - Tuple(fs) => Tuple(fs.iter().map(rebase_join_term).collect()), - List(fs) => List(fs.iter().map(rebase_join_term).collect()), - Spread(inner) => Spread(Box::new(rebase_join_term(inner))), - Proj(inner, i) => Proj(Box::new(rebase_join_term(inner)), *i), - Inject { tag, payload, sum } => Inject { tag: Box::new(rebase_join_term(tag)), payload: Box::new(rebase_join_term(payload)), sum: sum.clone() }, - Case { scrutinee, arms, default } => Case { - scrutinee: Box::new(rebase_join_term(scrutinee)), - arms: arms.iter().map(rebase_join_term).collect(), - default: default.as_ref().map(|d| Box::new(rebase_join_term(d))), - }, - Fold { list, init, step } => Fold { - list: Box::new(rebase_join_term(list)), - init: Box::new(rebase_join_term(init)), - step: Box::new(rebase_join_term(step)), - }, - If { cond, then, els } => If { - cond: Box::new(rebase_join_term(cond)), - then: Box::new(rebase_join_term(then)), - els: Box::new(rebase_join_term(els)), - }, - Binary(op, l, r) => Binary(*op, Box::new(rebase_join_term(l)), Box::new(rebase_join_term(r))), - Unary(op, inner) => Unary(*op, Box::new(rebase_join_term(inner))), - Hash(args) => Hash(args.iter().map(rebase_join_term).collect()), - } -} - /// The compiled form of one `LinearOp`, pinned to the shapes it was compiled against. Shapes are /// static per collection, so a chain compiles ONCE, on the first non-empty batch, and every later /// batch reuses the graph; a batch of a different shape is the invariant violation, not a @@ -293,23 +254,8 @@ impl Backend for CorgiBackend { // The proxy-join seam drives the backend blockwise under the driver's fuel; the backend // compiles the projection per container (shape-directed, for `Spread`) and emits corgi // columns directly as `CorgiContainer`s — column-native, no row round-trip. - if compilable(&projection.key) && compilable(&projection.val) { - let tactic = ProxyJoinTactic::new(CorgiJoinBackend::new(projection.key.clone(), projection.val.clone())); - join_with_tactic::<_, _, _, CC>(l, r, "Join", tactic).as_collection() - } else { - // Projections the lowering can't compile take the same shape as `linear`'s gate: - // join with the identity projection (compilable by construction), then apply the - // original terms as a row-wise `Project`, rebased from the join env - // `[$0=key, $1=left val, $2=right val]` onto the row env `[$0=key, $1=(lv, rv)]`. - // Capability never depends on the lowering's coverage; only speed does. - use crate::ir::Term; - let key = Term::Var(0); - let val = Term::Tuple(vec![Term::Var(1), Term::Var(2)]); - let tactic = ProxyJoinTactic::new(CorgiJoinBackend::new(key, val)); - let joined = join_with_tactic::<_, _, _, CC>(l, r, "Join", tactic).as_collection(); - let rebased = Projection { key: rebase_join_term(&projection.key), val: rebase_join_term(&projection.val) }; - Self::linear(joined, vec![LinearOp::Project(rebased)], 0) - } + let tactic = ProxyJoinTactic::new(CorgiJoinBackend::new(projection.key.clone(), projection.val.clone())); + join_with_tactic::<_, _, _, CC>(l, r, "Join", tactic).as_collection() } fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer) -> Self::Arr<'s> { diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs index b99e326a2..630a37899 100644 --- a/interactive/src/corgi/logic.rs +++ b/interactive/src/corgi/logic.rs @@ -189,36 +189,6 @@ fn mentions_env(t: &Term, depth: usize) -> bool { } } -/// Whether [`compile`] can lower this term WITHOUT knowing its operands' shapes — the gate for -/// join-INLINE projections, which are compiled before any container is in hand. It is therefore -/// deliberately narrower than `compile`: every shape-dependent form (`List`, `Case`, a built-in -/// sum whose lane the payload fixes, a data-driven tag) answers false here and is compiled by the -/// linear stage the join defers it to, which does have shapes. -pub fn compilable(t: &Term) -> bool { - match t { - Term::Var(_) | Term::Bound(_) | Term::Int(_) => true, - Term::Proj(inner, _) | Term::Spread(inner) => compilable(inner), - Term::Tuple(fs) => fs.iter().all(compilable), - Term::Binary(_, l, r) => compilable(l) && compilable(r), - Term::If { cond, then, els } => compilable(cond) && compilable(then) && compilable(els), - Term::Fold { list, init, step } => compilable(list) && compilable(init) && compilable(step), - // Keep this exhaustive so a new unary operator needs an explicit - // decision about whether it supports shape-free lowering. - Term::Unary(op, inner) => match op { - UnOp::Neg | UnOp::ToF64 | UnOp::F64Neg | UnOp::Not | UnOp::Len | UnOp::IsTag(_) => compilable(inner), - }, - // A literal tag into a declared type knows its whole sum; the built-ins and a data-driven - // tag need the payload's shape. - Term::Inject { tag, payload, sum } => { - matches!(&**tag, Term::Int(_)) && matches!(sum, SumTy::Declared(_)) && compilable(payload) - } - // `Op::Hash` is shape-generic (it folds whatever structure it is handed), so `hash` - // needs no shapes to lower and can answer true here. - Term::Hash(args) => args.iter().all(compilable), - _ => false, // List intro, Case — see `compile`. - } -} - /// The lane shapes of the sum an `Inject` builds: the declaration's, or a built-in's with the /// payload in its lane and the other lane from `expected`. fn lanes_of(sum: &SumTy, tag: usize, payload: &Shape, expected: Option<&Shape>) -> Res> { diff --git a/interactive/tests/programs/join_fallback.ddp b/interactive/tests/programs/join_fallback.ddp index 12aec3e6a..1fa0575a1 100644 --- a/interactive/tests/programs/join_fallback.ddp +++ b/interactive/tests/programs/join_fallback.ddp @@ -1,13 +1,6 @@ --- A join whose projection the SHAPE-FREE gate cannot admit: `compilable` runs before any --- container is in hand, so it declines every shape-dependent term — here a `case`, whose arm --- homogeneity is a fact about the data. The join therefore joins with the identity projection --- and applies the original terms as a rebased row-environment `Project`, which the shape-aware --- lowering then compiles columnar. The gate is agreement with the vec backend. --- --- This is a PERMANENT path, not a gap: `compilable` will always be narrower than `compile`, --- because the join has no shapes to reason with. (It previously used `hash`, which stopped --- driving the fallback once `hash` became corgi's structural hash — a test keyed to a hole in --- the compiler rather than to a property of the design.) +-- A join whose projection depends on the data's shape: a `case`, whose arm homogeneity is a +-- fact about the data. The join compiles its projection against each block's actual column +-- shapes, so such terms need no separate path. The gate is agreement with the vec backend. type L = L u64; let left = input 0 | key($0[0] ; $0[1]);