From adf1ec22b4da90f71fb102f18aa2a80d1a510a65 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 10 Oct 2026 21:08:17 -0400 Subject: [PATCH] DDIR explain: keep the user's key and value whole in demand rows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Explain's demand rows spliced the user's value into one flat tuple with the iteration chain and query id: `(K ; V fields…, chain…, q)`. That needed each collection's field counts, which explain read from projection syntax, and it broke user terms that read a whole row. An always-true `filter($1 == tuple($1[0]))` gave an insufficient explanation, a flatmap list reading `$0` saw explain's (key, row) pair, a projection given as one term counted as one field whatever its shape, and keys or values that are not products were read by field position. Demand rows are now `(K ; (V, chain…, q))`, with the user's key and value whole. No rule takes them apart, so explain needs no shapes or field counts: the syntax-based counting and every width parameter of the reverse rules are gone. User terms read `$0` as the key and `$1` as the value's first field. A scope's own sites are wrapped as `(V)` before their first lift, so lifts give `(V, chain…)`; root-level host forms and sources enter the explain scope as `(K ; (V))`. Demand sets stay the user's rows, for the forward clone and the exports. `explain` and `explain_with` take no source shapes, and `source_arities` and `export_shape` are removed. A query row is `(key ; (val, q))`. The server's load option is `explain` or `explain=debug`; the arity is no longer needed. The demand-excess report is unchanged from before, row for row. Co-Authored-By: Claude Opus 5.5 --- interactive/examples/dump_explain.rs | 13 +- interactive/server/README.md | 8 +- interactive/server/src/cmd.rs | 59 +-- interactive/src/explain/decouple.rs | 229 +++++------ interactive/src/explain/mod.rs | 585 ++++++++++----------------- interactive/tests/explain.rs | 165 +++++--- 6 files changed, 460 insertions(+), 599 deletions(-) diff --git a/interactive/examples/dump_explain.rs b/interactive/examples/dump_explain.rs index 348d482a6..0f17c479f 100644 --- a/interactive/examples/dump_explain.rs +++ b/interactive/examples/dump_explain.rs @@ -1,16 +1,15 @@ //! Print a program's scope-tree IR, before and after the explanation rewrite. //! -//! Usage: `dump_explain `. +//! Usage: `dump_explain `. //! //! Output is the structural tree dump (readability, not parseability): //! per scope, imports and vars first, items in order with `Sub`s nested, //! then binds and exports. -use interactive::{lower, parse, scope_ir}; +use interactive::{lower, parse}; fn main() { - let path = std::env::args().nth(1).expect("usage: dump_explain "); - let arity: usize = std::env::args().nth(2).unwrap_or("2".into()).parse().unwrap(); + let path = std::env::args().nth(1).expect("usage: dump_explain "); let source = interactive::load_program(&path); let original = lower::lower_tree(parse::pipe::parse(&source)); @@ -20,11 +19,7 @@ fn main() { println!("-- ===================================================="); original.dump(); - let source_shapes: Vec<(usize, usize)> = original.root.imports.iter().map(|imp| match &imp.from { - scope_ir::Source::Input(_) | scope_ir::Source::Trace(_) => (arity, 0usize), - scope_ir::Source::Parent(_) => unreachable!("root scope cannot import from a parent"), - }).collect(); - let rewritten = interactive::explain::explain(&original, &source_shapes); + let rewritten = interactive::explain::explain(&original); println!(); println!("-- ===================================================="); diff --git a/interactive/server/README.md b/interactive/server/README.md index 240e11531..0a8580b17 100644 --- a/interactive/server/README.md +++ b/interactive/server/README.md @@ -47,10 +47,10 @@ scripts can be piped to stdin (see `demo/`). The useful commands are `load`, `drop`, `list`, `feed`, `bind`, `unbind`, `peek`, `tail`, `stop`, `tick`, and `exit`. `load from ` installs a program file (`.ddp`, pipe syntax); `load from -explain=[,debug]` applies the explanation rewrite first, every source -taken to have `arity` key fields and no value, after which the -query input is the one after the program's own and the demand sets are -`peek`able exports. `feed from ` fills an input from a +explain` applies the explanation rewrite first (`explain=debug` also taps +every demand collection with an inspect), after which the query input is +the one after the program's own, its rows `(key ; (val, q))`, and the demand +sets are `peek`able exports. `feed from ` fills an input from a source the server reads itself — a recipe such as `random:nodes=N,edges=E,churn=C` or `iota:N`, or a file of integer rows — with each worker taking its shard, so no row crosses the wire; a `churn=C` recipe then replaces `C` rows on every diff --git a/interactive/server/src/cmd.rs b/interactive/server/src/cmd.rs index 27ebd4dda..f11a72982 100644 --- a/interactive/server/src/cmd.rs +++ b/interactive/server/src/cmd.rs @@ -38,15 +38,14 @@ pub enum Cmd { /// registered trace name or a builtin call (`random(...)`). /// `program` — DDIR text, inline (`load … begin` … `end-load`) or read /// from a file (`load from `). - /// `explain` — `explain=[,debug]`: apply the explanation rewrite, - /// every source without a declared shape taken to have `arity` key fields - /// and no value; the query input is the one after the program's own, the - /// demand sets its exports. + /// `explain` — `explain` or `explain=debug`: apply the explanation rewrite, + /// with `debug` tapping every demand collection with an inspect; the query + /// input is the one after the program's own, the demand sets its exports. Load { id_hint: String, bindings: BTreeMap, program: String, - explain: Option<(usize, bool)>, + explain: Option, }, /// Drop the named dataflow. Fails if any export of this dataflow is /// still imported by another live dataflow or held by a reader. @@ -132,11 +131,9 @@ pub fn prepare(command: Cmd) -> Result { .map_err(panic_message)?; apply_bindings(&mut program, &bindings)?; // The rewrite precedes optimization (its rules assume single-op Linears). - if let Some((arity, debug)) = explain { - // A source's declared shape gives its field counts; `arity` covers the rest. - let shapes = interactive::explain::source_arities(&program, (arity, 0)); + if let Some(debug) = explain { let options = interactive::explain::Options { debug_inspects: debug }; - program = catch_unwind(AssertUnwindSafe(|| interactive::explain::explain_with(&program, &shapes, options))) + program = catch_unwind(AssertUnwindSafe(|| interactive::explain::explain_with(&program, options))) .map_err(panic_message)?; } program.optimize(); @@ -297,7 +294,7 @@ struct PendingLoad { reqid: ReqId, id_hint: String, bindings: BTreeMap, - explain: Option<(usize, bool)>, + explain: Option, body: String, /// Set once the body exceeds the size gate; the rest of the body is /// swallowed (not stored) and `end-load` reports this error, so an @@ -498,7 +495,7 @@ enum ParseOutcome { BeginLoad { id_hint: String, bindings: BTreeMap, - explain: Option<(usize, bool)>, + explain: Option, }, BeginFeed { prog: String, @@ -506,19 +503,6 @@ enum ParseOutcome { }, } -/// `explain=[,debug]`: the key arity of sources without a declared shape, -/// and whether to tap every demand collection with an inspect. -fn parse_explain(spec: &str) -> Result<(usize, bool), String> { - let (arity, debug) = match spec.split_once(',') { - Some((arity, "debug")) => (arity, true), - Some(_) => return Err(format!("load: explain=[,debug], got {spec:?}")), - None => (spec, false), - }; - let arity = arity - .parse() - .map_err(|_| format!("load: explain= arity must be a number, got {arity:?}"))?; - Ok((arity, debug)) -} /// A `` for `feed`: a comma-separated integer row → `Tuple`, `_` or /// empty → unit, else a closed scalar term (one whitespace-free token, e.g. @@ -588,7 +572,7 @@ fn parse_cmd(cmd: &str, args: &[&str]) -> ParseOutcome { // file, read here on the session thread. `load [options] begin` // switches the parser into // body-collection mode; subsequent lines are program text terminated by - // ` end-load`. Options: `explain=[,debug]`, `name=binding`. + // ` end-load`. Options: `explain`, `explain=debug`, `name=binding`. const USAGE: &str = "load: expected ` from [options]` or ` [options] begin`"; let Some((id_hint, rest)) = args.split_first() else { return ParseOutcome::Err(USAGE.into()); @@ -601,21 +585,14 @@ fn parse_cmd(cmd: &str, args: &[&str]) -> ParseOutcome { let mut bindings = BTreeMap::new(); let mut explain = None; for tok in options { - if let Some(spec) = tok.strip_prefix("explain=") { - explain = match parse_explain(spec) { - Ok(explain) => Some(explain), - Err(error) => return ParseOutcome::Err(error), - }; - continue; - } - if *tok == "--explain" { - return ParseOutcome::Err( - "load: `--explain` needs the sources' arity: `explain=[,debug]`".into(), - ); + match *tok { + "explain" => { explain = Some(false); continue; } + "explain=debug" => { explain = Some(true); continue; } + _ => {} } let Some((k, v)) = tok.split_once('=') else { return ParseOutcome::Err(format!( - "load: argument {:?} must be `explain=[,debug]` or `name=binding`", + "load: argument {:?} must be `explain`, `explain=debug`, or `name=binding`", tok )); }; @@ -901,23 +878,25 @@ mod tests { std::fs::write(&ddp2, "export \"s\" = input 0;\n").unwrap(); let mut p = LineParser::default(); let got = feed_all(&mut p, &[ - &format!("r1 load p from {} explain=2,debug e=random(seed=1,arity=2,range=8,count=12)\n", ddp.display()), + &format!("r1 load p from {} explain=debug e=random(seed=1,arity=2,range=8,count=12)\n", ddp.display()), &format!("r2 load q from {}\n", ddp2.display()), "r3 load p from /nonexistent/p.ddp\n", "r4 load p --explain begin\n", + &format!("r5 load p from {} explain\n", ddp.display()), ]); match &got[0].1 { Ok(Cmd::Load { id_hint, bindings, program, explain }) => { assert_eq!(id_hint, "p"); assert_eq!(program, "export \"r\" = input 0;\n"); - assert_eq!(*explain, Some((2, true))); + assert_eq!(*explain, Some(true)); assert_eq!(bindings.get("e").map(String::as_str), Some("random(seed=1,arity=2,range=8,count=12)")); } other => panic!("unexpected {other:?}"), } assert!(matches!(&got[1].1, Ok(Cmd::Load { explain: None, .. }))); assert!(matches!(&got[2].1, Err(e) if e.starts_with("load: cannot read"))); - assert!(matches!(&got[3].1, Err(e) if e.contains("explain="))); + assert!(matches!(&got[3].1, Err(e) if e.contains("`explain`"))); + assert!(matches!(&got[4].1, Ok(Cmd::Load { explain: Some(false), .. }))); let _ = std::fs::remove_dir_all(&dir); } diff --git a/interactive/src/explain/decouple.rs b/interactive/src/explain/decouple.rs index ca43c2cc5..b5731c3e0 100644 --- a/interactive/src/explain/decouple.rs +++ b/interactive/src/explain/decouple.rs @@ -15,17 +15,19 @@ //! (K ; V, chain, [q]) //! ``` //! -//! where `;` separates the *key* from the *value*. The value packs three -//! logical parts, in order: +//! where `;` separates the *key* from the *value*. The key `K` is the user's +//! key, whole. The value is a tuple of three logical parts, in order: //! -//! * **`V`** — the underlying data value. The -//! demanded output's value on the `dep` side, the candidate input's value on -//! the pair-table side. +//! * **`V`** — the user's value, whole, as one field. The demanded output's +//! value on the `dep` side, the candidate input's value on the pair-table +//! side. //! * **`chain`** — the loop-iteration coordinates, innermost-first, length = //! the node's scope depth. Time lives in data here. //! * **`q`** — a single trailing query id, present on `dep` rows (the thing //! being explained), absent on pair-table rows. //! +//! Neither `K` nor `V` is taken apart, so no rule depends on their shapes. +//! //! ## Reading the method contracts //! //! Each method returns a projection/predicate that the rule applies to a row of @@ -34,10 +36,9 @@ //! right input's value; before a join, `$0`/`$1` are key/value. A method's doc //! gives the *input row shape it expects* and the *output row shape it //! produces*, both in the `(K ; …)` notation above; an implementer needs only -//! those shapes, not the flat positions. Sizes arrive as parameters: `k`/`v` -//! column counts, `*_len`/`*_depth` chain lengths, `min` toggling Min's -//! value-narrowing. Predicate methods return `None` for "no constraint" (the -//! rule then skips the filter). +//! those shapes, not the positions. Sizes arrive as parameters: `*_len`/`*_depth` +//! chain lengths, `min` toggling Min's value-narrowing. Predicate methods return +//! `None` for "no constraint" (the rule then skips the filter). /// Builds the projections/predicates the reverse rules need, in terms of the /// demand envelope `(K ; V, chain, [q])` documented at the module level. @@ -45,27 +46,21 @@ pub trait RowModel { type Proj; type Pred; - // --- packing / unpacking data and the join key --- + // --- packing data into the join key --- /// Move the data `(K, V)` into the join key, keeping the row's `chain` (and - /// `q` when `has_q`): `(K ; V, chain, [q]) -> (pack[K,V] ; chain, [q])`. The - /// chain and the query id are kept as distinct parts — `chain_len` and - /// `has_q` are passed separately (not folded into a column count) so a - /// nested model can preserve their boundary. Pair rows pass `has_q = false`, - /// `dep` rows `has_q = true`, and a full-row distinct `chain_len = 0`. - fn pack(k: usize, v: usize, chain_len: usize, has_q: bool) -> Self::Proj; - /// Unpack a packed-and-distinct key back to a row: `(pack[K,V] ; ) -> (K ; V)`. - fn distinct_unpack(k: usize, v: usize) -> Self::Proj; - /// The identity reshaping `(K ; V ++ rest) -> (K ; V)` — drops a row's - /// chain/`q` (used by demand-set seeding and the semijoin). - fn project_kv(k: usize, v: usize) -> Self::Proj; + /// `q` when `has_q`): `(K ; V, chain, [q]) -> ((K, V) ; chain, [q])`. Pair + /// rows pass `has_q = false`, `dep` rows `has_q = true`. + fn pack(chain_len: usize, has_q: bool) -> Self::Proj; + /// Drop a row's chain and `q`, leaving the user's row: `(K ; V, chain, [q]) -> (K ; V)`. + fn strip_all() -> Self::Proj; // --- shape-preserving lookup (Concat/Leave) --- - /// Reassemble after the SP join on `pack[K,V]`. Input: the joined row - /// `[$0 = pack[K,V], $1 = chain_out ++ q (dep), $2 = chain_in (pair)]`. + /// Reassemble after the SP join on `(K, V)`. Input: the joined row + /// `[$0 = (K, V), $1 = chain_out ++ q (dep), $2 = chain_in (pair)]`. /// Output: `(K ; V, chain_in, chain_out, q)`. - fn sp_reassemble(k: usize, v: usize, in_len: usize, out_len: usize) -> Self::Proj; + fn sp_reassemble(in_len: usize, out_len: usize) -> Self::Proj; // --- keyed lookup (Reduce) --- @@ -73,69 +68,67 @@ pub trait RowModel { /// `[$0 = K, $1 = V_out ++ chain_out ++ q (dep), $2 = V_in ++ chain_in (pair)]`. /// Output: `(K ; V_in, [V_out,] chain_in, chain_out, q)` — `V_out` is /// carried only when `min`, for the narrowing below. - fn ky_join(k: usize, v_in: usize, v_out: usize, in_len: usize, out_len: usize, min: bool) -> Self::Proj; - /// Min value-narrowing on the `min` row above: keep rows with `V_in == V_out` - /// (both `v_in` columns of the value, at `[0..v_in]` and `[v_in..2*v_in]`). - fn ky_data_eq(v_in: usize) -> Option; + fn ky_join(in_len: usize, out_len: usize, min: bool) -> Self::Proj; + /// Min value-narrowing on the `min` row above: keep rows with `V_in == V_out`. + fn ky_data_eq() -> Option; /// Drop the carried `V_out`, restoring the SP-shaped row /// `(K ; V_in, chain_in, chain_out, q)`. - fn ky_drop_vout(k: usize, v_in: usize, v_out: usize, in_len: usize, out_len: usize) -> Self::Proj; + fn ky_drop_vout(in_len: usize, out_len: usize) -> Self::Proj; // --- lossy lookup (Linear[Project]) --- /// Pair re-key: apply the user `proj.key` to a pair row to compute /// `K_out`, carrying the input data through. - /// `(K_in ; V_in, chain_in) -> (K_out ; K_in, V_in, chain_in)`. - fn lossy_pair(proj: &Self::Proj, k_in: usize, v_in: usize, in_len: usize) -> Self::Proj; + /// `(K_in ; V_in, chain_in) -> (K_out ; K_in, V_in, chain_in)`. `proj` reads + /// the pair row's slots, so the caller has rebuilt any read of the user's row. + fn lossy_pair(proj: &Self::Proj, in_len: usize) -> Self::Proj; /// Reassemble after the join on `K_out`. Input: /// `[$0 = K_out, $1 = V_out ++ chain_out ++ q (dep), $2 = K_in ++ V_in ++ chain_in (pair)]`. /// Output: `(K_in ; V_in, chain_in, chain_out, q)`. - fn lossy_reassemble(k_in: usize, v_in: usize, v_out: usize, in_len: usize, out_len: usize) -> Self::Proj; + fn lossy_reassemble(in_len: usize, out_len: usize) -> Self::Proj; // --- join lookup (two inputs; #758 filters both times before splitting) --- /// Forward-join the two pair tables on `K`, applying the user `proj`. Input: /// `[$0 = K, $1 = V_L ++ chain_L (left), $2 = V_R ++ chain_R (right)]`. /// Output: `(K_out ; K, V_L, V_R, chain_L, chain_R, V_out)`, where - /// `K_out`/`V_out` are `proj.key`/`proj.val` over `(K, V_L, V_R)`. - fn join_forward(proj: &Self::Proj, k: usize, v_l: usize, v_r: usize, l_len: usize, r_len: usize) -> Self::Proj; + /// `K_out`/`V_out` are `proj.key`/`proj.val` over `(K, V_L, V_R)`, reading the + /// joined row's slots as for `lossy_pair`. + fn join_forward(proj: &Self::Proj, l_len: usize, r_len: usize) -> Self::Proj; /// Reassemble after joining `dep` against the forward-join pair on `K_out`, /// into one wide row carrying everything both sides' filters and splits need. /// Input: `[$0 = K_out, $1 = V_out ++ chain_out ++ q (dep), $2 = pair row above]`. /// Output: `(K ; V_L, V_R, chain_L, chain_R, chain_out, q, V_out_dep, V_out_pair)`. - #[allow(clippy::too_many_arguments)] - fn join_combined(k: usize, v_l: usize, v_r: usize, v_out: usize, l_len: usize, r_len: usize, out_len: usize) -> Self::Proj; + fn join_combined(l_len: usize, r_len: usize, out_len: usize) -> Self::Proj; /// Narrowing on the wide row: `chain_L ≤ chain_out` ∧ `chain_R ≤ chain_out` /// (both outer-aligned) ∧ `V_out_dep == V_out_pair`. Both sides' times are /// filtered here, before either side's chain is projected away (#758). - fn join_filter(v_l: usize, v_r: usize, v_out: usize, l_len: usize, r_len: usize, out_len: usize) -> Option; + fn join_filter(l_len: usize, r_len: usize, out_len: usize) -> Option; /// Project one side's contrib out of the filtered wide row. `left` selects /// `(K ; V_L, chain_L, q)` vs `(K ; V_R, chain_R, q)`. - #[allow(clippy::too_many_arguments)] - fn join_split(left: bool, k: usize, v_l: usize, v_r: usize, l_len: usize, r_len: usize, out_len: usize) -> Self::Proj; + fn join_split(left: bool, l_len: usize, r_len: usize, out_len: usize) -> Self::Proj; // --- the chain algebra, shared by SP / keyed / lossy --- /// Soundness filter on a `(K ; V, chain_in, chain_out, q)` row: keep rows /// with `chain_in ≤ chain_out`, compared outer-aligned (an input can only /// explain an output that came no earlier). `None` when the chains share no - /// coordinate. (`v` locates the chains after the `v`-column value.) - fn time_le(v: usize, in_len: usize, out_len: usize) -> Option; + /// coordinate. + fn time_le(in_len: usize, out_len: usize) -> Option; /// Strip `chain_out` and the inner `chain_in` coords past `keep` from a /// `(K ; V, chain_in, chain_out, q)` row, leaving `(K ; V, chain_in[..keep], q)`. /// Kept coords are the *outer* ones, matching `time_le`'s alignment. - fn strip(k: usize, v: usize, in_len: usize, out_len: usize, keep: usize) -> Self::Proj; + fn strip(in_len: usize, out_len: usize, keep: usize) -> Self::Proj; // --- bind iter-decrement (inverts a loop variable's feedback) --- /// Keep only rows whose innermost chain coordinate is `> 0` (iteration-0 - /// demand has no body-side source). The coordinate sits just after the - /// `v`-column value. - fn bind_filter(v: usize) -> Option; + /// demand has no body-side source). + fn bind_filter() -> Option; /// Decrement the innermost chain coordinate by one, leaving everything else /// intact: `(K ; V, chain, q) -> (K ; V, chain with chain[0]-1, q)`. /// `user_len` is the chain length. - fn bind_decrement(k: usize, v: usize, user_len: usize) -> Self::Proj; + fn bind_decrement(user_len: usize) -> Self::Proj; } /// Orchestration backend: collection handles and the four primitive ops. @@ -153,75 +146,65 @@ fn opt_filter(df: &mut D, c: D::Handle, p: Option) -> D::H match p { Some(p) => df.filter(&c, p), None => c } } -/// One side's inputs to a reverse rule: its pair-table halves, shape, chain length. +/// One side's inputs to a reverse rule: its pair-table halves and chain length. pub struct SideInfo { pub witness: D::Handle, pub forward: D::Handle, - pub shape: (usize, usize), pub user_len: usize, } /// Shape-preserving lookup (Concat/Leave). `keep == in_len` here. pub fn shape_preserving_lookup(df: &mut D, dep: &D::Handle, side: &SideInfo, out_depth: usize) -> D::Handle where M: RowModel, D: Dataflow { - let (k, v) = side.shape; let in_len = side.user_len; let pair = df.concat(vec![side.witness.clone(), side.forward.clone()]); - let pair_keyed = df.project(&pair, M::pack(k, v, in_len, false)); - let dep_keyed = df.project(dep, M::pack(k, v, out_depth, true)); - let joined = df.join(&dep_keyed, &pair_keyed, M::sp_reassemble(k, v, in_len, out_depth)); - let filtered = opt_filter(df, joined, M::time_le(v, in_len, out_depth)); - df.project(&filtered, M::strip(k, v, in_len, out_depth, in_len)) + let pair_keyed = df.project(&pair, M::pack(in_len, false)); + let dep_keyed = df.project(dep, M::pack(out_depth, true)); + let joined = df.join(&dep_keyed, &pair_keyed, M::sp_reassemble(in_len, out_depth)); + let filtered = opt_filter(df, joined, M::time_le(in_len, out_depth)); + df.project(&filtered, M::strip(in_len, out_depth, in_len)) } /// Keyed (Reduce) lookup; `min` enables value-narrowing. -pub fn keyed_lookup(df: &mut D, dep: &D::Handle, side: &SideInfo, output_shape: (usize, usize), out_len: usize, reducer_min: bool) -> D::Handle +pub fn keyed_lookup(df: &mut D, dep: &D::Handle, side: &SideInfo, out_len: usize, min: bool) -> D::Handle where M: RowModel, D: Dataflow { - let (k, v_in) = side.shape; let in_len = side.user_len; - let (_, v_out) = output_shape; - let min = reducer_min && v_in == v_out && v_in > 0; let pair = df.concat(vec![side.witness.clone(), side.forward.clone()]); - let joined = df.join(dep, &pair, M::ky_join(k, v_in, v_out, in_len, out_len, min)); + let joined = df.join(dep, &pair, M::ky_join(in_len, out_len, min)); let after = if min { - let f = opt_filter(df, joined, M::ky_data_eq(v_in)); - df.project(&f, M::ky_drop_vout(k, v_in, v_out, in_len, out_len)) + let f = opt_filter(df, joined, M::ky_data_eq()); + df.project(&f, M::ky_drop_vout(in_len, out_len)) } else { joined }; - let filtered = opt_filter(df, after, M::time_le(v_in, in_len, out_len)); - df.project(&filtered, M::strip(k, v_in, in_len, out_len, in_len)) + let filtered = opt_filter(df, after, M::time_le(in_len, out_len)); + df.project(&filtered, M::strip(in_len, out_len, in_len)) } /// Lossy (Project) lookup, through the pair table. -pub fn lossy_lookup(df: &mut D, dep: &D::Handle, side: &SideInfo, output_shape: (usize, usize), out_len: usize, proj: &M::Proj) -> D::Handle +pub fn lossy_lookup(df: &mut D, dep: &D::Handle, side: &SideInfo, out_len: usize, proj: &M::Proj) -> D::Handle where M: RowModel, D: Dataflow { - let (k_in, v_in) = side.shape; let in_len = side.user_len; - let (_, v_out) = output_shape; let pair = df.concat(vec![side.witness.clone(), side.forward.clone()]); - let pair_keyed = df.project(&pair, M::lossy_pair(proj, k_in, v_in, in_len)); - let joined = df.join(dep, &pair_keyed, M::lossy_reassemble(k_in, v_in, v_out, in_len, out_len)); - let filtered = opt_filter(df, joined, M::time_le(v_in, in_len, out_len)); - df.project(&filtered, M::strip(k_in, v_in, in_len, out_len, in_len)) + let pair_keyed = df.project(&pair, M::lossy_pair(proj, in_len)); + let joined = df.join(dep, &pair_keyed, M::lossy_reassemble(in_len, out_len)); + let filtered = opt_filter(df, joined, M::time_le(in_len, out_len)); + df.project(&filtered, M::strip(in_len, out_len, in_len)) } /// Join lookup (#758): forward-join pair tables, one combined dep×pair join /// carrying both chains, one combined time+value filter, then split per side. -pub fn join_lookup(df: &mut D, dep: &D::Handle, left: &SideInfo, right: &SideInfo, output_shape: (usize, usize), out_len: usize, proj: &M::Proj) -> (D::Handle, D::Handle) +pub fn join_lookup(df: &mut D, dep: &D::Handle, left: &SideInfo, right: &SideInfo, out_len: usize, proj: &M::Proj) -> (D::Handle, D::Handle) where M: RowModel, D: Dataflow { - let (k, v_l) = left.shape; - let (_, v_r) = right.shape; - let (_, v_out) = output_shape; let l_len = left.user_len; let r_len = right.user_len; let lp = df.concat(vec![left.witness.clone(), left.forward.clone()]); let rp = df.concat(vec![right.witness.clone(), right.forward.clone()]); - let pair = df.join(&lp, &rp, M::join_forward(proj, k, v_l, v_r, l_len, r_len)); - let joined = df.join(dep, &pair, M::join_combined(k, v_l, v_r, v_out, l_len, r_len, out_len)); - let timed = opt_filter(df, joined, M::join_filter(v_l, v_r, v_out, l_len, r_len, out_len)); - let lc = df.project(&timed, M::join_split(true, k, v_l, v_r, l_len, r_len, out_len)); - let rc = df.project(&timed, M::join_split(false, k, v_l, v_r, l_len, r_len, out_len)); + let pair = df.join(&lp, &rp, M::join_forward(proj, l_len, r_len)); + let joined = df.join(dep, &pair, M::join_combined(l_len, r_len, out_len)); + let timed = opt_filter(df, joined, M::join_filter(l_len, r_len, out_len)); + let lc = df.project(&timed, M::join_split(true, l_len, r_len, out_len)); + let rc = df.project(&timed, M::join_split(false, l_len, r_len, out_len)); (lc, rc) } @@ -271,7 +254,7 @@ mod mem { #[cfg(test)] mod value_contract { //! Executable contract for the reverse rules: by-example specs on `Value` - //! rows in `Val`'s envelope `[V | chain | q]`, run through the in-memory + //! rows in `Val`'s envelope `(K ; V, chain, [q])`, run through the in-memory //! dataflow against `crate::explain::Val`. use super::*; @@ -282,75 +265,81 @@ mod value_contract { type Row = Value; - /// A 1-field key `(n)`. - fn key(n: i64) -> Row { Value::Tuple(vec![Value::Int(n)]) } - /// A value tuple from a flat field list (`[V… | chain… | q]`). - fn val(xs: &[i64]) -> Row { Value::Tuple(xs.iter().map(|&n| Value::Int(n)).collect()) } - fn side(witness: Coll, shape: (usize, usize), user_len: usize) -> SideInfo { - SideInfo { witness, forward: vec![], shape, user_len } + /// A one-field tuple `(n)`: the keys and values below. + fn one(n: i64) -> Row { Value::Tuple(vec![Value::Int(n)]) } + /// An envelope value: `V`, then the chain and `q` (or the chain alone). + fn env(v: i64, rest: &[i64]) -> Row { + let mut fields = vec![one(v)]; + fields.extend(rest.iter().map(|&n| Value::Int(n))); + Value::Tuple(fields) + } + fn side(witness: Coll, user_len: usize) -> SideInfo { + SideInfo { witness, forward: vec![], user_len } } fn sorted(mut c: Coll) -> Coll { c.sort(); c } fn proj(k: Term, v: Term) -> Projection { Projection { key: k, val: v } } - /// `Spread($n)` — a bare-`$n` projection field (splices row n's value). - fn spread(n: usize) -> Term { Term::Tuple(vec![Term::Spread(Box::new(Term::Var(n)))]) } + /// Slot `n`'s user value: the whole key for `0`, else the envelope's first field. + fn user(n: usize) -> Term { + if n == 0 { Term::Var(0) } else { Term::Proj(Box::new(Term::Var(n)), 0) } + } #[test] fn sp_depth0_keeps_only_value_matching_pair() { // Packing (K,V) into the join key means only the V=7 pair can match. - let pair = vec![(key(5), val(&[7])), (key(5), val(&[8]))]; - let dep = vec![(key(5), val(&[7, 9]))]; // (K=5 ; V=7, q=9) - let out = shape_preserving_lookup::(&mut Mem, &dep, &side(pair, (1, 1), 0), 0); - assert_eq!(out, vec![(key(5), val(&[7, 9]))]); + let pair = vec![(one(5), env(7, &[])), (one(5), env(8, &[]))]; + let dep = vec![(one(5), env(7, &[9]))]; // (K=5 ; V=7, q=9) + let out = shape_preserving_lookup::(&mut Mem, &dep, &side(pair, 0), 0); + assert_eq!(out, vec![(one(5), env(7, &[9]))]); } #[test] fn sp_depth1_time_filters_late_inputs() { // chain length 1: an input at iter 2 explains an output demanded at // iter 3 (kept); one at iter 4 does not (dropped). - let keep = vec![(key(5), val(&[7, 2]))]; // chain_in = 2 - let drop = vec![(key(5), val(&[7, 4]))]; // chain_in = 4 - let dep = vec![(key(5), val(&[7, 3, 9]))]; // chain_out = 3, q = 9 - let kept = shape_preserving_lookup::(&mut Mem, &dep, &side(keep, (1, 1), 1), 1); - assert_eq!(kept, vec![(key(5), val(&[7, 2, 9]))]); - let dropped = shape_preserving_lookup::(&mut Mem, &dep, &side(drop, (1, 1), 1), 1); + let keep = vec![(one(5), env(7, &[2]))]; // chain_in = 2 + let drop = vec![(one(5), env(7, &[4]))]; // chain_in = 4 + let dep = vec![(one(5), env(7, &[3, 9]))]; // chain_out = 3, q = 9 + let kept = shape_preserving_lookup::(&mut Mem, &dep, &side(keep, 1), 1); + assert_eq!(kept, vec![(one(5), env(7, &[2, 9]))]); + let dropped = shape_preserving_lookup::(&mut Mem, &dep, &side(drop, 1), 1); assert!(dropped.is_empty()); } #[test] fn keyed_min_narrows_to_the_demanded_value() { - let pair = vec![(key(5), val(&[7])), (key(5), val(&[6]))]; - let dep = vec![(key(5), val(&[7, 9]))]; - let out = keyed_lookup::(&mut Mem, &dep, &side(pair, (1, 1), 0), (1, 1), 0, true); - assert_eq!(out, vec![(key(5), val(&[7, 9]))]); + let pair = vec![(one(5), env(7, &[])), (one(5), env(6, &[]))]; + let dep = vec![(one(5), env(7, &[9]))]; + let out = keyed_lookup::(&mut Mem, &dep, &side(pair, 0), 0, true); + assert_eq!(out, vec![(one(5), env(7, &[9]))]); } #[test] fn keyed_nonmin_demands_all_same_key_inputs() { - let pair = vec![(key(5), val(&[7])), (key(5), val(&[6]))]; - let dep = vec![(key(5), val(&[1, 9]))]; // V_out unrelated (a count) - let out = keyed_lookup::(&mut Mem, &dep, &side(pair, (1, 1), 0), (1, 1), 0, false); - assert_eq!(sorted(out), vec![(key(5), val(&[6, 9])), (key(5), val(&[7, 9]))]); + let pair = vec![(one(5), env(7, &[])), (one(5), env(6, &[]))]; + let dep = vec![(one(5), env(1, &[9]))]; // V_out unrelated (a count) + let out = keyed_lookup::(&mut Mem, &dep, &side(pair, 0), 0, false); + assert_eq!(sorted(out), vec![(one(5), env(6, &[9])), (one(5), env(7, &[9]))]); } #[test] fn lossy_via_fallback_recovers_input() { - // proj: K_out = $1 (V_in), V_out = $0 (K_in). - let p = proj(spread(1), spread(0)); - let pair = vec![(key(3), val(&[8]))]; // (K_in=3 ; V_in=8) - let dep = vec![(key(8), val(&[3, 9]))]; // (K_out=8 ; V_out=3, q=9) - let out = lossy_lookup::(&mut Mem, &dep, &side(pair, (1, 1), 0), (1, 1), 0, &p); - assert_eq!(out, vec![(key(3), val(&[8, 9]))]); + // proj: K_out = V_in, V_out = K_in. + let p = proj(user(1), user(0)); + let pair = vec![(one(3), env(8, &[]))]; // (K_in=3 ; V_in=8) + let dep = vec![(one(8), env(3, &[9]))]; // (K_out=8 ; V_out=3, q=9) + let out = lossy_lookup::(&mut Mem, &dep, &side(pair, 0), 0, &p); + assert_eq!(out, vec![(one(3), env(8, &[9]))]); } #[test] fn join_demands_both_inputs() { - // proj: K_out = $0 (K), V_out = $1 (V_L). - let p = proj(spread(0), spread(1)); - let left = side(vec![(key(5), val(&[7]))], (1, 1), 0); - let right = side(vec![(key(5), val(&[9]))], (1, 1), 0); - let dep = vec![(key(5), val(&[7, 1]))]; // (K_out=5 ; V_out=7, q=1) - let (lc, rc) = join_lookup::(&mut Mem, &dep, &left, &right, (1, 1), 0, &p); - assert_eq!(lc, vec![(key(5), val(&[7, 1]))]); - assert_eq!(rc, vec![(key(5), val(&[9, 1]))]); + // proj: K_out = K, V_out = V_L. + let p = proj(user(0), user(1)); + let left = side(vec![(one(5), env(7, &[]))], 0); + let right = side(vec![(one(5), env(9, &[]))], 0); + let dep = vec![(one(5), env(7, &[1]))]; // (K_out=5 ; V_out=7, q=1) + let (lc, rc) = join_lookup::(&mut Mem, &dep, &left, &right, 0, &p); + assert_eq!(lc, vec![(one(5), env(7, &[1]))]); + assert_eq!(rc, vec![(one(5), env(9, &[1]))]); } } diff --git a/interactive/src/explain/mod.rs b/interactive/src/explain/mod.rs index 46387ee33..9a702aaa2 100644 --- a/interactive/src/explain/mod.rs +++ b/interactive/src/explain/mod.rs @@ -11,7 +11,8 @@ //! program's scopes into an output scope under construction; every nested //! scope additionally `lift_iter`s and exports each internal collection, so //! every value-producing site in the subtree has a "host-visible" form — its -//! user-iter coordinates folded into the value, innermost first — at the +//! value as `(V, chain…)`, the user's value whole and then its user-iter +//! coordinates, innermost first — at the //! embedding level. //! //! A scope exports its lifted internals, and the cascade through enclosing @@ -136,9 +137,19 @@ fn clone_rec(orig: &Scope, out: &mut Scope, import_map: &[Ref], path: &[usize]) // Lift each subtree site at this scope's exit and export it: // the lift folds this scope's iteration coordinate into the // value (innermost coordinates were folded by deeper exits), - // and the export is the host-visible edge upward. + // and the export is the host-visible edge upward. A site of + // this scope first has its value wrapped as `(V)`, so the + // lifts extend it to `(V, chain…)` with `V` whole. let sub_idx = out.items.len(); for (a, vref) in child_visible { + let vref = if a.path == child_path { + let wrap_idx = child_out.items.len(); + let wrap = Projection { key: Term::Var(0), val: Term::Tuple(vec![Term::Var(1)]) }; + child_out.items.push(Item::Op(Node::Linear { input: vref, ops: vec![LinearOp::Project(wrap)] })); + Ref::Local(wrap_idx) + } else { + vref + }; let lift_idx = child_out.items.len(); child_out.items.push(Item::Op(Node::Linear { input: vref, ops: vec![LinearOp::LiftIter] })); let export_idx = child_out.exports.len(); @@ -235,111 +246,53 @@ fn resolve(root: &Scope, path: &[usize], r: &Ref) -> Target { } } -// ===== Term-arity: how a projection reshapes a flat `(k, v)` row ===== +// ===== User terms over explain's rows ===== // -// The demand envelope is a flat value tuple `[V (v) | chain | q]`, so explain -// needs each site's `v` to know where the chain starts. A projection side is a -// `Term::Tuple` of fields; a bare-row `Spread($r)` field splices row r's `v` -// fields, every other field is one value. (Explain targets flat relational -// programs — tuples of scalar fields.) -fn field_width(t: &Term, rows: &[usize]) -> usize { +// Explain reuses the user's terms (a filter's predicate, a projection, a +// flatmap's list) over its own rows, whose values carry the chain (and `q`) +// after the user's value. A term must see the user's row, not that one, so +// each `$r` is replaced by the user's row as it sits in explain's. + +/// `t` with each input row `$r` replaced by `rows[r]`. +fn with_rows(t: &Term, rows: &[Term]) -> Term { + let go = |t: &Term| with_rows(t, rows); + let bx = |t: &Term| Box::new(with_rows(t, rows)); match t { - Term::Spread(inner) => match &**inner { - Term::Var(r) => rows.get(*r).copied().unwrap_or(0), - _ => 1, + Term::Var(r) => rows[*r].clone(), + Term::Bound(_) | Term::Int(_) => t.clone(), + Term::Tuple(ts) => Term::Tuple(ts.iter().map(go).collect()), + Term::List(ts) => Term::List(ts.iter().map(go).collect()), + Term::Hash(ts) => Term::Hash(ts.iter().map(go).collect()), + Term::Call(name, ts) => Term::Call(name.clone(), ts.iter().map(go).collect()), + Term::Spread(t) => Term::Spread(bx(t)), + Term::Proj(t, i) => Term::Proj(bx(t), *i), + Term::Unary(op, t) => Term::Unary(*op, bx(t)), + Term::Binary(op, a, b) => Term::Binary(*op, bx(a), bx(b)), + Term::Inject { tag, payload, sum } => Term::Inject { tag: bx(tag), payload: bx(payload), sum: sum.clone() }, + Term::Case { scrutinee, arms, default } => Term::Case { + scrutinee: bx(scrutinee), + arms: arms.iter().map(go).collect(), + default: default.as_deref().map(bx), }, - _ => 1, + Term::Fold { list, init, step } => Term::Fold { list: bx(list), init: bx(init), step: bx(step) }, + Term::If { cond, then, els } => Term::If { cond: bx(cond), then: bx(then), els: bx(els) }, } } -fn proj_arity(t: &Term, rows: &[usize]) -> usize { - match t { Term::Tuple(fields) => fields.iter().map(|f| field_width(f, rows)).sum(), _ => 1 } -} -fn apply_ops_arity((mut k, mut v): (usize, usize), ops: &[LinearOp]) -> (usize, usize) { - for op in ops { - match op { - LinearOp::Project(p) => { let rows = [k, v]; k = proj_arity(&p.key, &rows); v = proj_arity(&p.val, &rows); } - LinearOp::Filter(_) | LinearOp::Negate | LinearOp::EnterAt(_) => {} - LinearOp::LiftIter => v += 1, - LinearOp::FlatMap(_) => v = 2, // value becomes tuple(pos, element) - } - } - (k, v) -} -/// `(k, v)` per site, forward-propagated from source arities to a fixed point. -fn site_shapes( - p: &Program, - source_shapes: &[(usize, usize)], -) -> BTreeMap { - let mut shapes: BTreeMap = BTreeMap::new(); - loop { - let before = shapes.len(); - walk_shapes(&p.root, &p.root, &[], source_shapes, &mut shapes); - if shapes.len() == before { break; } - } - shapes +/// `p` over rows whose input rows are `rows` (see [`with_rows`]). +fn proj_with_rows(p: &Projection, rows: &[Term]) -> Projection { + Projection { key: with_rows(&p.key, rows), val: with_rows(&p.val, rows) } } -fn shape_of_ref( - root: &Scope, - path: &[usize], - r: &Ref, - source_shapes: &[(usize, usize)], - shapes: &BTreeMap, -) -> Option<(usize, usize)> { - match resolve(root, path, r) { - Target::Source(k) => Some(source_shapes[k]), - Target::Site(a) => shapes.get(&a).copied(), - } +/// The user's rows in an explain row whose slots after the key are envelope +/// values: the key whole, and each value the envelope's first field. +fn user_rows(slots: usize) -> Vec { + std::iter::once(var(0)).chain((1..slots).map(|s| f(s, 0))).collect() } -fn walk_shapes( - root: &Scope, - s: &Scope, - path: &[usize], - source_shapes: &[(usize, usize)], - shapes: &mut BTreeMap, -) { - let addr = |site: Site| Addr { path: path.to_vec(), site }; - for b in &s.binds { - if !shapes.contains_key(&addr(Site::Var(b.var))) { - if let Some(sh) = shape_of_ref(root, path, &b.value, source_shapes, shapes) { - shapes.insert(addr(Site::Var(b.var)), sh); - } - } - } - for (i, item) in s.items.iter().enumerate() { - match item { - Item::Op(node) => { - if shapes.contains_key(&addr(Site::Op(i))) { continue; } - let of = |r: &Ref| shape_of_ref(root, path, r, source_shapes, shapes); - let sh = match node { - Node::Linear { input, ops } => of(input).map(|s| apply_ops_arity(s, ops)), - Node::Concat(refs) => refs.iter().find_map(|r| of(r)), - Node::Arrange(r) | Node::Inspect { input: r, .. } => of(r), - Node::Join { left, right, projection } => match (of(left), of(right)) { - (Some((kl, vl)), Some((_, vr))) => { - let rows = [kl, vl, vr]; - Some((proj_arity(&projection.key, &rows), proj_arity(&projection.val, &rows))) - } - _ => None, - }, - Node::Reduce { input, reducer } => of(input).map(|(k, v)| match reducer { - Reducer::Distinct => (k, 0), - Reducer::Min => (k, v), - Reducer::Count => (k, 1), - Reducer::Collect => (k, 1), - }), - }; - if let Some(sh) = sh { shapes.insert(addr(Site::Op(i)), sh); } - } - Item::Sub(child) => { - let mut cp = path.to_vec(); - cp.push(i); - walk_shapes(root, child, &cp, source_shapes, shapes); - } - } - } +/// A user row `(K ; V)` as an envelope row `(K ; (V))`, with an empty chain. +fn envelope() -> Projection { + Projection { key: var(0), val: tup(vec![var(1)]) } } /// Options for the explanation rewrite. @@ -403,41 +356,33 @@ impl Sb { self.op(Node::Inspect { input, label }); } } - /// Semijoin `left (K; V)` with `right (K; V)` by `(K)`, keep left's rows. - fn semijoin_data(&mut self, left: Ref, right: Ref, k_arity: usize, v_arity: usize) -> Ref { - self.join(left, right, ::project_kv(k_arity, v_arity)) + /// Semijoin `left (K ; V)` with `right` by `K`, keeping left's rows. + fn semijoin(&mut self, left: Ref, right: Ref) -> Ref { + self.join(left, right, Projection { key: var(0), val: var(1) }) } - /// Set-level distinct on whole `(K ; V ++ chain ++ q)` rows. `v_arity` is - /// the *full* value width (V + chain + q), so `pack` moves the entire row - /// into the join key and `distinct_unpack` restores it — dedup on identical - /// demand rows, not on `(K, V)`. - fn distinct_full(&mut self, input: Ref, k_arity: usize, v_arity: usize) -> Ref { - let packed = self.project(input, ::pack(k_arity, v_arity, 0, false)); + /// Set-level distinct on whole rows: move the row into the key, distinct, + /// and move it back. + fn distinct_full(&mut self, input: Ref) -> Ref { + let packed = self.project(input, Projection { key: tup(vec![var(0), var(1)]), val: tup(Vec::new()) }); let dist = self.reduce(packed, Reducer::Distinct); - self.project(dist, ::distinct_unpack(k_arity, v_arity)) + self.project(dist, Projection { key: f(0, 0), val: f(0, 1) }) } } /// One upstream edge into a backward rule: the target's host-side `(data, -/// user)` collections from both clones, its shape, and its user-chain length. +/// user)` collections from both clones, and its user-chain length. struct Side { witness: Ref, forward: Ref, - shape: (usize, usize), user_len: usize, } -fn strip_user_and_q(k_arity: usize, v_arity: usize) -> Projection { - ::project_kv(k_arity, v_arity) -} - impl Side { /// View this edge as the data-model-agnostic `SideInfo` the reverse rules take. fn info(&self) -> decouple::SideInfo { decouple::SideInfo { witness: self.witness.clone(), forward: self.forward.clone(), - shape: self.shape, user_len: self.user_len, } } @@ -453,21 +398,19 @@ impl Sb { } /// Keyed lookup (Reduce-style), with Min's value-narrowing. - fn emit_lookup_keyed(&mut self, dep_y: Ref, side: &Side, output_shape: (usize, usize), out_user_len: usize, reducer: &Reducer) -> Ref { + fn emit_lookup_keyed(&mut self, dep_y: Ref, side: &Side, out_user_len: usize, reducer: &Reducer) -> Ref { let min = matches!(reducer, Reducer::Min); - decouple::keyed_lookup::(self, &dep_y, &side.info(), output_shape, out_user_len, min) + decouple::keyed_lookup::(self, &dep_y, &side.info(), out_user_len, min) } /// Lossy lookup (Linear[Project]), through the pair table. - fn emit_lookup_lossy(&mut self, dep_y: Ref, side: &Side, output_shape: (usize, usize), out_user_len: usize, proj: &Projection) -> Ref { - decouple::lossy_lookup::(self, &dep_y, &side.info(), output_shape, out_user_len, proj) + fn emit_lookup_lossy(&mut self, dep_y: Ref, side: &Side, out_user_len: usize, proj: &Projection) -> Ref { + decouple::lossy_lookup::(self, &dep_y, &side.info(), out_user_len, proj) } /// Join's backward rule: two contribs (left, right). - fn emit_lookup_join(&mut self, dep_y: Ref, left: &Side, right: &Side, output_shape: (usize, usize), out_user_len: usize, projection: &Projection) -> (Ref, Ref) { - decouple::join_lookup::( - self, &dep_y, &left.info(), &right.info(), output_shape, out_user_len, projection, - ) + fn emit_lookup_join(&mut self, dep_y: Ref, left: &Side, right: &Side, out_user_len: usize, projection: &Projection) -> (Ref, Ref) { + decouple::join_lookup::(self, &dep_y, &left.info(), &right.info(), out_user_len, projection) } /// FlatMap's (UNNEST's) backward rule — the universal backstop instantiated @@ -480,42 +423,40 @@ impl Sb { /// The one wrinkle is that a plain flatmap drops the source row and the key /// isn't unique, so we **re-key the input by itself** before exploding (the /// source then rides through in the join key) and re-project to the packed - /// output afterwards. `output_shape.1 == 2` (the `tuple(pos, elem)` value). - fn emit_lookup_flatmap(&mut self, dep_y: Ref, side: &Side, output_shape: (usize, usize), out_user_len: usize, list_term: &Term) -> Ref { - let (_, v_in) = side.shape; - let v_out = output_shape.1; // 2: pos, elem + /// output afterwards. `list_term` reads the re-keyed rows: the input's key + /// is the key's first field, and its value is the row's value. + fn emit_lookup_flatmap(&mut self, dep_y: Ref, side: &Side, out_user_len: usize, list_term: &Term) -> Ref { let d = out_user_len; // same depth for input and output // Pair table: re-key the input by the whole input row, explode (the - // source survives in the key), then re-key to (k, pos, elem) carrying + // source survives in the key), then re-key to (k, (pos, elem)) carrying // the source as the value. let src = self.concat(vec![side.witness.clone(), side.forward.clone()]); - let rekeyed = self.project(src, Projection { key: tup(vec![var(0), var(1)]), val: var(1) }); + let rekeyed = self.project(src, Projection { key: tup(vec![var(0), var(1)]), val: f(1, 0) }); let exploded = self.linear(rekeyed, vec![LinearOp::FlatMap(list_term.clone())]); - let pair = self.project(exploded, Projection { key: tup(vec![f(0, 0), f(1, 0), f(1, 1)]), val: var(0) }); - // Re-key the demand by the same packed output; keep its chain ++ q. - let dep_keyed = self.project(dep_y, Projection { key: tup(vec![var(0), f(1, 0), f(1, 1)]), val: tup(fidx(1, v_out, v_out + d + 1)) }); + let pair = self.project(exploded, Projection { key: tup(vec![f(0, 0), var(1)]), val: var(0) }); + // Re-key the demand by the same output (k, (pos, elem)); keep its chain ++ q. + let dep_keyed = self.project(dep_y, Projection { key: tup(vec![var(0), f(1, 0)]), val: tup(fidx(1, 1, d + 2)) }); // Join into a wide row [V_in | chain_in | chain_out | q]. Source is `$2` - // = (k, host_val); host_val = [V_in | chain_in]. Dep' is `$1` = [chain_out | q]. - let mut wide = (0..v_in).map(|c| fld(f(2, 1), c)).collect::>(); // V_in - wide.extend((0..d).map(|c| fld(f(2, 1), v_in + c))); // chain_in - wide.extend(fidx(1, 0, d)); // chain_out - wide.push(f(1, d)); // q + // = (k, host_val); host_val = (V_in, chain_in). Dep' is `$1` = [chain_out | q]. + let mut wide = vec![fld(f(2, 1), 0)]; // V_in + wide.extend((0..d).map(|c| fld(f(2, 1), 1 + c))); // chain_in + wide.extend(fidx(1, 0, d)); // chain_out + wide.push(f(1, d)); // q let joined = self.join(dep_keyed, pair, Projection { key: f(2, 0), val: tup(wide) }); // Soundness: chain_in ≤ chain_out (same depth, so coordinate-aligned). - let cond = and_all((0..d).map(|c| le(f(1, v_in + c), f(1, v_in + d + c))).collect()); + let cond = and_all((0..d).map(|c| le(f(1, 1 + c), f(1, 1 + d + c))).collect()); let filtered = match cond { Some(c) => self.filter(joined, c), None => joined }; - // Reform the demanded input (k ; V_in ++ chain_out ++ q). - let mut val = fidx(1, 0, v_in); - val.extend(fidx(1, v_in + d, v_in + 2 * d)); - val.push(f(1, v_in + 2 * d)); + // Reform the demanded input (k ; V_in, chain_out, q). + let mut val = vec![f(1, 0)]; + val.extend(fidx(1, 1 + d, 2 + 2 * d)); self.project(filtered, Projection { key: var(0), val: tup(val) }) } } -/// The `Value` data model for explain: a demand row is `(K ; Tuple([V…, chain…, -/// q]))` — a flat value tuple of `V`'s fields, then the loop-iteration chain +/// The `Value` data model for explain: a demand row is `(K ; (V, chain…, q))` +/// — the user's key and value whole, then the loop-iteration chain /// (innermost-first), then the trailing query id, matching the host lift's -/// `append_iter`. Every builder works in field-index ranges. +/// `append_iter` on `(V)`. Every builder works in field positions of that tuple. pub(crate) struct Val; impl Dataflow for Sb { @@ -528,10 +469,8 @@ impl Dataflow for Sb { fn concat(&mut self, cs: Vec) -> Ref { Sb::concat(self, cs) } } -// Term-building helpers for the Value reverse rules. The demand envelope is a -// flat value tuple `[V (v fields) | chain (innermost-first) | q]`, matching the -// host lift's `append_iter`. Each projection side is built as a `Term::Tuple` -// of fields; `f(s, c)` selects field c of slot s, `fidx(s, lo, hi)` a range. +// Term-building helpers for the Value reverse rules. `f(s, c)` selects field c +// of slot s, `fidx(s, lo, hi)` a range. fn var(s: usize) -> Term { Term::Var(s) } fn fld(t: Term, i: usize) -> Term { Term::Proj(Box::new(t), i) } fn f(s: usize, i: usize) -> Term { fld(var(s), i) } @@ -545,226 +484,149 @@ fn and_all(conds: Vec) -> Option { conds.into_iter().reduce(|a, b| Term::Binary(BinOp::And, Box::new(a), Box::new(b))) } -/// Expand a user projection side over slots whose value-columns have the given -/// `arities`. A bare-row `Spread($r)` field splices only row r's value columns -/// (`arities[r]`), excluding any trailing chain; every other field passes -/// through unchanged (an indexed `$r[c]` reads a value column directly, as the -/// chain sits after the value). -fn expand_value_fields(side: &Term, arities: &[usize]) -> Vec { - let fields = match side { Term::Tuple(fs) => fs.as_slice(), other => std::slice::from_ref(other) }; - let mut out = Vec::new(); - for field in fields { - match field { - Term::Spread(inner) => match &**inner { - Term::Var(r) => out.extend(fidx(*r, 0, arities.get(*r).copied().unwrap_or(0))), - _ => out.push(field.clone()), - }, - other => out.push(other.clone()), - } - } - out -} - impl RowModel for Val { type Proj = Projection; type Pred = Term; - fn pack(k: usize, v: usize, chain_len: usize, has_q: bool) -> Projection { - let mut key = fidx(0, 0, k); key.extend(fidx(1, 0, v)); + fn pack(chain_len: usize, has_q: bool) -> Projection { let tail = chain_len + has_q as usize; - Projection { key: tup(key), val: tup(fidx(1, v, v + tail)) } - } - fn project_kv(k: usize, v: usize) -> Projection { - Projection { key: tup(fidx(0, 0, k)), val: tup(fidx(1, 0, v)) } - } - fn sp_reassemble(k: usize, v: usize, user_len: usize, out_depth: usize) -> Projection { - let key = fidx(0, 0, k); - let mut val = fidx(0, k, k + v); // V (from packed key) - val.extend(fidx(2, 0, user_len)); // chain_in (pair) - val.extend(fidx(1, 0, out_depth)); // chain_out (dep) - val.push(f(1, out_depth)); // q - Projection { key: tup(key), val: tup(val) } - } - - fn ky_join(k: usize, v_in: usize, v_out: usize, in_len: usize, out_len: usize, min: bool) -> Projection { - let mut val = fidx(2, 0, v_in); // V_in - if min { val.extend(fidx(1, 0, v_out)); } // V_out (for narrowing) - val.extend(fidx(2, v_in, v_in + in_len)); // chain_in - val.extend(fidx(1, v_out, v_out + out_len)); // chain_out - val.push(f(1, v_out + out_len)); // q - Projection { key: tup(fidx(0, 0, k)), val: tup(val) } - } - fn ky_data_eq(v_in: usize) -> Option { - and_all((0..v_in).map(|i| eq(f(1, i), f(1, v_in + i))).collect()) - } - fn ky_drop_vout(k: usize, v_in: usize, v_out: usize, in_len: usize, out_len: usize) -> Projection { - let after_vout = v_in + v_out; - let mut val = fidx(1, 0, v_in); - val.extend(fidx(1, after_vout, after_vout + in_len)); - val.extend(fidx(1, after_vout + in_len, after_vout + in_len + out_len)); - val.push(f(1, after_vout + in_len + out_len)); - Projection { key: tup(fidx(0, 0, k)), val: tup(val) } - } - - // Bounds Spread, so the pair table serves every projection. - fn lossy_pair(proj: &Projection, k_in: usize, v_in: usize, in_len: usize) -> Projection { - let mut val = fidx(0, 0, k_in); - val.extend(fidx(1, 0, v_in + in_len)); - Projection { key: tup(expand_value_fields(&proj.key, &[k_in, v_in])), val: tup(val) } - } - fn lossy_reassemble(k_in: usize, v_in: usize, v_out: usize, in_len: usize, out_len: usize) -> Projection { - let key = fidx(2, 0, k_in); - let mut val = fidx(2, k_in, k_in + v_in); - val.extend(fidx(2, k_in + v_in, k_in + v_in + in_len)); - val.extend(fidx(1, v_out, v_out + out_len)); - val.push(f(1, v_out + out_len)); - Projection { key: tup(key), val: tup(val) } - } - - fn join_forward(proj: &Projection, k: usize, v_l: usize, v_r: usize, l_len: usize, r_len: usize) -> Projection { - let key = expand_value_fields(&proj.key, &[k, v_l, v_r]); - let mut val = fidx(0, 0, k); // K - val.extend(fidx(1, 0, v_l)); // V_L - val.extend(fidx(2, 0, v_r)); // V_R - val.extend(fidx(1, v_l, v_l + l_len)); // chain_L - val.extend(fidx(2, v_r, v_r + r_len)); // chain_R - val.extend(expand_value_fields(&proj.val, &[k, v_l, v_r])); // V_out (pair-produced) - Projection { key: tup(key), val: tup(val) } - } - fn join_combined(k: usize, v_l: usize, v_r: usize, v_out: usize, l_len: usize, r_len: usize, out_len: usize) -> Projection { - let vl_s = k; - let vr_s = vl_s + v_l; - let ul_s = vr_s + v_r; + Projection { key: tup(vec![var(0), f(1, 0)]), val: tup(fidx(1, 1, 1 + tail)) } + } + fn strip_all() -> Projection { + Projection { key: var(0), val: f(1, 0) } + } + fn sp_reassemble(in_len: usize, out_len: usize) -> Projection { + let mut val = vec![f(0, 1)]; // V (from packed key) + val.extend(fidx(2, 0, in_len)); // chain_in (pair) + val.extend(fidx(1, 0, out_len)); // chain_out (dep) + val.push(f(1, out_len)); // q + Projection { key: f(0, 0), val: tup(val) } + } + + fn ky_join(in_len: usize, out_len: usize, min: bool) -> Projection { + let mut val = vec![f(2, 0)]; // V_in + if min { val.push(f(1, 0)); } // V_out (for narrowing) + val.extend(fidx(2, 1, 1 + in_len)); // chain_in + val.extend(fidx(1, 1, 1 + out_len)); // chain_out + val.push(f(1, 1 + out_len)); // q + Projection { key: var(0), val: tup(val) } + } + fn ky_data_eq() -> Option { + Some(eq(f(1, 0), f(1, 1))) + } + fn ky_drop_vout(in_len: usize, out_len: usize) -> Projection { + let mut val = vec![f(1, 0)]; + val.extend(fidx(1, 2, 3 + in_len + out_len)); // chain_in, chain_out, q + Projection { key: var(0), val: tup(val) } + } + + fn lossy_pair(proj: &Projection, in_len: usize) -> Projection { + let mut val = vec![var(0), f(1, 0)]; // K_in, V_in + val.extend(fidx(1, 1, 1 + in_len)); // chain_in + Projection { key: proj.key.clone(), val: tup(val) } + } + fn lossy_reassemble(in_len: usize, out_len: usize) -> Projection { + let mut val = vec![f(2, 1)]; // V_in + val.extend(fidx(2, 2, 2 + in_len)); // chain_in + val.extend(fidx(1, 1, 1 + out_len)); // chain_out + val.push(f(1, 1 + out_len)); // q + Projection { key: f(2, 0), val: tup(val) } + } + + fn join_forward(proj: &Projection, l_len: usize, r_len: usize) -> Projection { + let mut val = vec![var(0), f(1, 0), f(2, 0)]; // K, V_L, V_R + val.extend(fidx(1, 1, 1 + l_len)); // chain_L + val.extend(fidx(2, 1, 1 + r_len)); // chain_R + val.push(proj.val.clone()); // V_out (pair-produced) + Projection { key: proj.key.clone(), val: tup(val) } + } + fn join_combined(l_len: usize, r_len: usize, out_len: usize) -> Projection { + let ul_s = 3; let ur_s = ul_s + l_len; let vout_s = ur_s + r_len; - let key = fidx(2, 0, k); - let mut val = fidx(2, vl_s, vl_s + v_l); // V_L - val.extend(fidx(2, vr_s, vr_s + v_r)); // V_R - val.extend(fidx(2, ul_s, ul_s + l_len)); // chain_L - val.extend(fidx(2, ur_s, ur_s + r_len)); // chain_R - val.extend(fidx(1, v_out, v_out + out_len)); // chain_out - val.push(f(1, v_out + out_len)); // q - val.extend(fidx(1, 0, v_out)); // V_out_dep - val.extend(fidx(2, vout_s, vout_s + v_out)); // V_out_pair - Projection { key: tup(key), val: tup(val) } - } - fn join_filter(v_l: usize, v_r: usize, v_out: usize, l_len: usize, r_len: usize, out_len: usize) -> Option { - let ul_j = v_l + v_r; + let mut val = vec![f(2, 1), f(2, 2)]; // V_L, V_R + val.extend(fidx(2, ul_s, ul_s + l_len)); // chain_L + val.extend(fidx(2, ur_s, ur_s + r_len)); // chain_R + val.extend(fidx(1, 1, 1 + out_len)); // chain_out + val.push(f(1, 1 + out_len)); // q + val.push(f(1, 0)); // V_out_dep + val.push(f(2, vout_s)); // V_out_pair + Projection { key: f(2, 0), val: tup(val) } + } + fn join_filter(l_len: usize, r_len: usize, out_len: usize) -> Option { + let ul_j = 2; let ur_j = ul_j + l_len; let uo_j = ur_j + r_len; let q_j = uo_j + out_len; - let vdep_j = q_j + 1; - let vpair_j = vdep_j + v_out; let time_cond = |off: usize, in_len: usize| -> Option { let n = in_len.min(out_len); let in_skip = in_len - n; let out_skip = out_len - n; and_all((0..n).map(|i| le(f(1, off + in_skip + i), f(1, uo_j + out_skip + i))).collect()) }; - let value_cond = and_all((0..v_out).map(|i| eq(f(1, vdep_j + i), f(1, vpair_j + i))).collect()); + let value_cond = Some(eq(f(1, q_j + 1), f(1, q_j + 2))); [time_cond(ul_j, l_len), time_cond(ur_j, r_len), value_cond] .into_iter().flatten().reduce(|a, b| Term::Binary(BinOp::And, Box::new(a), Box::new(b))) } - fn join_split(left: bool, k: usize, v_l: usize, v_r: usize, l_len: usize, r_len: usize, out_len: usize) -> Projection { - let ul_j = v_l + v_r; + fn join_split(left: bool, l_len: usize, r_len: usize, out_len: usize) -> Projection { + let ul_j = 2; let ur_j = ul_j + l_len; - let uo_j = ur_j + r_len; - let q_j = uo_j + out_len; - let key = fidx(0, 0, k); - let val = if left { - let mut v = fidx(1, 0, v_l); // V_L - v.extend(fidx(1, ul_j, ul_j + l_len)); // chain_L - v.push(f(1, q_j)); // q - v + let q_j = ur_j + r_len + out_len; + let mut val = if left { vec![f(1, 0)] } else { vec![f(1, 1)] }; // V_L or V_R + if left { + val.extend(fidx(1, ul_j, ul_j + l_len)); // chain_L } else { - let mut v = fidx(1, v_l, v_l + v_r); // V_R - v.extend(fidx(1, ur_j, ur_j + r_len)); // chain_R - v.push(f(1, q_j)); // q - v - }; - Projection { key: tup(key), val: tup(val) } + val.extend(fidx(1, ur_j, ur_j + r_len)); // chain_R + } + val.push(f(1, q_j)); // q + Projection { key: var(0), val: tup(val) } } - fn time_le(v: usize, in_len: usize, out_len: usize) -> Option { - // Joined row value: [V_pre(v) | user_in(in_len) | user_out(out_len) | q], - // each chain innermost-first; compare outer-aligned (last min(in,out)). + fn time_le(in_len: usize, out_len: usize) -> Option { + // Joined row value: (V, user_in(in_len), user_out(out_len), q), each + // chain innermost-first; compare outer-aligned (last min(in,out)). let n = in_len.min(out_len); let in_off = in_len - n; let out_off = out_len - n; - and_all((0..n).map(|i| le(f(1, v + in_off + i), f(1, v + in_len + out_off + i))).collect()) + and_all((0..n).map(|i| le(f(1, 1 + in_off + i), f(1, 1 + in_len + out_off + i))).collect()) } - fn strip(k: usize, v: usize, in_len: usize, out_len: usize, keep: usize) -> Projection { - let key = fidx(0, 0, k); - let mut val = fidx(1, 0, v); // V_pre + fn strip(in_len: usize, out_len: usize, keep: usize) -> Projection { + let mut val = vec![f(1, 0)]; // V let drop_in = in_len - keep; - for i in 0..keep { val.push(f(1, v + drop_in + i)); } // outer kept user_in coords - val.push(f(1, v + in_len + out_len)); // q - Projection { key: tup(key), val: tup(val) } + for i in 0..keep { val.push(f(1, 1 + drop_in + i)); } // outer kept user_in coords + val.push(f(1, 1 + in_len + out_len)); // q + Projection { key: var(0), val: tup(val) } } - fn bind_filter(v: usize) -> Option { - Some(gt(f(1, v), Term::Int(0))) + fn bind_filter() -> Option { + Some(gt(f(1, 1), Term::Int(0))) } - fn bind_decrement(k: usize, v: usize, user_len: usize) -> Projection { - let mut val = fidx(1, 0, v); - val.push(sub(f(1, v), Term::Int(1))); // innermost chain coord - 1 - val.extend(fidx(1, v + 1, v + user_len)); - val.push(f(1, v + user_len)); // q - Projection { key: tup(fidx(0, 0, k)), val: tup(val) } - } - fn distinct_unpack(k: usize, v: usize) -> Projection { - Projection { key: tup(fidx(0, 0, k)), val: tup(fidx(0, k, k + v)) } + fn bind_decrement(user_len: usize) -> Projection { + let mut val = vec![f(1, 0)]; + val.push(sub(f(1, 1), Term::Int(1))); // innermost chain coord - 1 + val.extend(fidx(1, 2, 2 + user_len)); // the rest of the chain, then q + Projection { key: var(0), val: tup(val) } } } -/// The `(k, v)` shape of the program's first export — what a query against it -/// must match. A query row is `(key[k]; val[v] ++ [q])`; a shape-mismatched -/// query addresses nothing and yields junk demand, so harnesses should check -/// loudly at seeding time. -pub fn export_shape(p: &Program, source_shapes: &[(usize, usize)]) -> (usize, usize) { - let shapes = site_shapes(p, source_shapes); - let first = p.root.exports.first().expect("export_shape: program has no export").value.clone(); - match resolve(&p.root, &[], &first) { - Target::Site(a) => shapes[&a], - Target::Source(k) => source_shapes[k], - } -} - -/// The `(k, v)` field counts of each root import: from its declared shape where it has one, -/// and otherwise `default`. A product counts its fields, a unit none, any other shape one. -pub fn source_arities(p: &Program, default: (usize, usize)) -> Vec<(usize, usize)> { - fn fields(s: &corgi::Shape) -> usize { - match s { corgi::Shape::Prod(fs) => fs.len(), corgi::Shape::Unit => 0, _ => 1 } - } - p.root.imports.iter().map(|imp| imp.shape.as_ref().map_or(default, |(k, v)| (fields(k), fields(v)))).collect() -} - -/// The shape of a query row against `p`'s first export, `(key ; val ++ [q])`, when the -/// export's shape follows from `p`'s declared sources. +/// The shape of a query row against `p`'s first export, `(key ; (val, q))`, +/// when the export's shape follows from `p`'s declared sources. fn query_shape(p: &Program) -> Option<(corgi::Shape, corgi::Shape)> { - use corgi::Shape; let (shapes, _) = crate::shapes::infer(p, &|_| None); let export = shapes.exports.first()?.clone()?; - let val = match export.val { - Shape::Prod(mut fs) => { fs.push(Shape::Int); Shape::Prod(fs) } - Shape::Unit => Shape::Prod(vec![Shape::Int]), - other => Shape::Prod(vec![other, Shape::Int]), - }; - Some((export.key, val)) + Some((export.key, corgi::Shape::Prod(vec![export.val, corgi::Shape::Int]))) } -/// The transform. `source_shapes[k]` is the `(k, v)` of the original root's -/// import `k` (positional inputs and named traces alike). The query arrives -/// as one extra positional input appended after the original inputs. -pub fn explain(p: &Program, source_shapes: &[(usize, usize)]) -> Program { - explain_with(p, source_shapes, Options::default()) +/// The transform. The query arrives as one extra positional input appended +/// after the original inputs, its rows `(key ; (val, q))`: a row of the first +/// export, and a query id. +pub fn explain(p: &Program) -> Program { + explain_with(p, Options::default()) } /// [`explain`], with [`Options`]. -pub fn explain_with(p: &Program, source_shapes: &[(usize, usize)], options: Options) -> Program { +pub fn explain_with(p: &Program, options: Options) -> Program { let n_sources = p.root.imports.len(); - assert_eq!(source_shapes.len(), n_sources, "one shape per root import"); - let shapes = site_shapes(p, source_shapes); // The largest positional-input index, for placing the query input after. let max_input = p.root.imports.iter().filter_map(|imp| match &imp.from { Source::Input(i) => Some(*i + 1), @@ -783,24 +645,31 @@ pub fn explain_with(p: &Program, source_shapes: &[(usize, usize)], options: Opti root.s.exports.clear(); // ---- explain scope ---- + // Root-level host forms and the sources are the user's rows, which enter + // the envelope here; a nested host form was lifted into it on export. let mut ex = Sb::new("explain"); ex.debug = options.debug_inspects; let ex_src: Vec = (0..n_sources) .map(|k| ex.import(format!("$src:{}", k), Source::Parent(src_refs[k].clone()))) .collect(); + let src_env: Vec = ex_src.iter().map(|r| ex.project(r.clone(), envelope())).collect(); let ex_query = ex.import("$query".into(), Source::Parent(query_ref)); let wit: BTreeMap = witness.iter() - .map(|(a, r)| (a.clone(), ex.import(format!("$wit:{:?}:{:?}", a.path, a.site), Source::Parent(r.clone())))) + .map(|(a, r)| { + let imported = ex.import(format!("$wit:{:?}:{:?}", a.path, a.site), Source::Parent(r.clone())); + let host = if a.path.is_empty() { ex.project(imported, envelope()) } else { imported }; + (a.clone(), host) + }) .collect(); - // Demand-set variables, one per source; forward inputs are the actual - // sources restricted to the demand-sets. + // Demand-set variables, one per source, of the user's rows; forward inputs + // are the actual sources restricted to the demand-sets. let dsets: Vec = (0..n_sources).map(|k| ex.variable(&format!("demand_set_{}", k))).collect(); - let fwd_inputs: Vec = (0..n_sources).map(|k| { - let (ka, va) = source_shapes[k]; - ex.semijoin_data(dsets[k].clone(), ex_src[k].clone(), ka, va) - }).collect(); - let forward: BTreeMap = clone_into(&p.root, &mut ex.s, &fwd_inputs).into_iter().collect(); + let fwd_inputs: Vec = (0..n_sources).map(|k| ex.semijoin(dsets[k].clone(), ex_src[k].clone())).collect(); + let fwd_env: Vec = fwd_inputs.iter().map(|r| ex.project(r.clone(), envelope())).collect(); + let forward: BTreeMap = clone_into(&p.root, &mut ex.s, &fwd_inputs).into_iter() + .map(|(a, r)| { let host = if a.path.is_empty() { ex.project(r, envelope()) } else { r }; (a, host) }) + .collect(); ex.s.exports.clear(); // Pre-allocated demand variables for the original feedback vars (forward @@ -816,12 +685,10 @@ pub fn explain_with(p: &Program, source_shapes: &[(usize, usize)], options: Opti // The reverse walk. let mut rev = Reverse { orig: &p.root, - shapes: &shapes, - source_shapes, wit: &wit, fwd: &forward, - ex_src: &ex_src, - fwd_inputs: &fwd_inputs, + src_env: &src_env, + fwd_env: &fwd_env, demand, contribs: BTreeMap::new(), }; @@ -836,17 +703,16 @@ pub fn explain_with(p: &Program, source_shapes: &[(usize, usize)], options: Opti // accumulate, and export. let mut demand_exports: Vec<(String, usize)> = Vec::new(); for k in 0..n_sources { - let (ka, va) = source_shapes[k]; let cs = rev.contribs.remove(&Target::Source(k)).unwrap_or_default(); let combined = if cs.is_empty() { dsets[k].clone() // no demand: the set stays empty (self-bind below) } else { let merged = ex.concat(cs); - let stripped = ex.project(merged, strip_user_and_q(ka, va)); - let semi = ex.semijoin_data(stripped, ex_src[k].clone(), ka, va); + let stripped = ex.project(merged, ::strip_all()); + let semi = ex.semijoin(stripped, ex_src[k].clone()); ex.concat(vec![dsets[k].clone(), semi]) }; - let dist = ex.distinct_full(combined, ka, va); + let dist = ex.distinct_full(combined); ex.debug_inspect(dist.clone(), format!("demand_set:{}", k)); ex.bind(dsets[k].clone(), dist.clone()); let name = match &p.root.imports[k].from { @@ -884,12 +750,10 @@ fn collect_var_addrs(s: &Scope, path: &[usize], out: &mut Vec) { /// State for the reverse walk: demand refs per site, contributions per target. struct Reverse<'a> { orig: &'a Scope, - shapes: &'a BTreeMap, - source_shapes: &'a [(usize, usize)], wit: &'a BTreeMap, fwd: &'a BTreeMap, - ex_src: &'a [Ref], - fwd_inputs: &'a [Ref], + src_env: &'a [Ref], + fwd_env: &'a [Ref], demand: BTreeMap, contribs: BTreeMap>, } @@ -900,13 +764,11 @@ impl<'a> Reverse<'a> { Target::Site(a) => Side { witness: self.wit[a].clone(), forward: self.fwd[a].clone(), - shape: self.shapes[a], user_len: a.path.len(), }, Target::Source(k) => Side { - witness: self.ex_src[*k].clone(), - forward: self.fwd_inputs[*k].clone(), - shape: self.source_shapes[*k], + witness: self.src_env[*k].clone(), + forward: self.fwd_env[*k].clone(), user_len: 0, }, } @@ -938,13 +800,12 @@ impl<'a> Reverse<'a> { for b in &s.binds { let var_addr = Addr { path: path.to_vec(), site: Site::Var(b.var) }; let dv = self.demand[&var_addr].clone(); - let (kx, vx) = self.shapes[&var_addr]; let var_user_len = path.len(); - let filtered = match ::bind_filter(vx) { + let filtered = match ::bind_filter() { Some(c) => ex.filter(dv, c), None => dv, }; - let contrib = ex.project(filtered, ::bind_decrement(kx, vx, var_user_len)); + let contrib = ex.project(filtered, ::bind_decrement(var_user_len)); self.push(ex, path, &b.value, contrib, var_user_len); } // Items in reverse: consumers have contributed by the time we arrive. @@ -969,8 +830,7 @@ impl<'a> Reverse<'a> { continue; } let combined = ex.concat(cs); - let (k, vx) = self.shapes[&addr]; - let dist = ex.distinct_full(combined, k, vx + path.len() + 1); + let dist = ex.distinct_full(combined); ex.debug_inspect(dist.clone(), format!("demand_{:?}:{:?}", path, Site::Var(v))); ex.bind(dv, dist); } @@ -981,12 +841,10 @@ impl<'a> Reverse<'a> { let cs = self.contribs.remove(&Target::Site(addr.clone())).unwrap_or_default(); if cs.is_empty() { return; } let combined = ex.concat(cs); - let (k, v) = self.shapes[&addr]; let out_user_len = path.len(); - let dist = ex.distinct_full(combined, k, v + out_user_len + 1); + let dist = ex.distinct_full(combined); ex.debug_inspect(dist.clone(), format!("demand_{:?}:{:?}", path, Site::Op(i))); self.demand.insert(addr.clone(), dist.clone()); - let out_shape = (k, v); let dep_this = dist; match node { @@ -999,11 +857,12 @@ impl<'a> Reverse<'a> { LinearOp::Project(proj) => { let target = resolve(self.orig, path, input); let side = self.side(&target); - let contrib = ex.emit_lookup_lossy(dep_this, &side, out_shape, out_user_len, proj); + let proj = proj_with_rows(proj, &user_rows(2)); + let contrib = ex.emit_lookup_lossy(dep_this, &side, out_user_len, &proj); self.contribs.entry(target).or_default().push(contrib); } LinearOp::Filter(cond) => { - let contrib = ex.filter(dep_this, cond.clone()); + let contrib = ex.filter(dep_this, with_rows(cond, &user_rows(2))); self.push(ex, path, input, contrib, out_user_len); } LinearOp::Negate | LinearOp::EnterAt(_) => { @@ -1016,7 +875,8 @@ impl<'a> Reverse<'a> { LinearOp::FlatMap(list_term) => { let target = resolve(self.orig, path, input); let side = self.side(&target); - let contrib = ex.emit_lookup_flatmap(dep_this, &side, out_shape, out_user_len, list_term); + let list_term = with_rows(list_term, &[f(0, 0), var(1)]); + let contrib = ex.emit_lookup_flatmap(dep_this, &side, out_user_len, &list_term); self.contribs.entry(target).or_default().push(contrib); } } @@ -1035,7 +895,7 @@ impl<'a> Reverse<'a> { Node::Reduce { input, reducer } => { let target = resolve(self.orig, path, input); let side = self.side(&target); - let contrib = ex.emit_lookup_keyed(dep_this, &side, out_shape, out_user_len, reducer); + let contrib = ex.emit_lookup_keyed(dep_this, &side, out_user_len, reducer); self.contribs.entry(target).or_default().push(contrib); } Node::Join { left, right, projection } => { @@ -1043,7 +903,8 @@ impl<'a> Reverse<'a> { let rt = resolve(self.orig, path, right); let ls = self.side(<); let rs = self.side(&rt); - let (lc, rc) = ex.emit_lookup_join(dep_this, &ls, &rs, out_shape, out_user_len, projection); + let projection = proj_with_rows(projection, &user_rows(3)); + let (lc, rc) = ex.emit_lookup_join(dep_this, &ls, &rs, out_user_len, &projection); self.contribs.entry(lt).or_default().push(lc); self.contribs.entry(rt).or_default().push(rc); } diff --git a/interactive/tests/explain.rs b/interactive/tests/explain.rs index fe36709a7..d9b43f598 100644 --- a/interactive/tests/explain.rs +++ b/interactive/tests/explain.rs @@ -51,7 +51,6 @@ const SCC_ROW: &str = r#" } export "result" = outer::scc; "#; -const SCC_SHAPES: &[(usize, usize)] = &[(2, 0)]; /// Transitive closure, with the closure's pairs as the result. const TC_ROW: &str = r#" @@ -67,7 +66,6 @@ const TC_ROW: &str = r#" } export "result" = outer::tc; "#; -const TC_SHAPES: &[(usize, usize)] = &[(2, 0)]; /// Reachability from roots (two inputs), the reached set as the result. const REACH_ROW: &str = r#" @@ -79,7 +77,6 @@ const REACH_ROW: &str = r#" } export "result" = reach::reach; "#; -const REACH_SHAPES: &[(usize, usize)] = &[(2, 0), (1, 0)]; fn lowered(src: &str) -> Program { lower::lower_tree(parse::pipe::parse(src)) @@ -128,18 +125,13 @@ fn optimized(src: &str) -> Program { p } -/// The query input's rows: each query is the flat demand envelope -/// `(K ; Tuple([V…, chain…, q]))` — V's fields, then the chain (empty, since the -/// first export is at depth 0), then the query id. +/// The query input's rows: each query is a demand row `(K ; (V, q))` — the +/// queried row, and its query id. fn query_rows(queries: &[(Row, Row)]) -> Vec<(Row, Row)> { queries .iter() .enumerate() - .map(|(q, (k, v))| { - let mut fields = match v { Value::Tuple(xs) => xs.clone(), other => vec![other.clone()] }; - fields.push(Value::Int(q as i64)); - (k.clone(), Value::Tuple(fields)) - }) + .map(|(q, (k, v))| (k.clone(), Value::Tuple(vec![v.clone(), Value::Int(q as i64)]))) .collect() } @@ -147,12 +139,11 @@ fn query_rows(queries: &[(Row, Row)]) -> Vec<(Row, Row)> { /// order) against `src`'s first export, with `src` run on `inputs`. fn demand_for_queries( src: &str, - shapes: &[(usize, usize)], inputs: &[Vec<(Row, Row)>], queries: &[(Row, Row)], ) -> Vec> { let tree = lowered(src); - let mut ex = explain::explain(&tree, shapes); + let mut ex = explain::explain(&tree); ex.optimize(); let mut ex_inputs: Vec> = inputs.to_vec(); ex_inputs.push(query_rows(queries)); @@ -176,12 +167,11 @@ fn demand_for_queries( /// Single-query convenience over [`demand_for_queries`]. fn demand_for( src: &str, - shapes: &[(usize, usize)], inputs: &[Vec<(Row, Row)>], q_key: &Row, q_val: &Row, ) -> Vec> { - demand_for_queries(src, shapes, inputs, &[(q_key.clone(), q_val.clone())]) + demand_for_queries(src, inputs, &[(q_key.clone(), q_val.clone())]) } fn demand_total(demand: &[Vec<(Row, Row)>]) -> usize { @@ -208,14 +198,13 @@ fn restrict(inputs: &[Vec<(Row, Row)>], demand: &[Vec<(Row, Row)>]) -> Vec], ) -> Vec<((Row, Row), usize)> { let p = optimized(src); let result = export_rows(&p, inputs, "result"); let mut sizes = Vec::new(); for (k, v) in &result { - let demand = demand_for(src, shapes, inputs, k, v); + let demand = demand_for(src, inputs, k, v); let replay = export_rows(&p, &restrict(inputs, &demand), "result"); assert!( replay.contains(&(k.clone(), v.clone())), @@ -244,7 +233,7 @@ fn clone_identity_preserves_scc_output() { /// plus one self-loop) is explained by a demand-set that regenerates it. #[test] fn scc_row_explanations_sufficient_small() { - let sizes = assert_all_rows_sufficient(SCC_ROW, SCC_SHAPES, &[gen_edges(50, 55)]); + let sizes = assert_all_rows_sufficient(SCC_ROW, &[gen_edges(50, 55)]); assert_eq!(sizes.len(), 12, "expected 12 scc edges at 50/55"); } @@ -252,7 +241,7 @@ fn scc_row_explanations_sufficient_small() { /// demand-set that regenerates it. #[test] fn tc_row_explanations_sufficient_small() { - let sizes = assert_all_rows_sufficient(TC_ROW, TC_SHAPES, &[gen_edges(20, 22)]); + let sizes = assert_all_rows_sufficient(TC_ROW, &[gen_edges(20, 22)]); assert_eq!(sizes.len(), 34, "expected 34 tc pairs at 20/22"); } @@ -261,7 +250,7 @@ fn tc_row_explanations_sufficient_small() { #[test] fn reach_row_explanations_sufficient_small() { let inputs = vec![gen_edges(50, 55), vec![(row(&[0]), Value::unit())]]; - let sizes = assert_all_rows_sufficient(REACH_ROW, REACH_SHAPES, &inputs); + let sizes = assert_all_rows_sufficient(REACH_ROW, &inputs); assert_eq!(sizes.len(), 5, "expected 5 reached nodes at 50/55 from root 0"); } @@ -273,7 +262,7 @@ fn demand_is_a_subset_of_the_input() { let result = export_rows(&p, &[edges.clone()], "result"); let edge_set: BTreeSet<(Row, Row)> = edges.iter().cloned().collect(); let (k, v) = result.iter().next().unwrap(); - for set in demand_for(SCC_ROW, SCC_SHAPES, &[edges.clone()], k, v) { + for set in demand_for(SCC_ROW, &[edges.clone()], k, v) { for row in set { assert!(edge_set.contains(&row), "demanded row {:?} is not an input row", row); } @@ -284,7 +273,7 @@ fn demand_is_a_subset_of_the_input() { #[test] #[ignore = "heavier sweep; run with --release -- --ignored"] fn scc_row_explanations_sufficient_100() { - let sizes = assert_all_rows_sufficient(SCC_ROW, SCC_SHAPES, &[gen_edges(100, 110)]); + let sizes = assert_all_rows_sufficient(SCC_ROW, &[gen_edges(100, 110)]); assert_eq!(sizes.len(), 22, "expected 22 scc edges at 100/110"); } @@ -300,7 +289,7 @@ fn scc_row_explanations_sufficient_100() { fn scc_join_partner_time_regression() { let edges = gen_edges(1000, 1100); let p = optimized(SCC_ROW); - let demand = demand_for(SCC_ROW, SCC_SHAPES, &[edges.clone()], &row(&[773]), &row(&[466])); + let demand = demand_for(SCC_ROW, &[edges.clone()], &row(&[773]), &row(&[466])); let replay = export_rows(&p, &restrict(&[edges], &demand), "result"); assert!( replay.contains(&(row(&[773]), row(&[466]))), @@ -341,19 +330,19 @@ fn greedy_shrink( #[test] #[ignore = "metric report; run with --release -- --ignored --nocapture"] fn report_demand_excess() { - let cases: Vec<(&str, &str, &[(usize, usize)], Vec>)> = vec![ - ("scc 50/55", SCC_ROW, SCC_SHAPES, vec![gen_edges(50, 55)]), - ("scc 100/110", SCC_ROW, SCC_SHAPES, vec![gen_edges(100, 110)]), - ("tc 20/22", TC_ROW, TC_SHAPES, vec![gen_edges(20, 22)]), - ("tc 50/55", TC_ROW, TC_SHAPES, vec![gen_edges(50, 55)]), - ("reach 50/55", REACH_ROW, REACH_SHAPES, vec![gen_edges(50, 55), vec![(row(&[0]), Value::unit())]]), + let cases: Vec<(&str, &str, Vec>)> = vec![ + ("scc 50/55", SCC_ROW, vec![gen_edges(50, 55)]), + ("scc 100/110", SCC_ROW, vec![gen_edges(100, 110)]), + ("tc 20/22", TC_ROW, vec![gen_edges(20, 22)]), + ("tc 50/55", TC_ROW, vec![gen_edges(50, 55)]), + ("reach 50/55", REACH_ROW, vec![gen_edges(50, 55), vec![(row(&[0]), Value::unit())]]), ]; - for (name, src, shapes, inputs) in cases { + for (name, src, inputs) in cases { let p = optimized(src); let result = export_rows(&p, &inputs, "result"); let (mut tot_d, mut tot_m, mut worst): (usize, usize, f64) = (0, 0, 1.0); for (k, v) in &result { - let demand = demand_for(src, shapes, &inputs, k, v); + let demand = demand_for(src, &inputs, k, v); let min = greedy_shrink(&p, &inputs, &demand, &(k.clone(), v.clone())); let (d, m) = (demand_total(&demand), demand_total(&min)); tot_d += d; @@ -374,14 +363,13 @@ const INDEG_ROW: &str = r#" let edges = input 0 : ((int, int) ; ()) | key($0[0] ; $0[1]); export "result" = edges | key($1 ;) | count; "#; -const INDEG_SHAPES: &[(usize, usize)] = &[(2, 0)]; /// Count outputs over duplicate-free inputs: sufficient today, because the /// keyed lookup demands every input row at the key, which is precisely the /// count's multiset. #[test] fn count_explanations_sufficient_small() { - let sizes = assert_all_rows_sufficient(INDEG_ROW, INDEG_SHAPES, &[gen_edges(20, 22)]); + let sizes = assert_all_rows_sufficient(INDEG_ROW, &[gen_edges(20, 22)]); assert!(!sizes.is_empty()); } @@ -402,7 +390,7 @@ fn count_duplicate_inputs_sufficient() { let p = optimized(INDEG_ROW); let result = export_rows(&p, &[edges.clone()], "result"); assert!(result.contains(&(row(&[5]), row(&[3]))), "expected indeg(5) = 3 in {:?}", result); - let demand = demand_for(INDEG_ROW, INDEG_SHAPES, &[edges.clone()], &row(&[5]), &row(&[3])); + let demand = demand_for(INDEG_ROW, &[edges.clone()], &row(&[5]), &row(&[3])); let replay = export_rows(&p, &restrict(&[edges], &demand), "result"); assert!( replay.contains(&(row(&[5]), row(&[3]))), @@ -423,7 +411,7 @@ fn two_simultaneous_queries_sufficient() { let q0 = (row(&[7]), row(&[44])); let q1 = (row(&[16]), row(&[19])); assert!(result.contains(&q0) && result.contains(&q1)); - let demand = demand_for_queries(SCC_ROW, SCC_SHAPES, &[edges.clone()], &[q0.clone(), q1.clone()]); + let demand = demand_for_queries(SCC_ROW, &[edges.clone()], &[q0.clone(), q1.clone()]); let replay = export_rows(&p, &restrict(&[edges], &demand), "result"); assert!(replay.contains(&q0), "union demand fails first query: {:?}", replay); assert!(replay.contains(&q1), "union demand fails second query: {:?}", replay); @@ -437,11 +425,11 @@ fn two_simultaneous_queries_sufficient() { fn fuzz_explanations_sufficient() { let mut queries = 0; for seed in 0..10u64 { - queries += assert_all_rows_sufficient(SCC_ROW, SCC_SHAPES, &[gen_edges_seeded(seed, 80, 88)]).len(); - queries += assert_all_rows_sufficient(TC_ROW, TC_SHAPES, &[gen_edges_seeded(seed, 25, 27)]).len(); + queries += assert_all_rows_sufficient(SCC_ROW, &[gen_edges_seeded(seed, 80, 88)]).len(); + queries += assert_all_rows_sufficient(TC_ROW, &[gen_edges_seeded(seed, 25, 27)]).len(); let root = (seed % 60) as i64; let inputs = vec![gen_edges_seeded(seed, 60, 66), vec![(row(&[root]), Value::unit())]]; - queries += assert_all_rows_sufficient(REACH_ROW, REACH_SHAPES, &inputs).len(); + queries += assert_all_rows_sufficient(REACH_ROW, &inputs).len(); } assert!(queries >= 100, "fuzz swept only {} queries — instances too sparse to mean much", queries); } @@ -454,14 +442,66 @@ const FLATMAP_ROW: &str = r#" let rows = input 0 : ((int, int) ; ()) | key($0[0] ; list($0[0], $0[1])); export "result" = rows | flatmap($1[0]); "#; -const FLATMAP_SHAPES: &[(usize, usize)] = &[(2, 0)]; #[test] fn flatmap_explanations_sufficient_small() { - let sizes = assert_all_rows_sufficient(FLATMAP_ROW, FLATMAP_SHAPES, &[gen_edges(20, 22)]); + let sizes = assert_all_rows_sufficient(FLATMAP_ROW, &[gen_edges(20, 22)]); assert!(!sizes.is_empty(), "expected some exploded outputs to explain"); } +/// User terms that read a whole row, `$1` or `$0` rather than one of its +/// fields, see the user's row: not explain's, whose value carries more after +/// the user's fields. The filter is always true, and the flatmap's list reads +/// the key. +const WHOLE_ROW: &str = r#" + let edges = input 0 : ((int, int) ; ()) | key($0[0] ; $0[1]) | filter($1 == tuple($1[0])); + let roots = input 1 : ((int) ; ()) | key($0[0] ;); + reach: { + let proposals = reach | join(edges, ($2 ;)); + var reach = roots + proposals | distinct; + } + export "result" = reach::reach | key($0 ; $0) | flatmap(list($0[0], $1[0])) | filter($1 == tuple($1[0], $1[1])); +"#; + +#[test] +fn whole_row_explanations_sufficient_small() { + let inputs = vec![gen_edges(50, 55), vec![(row(&[0]), Value::unit())]]; + let sizes = assert_all_rows_sufficient(WHOLE_ROW, &inputs); + assert_eq!(sizes.len(), 10, "expected 5 reached nodes, each in two positions"); +} + +/// Keys and values that are neither products nor units: a source of scalar +/// keys and values, a scalar value at the root (`value`), a list at the root +/// (`collect`) read by a flatmap, and scalar keys inside an iterative scope +/// (lifted host forms) read whole by a filter. +const SCALARS: &str = r#" + let pairs = input 0 : (int ; int); + let counted = pairs | filter($1 > 3) | count; + let lists = pairs | value($1 * 2) | collect; + let flat = lists | flatmap($1) | value($1[1]); + inner: { + var low = (flat | filter($1 < 15)) + (low | filter($0 == $0)) | min; + } + export "result" = inner::low | join(counted, ($0 ; $1, $2)); +"#; + +fn scalar_pairs() -> Vec<(Row, Row)> { + (0..30).map(|i| (Value::Int(i % 7), Value::Int((i * 5) % 11))).collect() +} + +#[test] +fn scalar_explanations_sufficient_small() { + let sizes = assert_all_rows_sufficient(SCALARS, &[scalar_pairs()]); + assert!(!sizes.is_empty(), "expected some rows to explain"); +} + +#[test] +fn corgi_agrees_on_scalar_explanation() { + let inputs = vec![scalar_pairs()]; + let q = first_output_row(SCALARS, &inputs); + assert_explained_backends_agree(SCALARS, &inputs, &[q]); +} + /// Collect (NEST): `(a ; b)` rows group to `(a ; [b…])`. Reversing a demanded /// list must demand all the members — which is exactly the non-min keyed /// lookup ("demand all same-key inputs"), so no new rule is needed; this @@ -470,11 +510,10 @@ const COLLECT_ROW: &str = r#" let rows = input 0 : ((int, int) ; ()) | key($0[0] ; $0[1]); export "result" = rows | collect; "#; -const COLLECT_SHAPES: &[(usize, usize)] = &[(2, 0)]; #[test] fn collect_explanations_sufficient_small() { - let sizes = assert_all_rows_sufficient(COLLECT_ROW, COLLECT_SHAPES, &[gen_edges(20, 22)]); + let sizes = assert_all_rows_sufficient(COLLECT_ROW, &[gen_edges(20, 22)]); assert!(!sizes.is_empty(), "expected some collected lists to explain"); } @@ -503,12 +542,11 @@ fn collect_explanations_sufficient_small() { /// and require identical exports. fn assert_explained_backends_agree( src: &str, - shapes: &[(usize, usize)], inputs: &[Vec<(Row, Row)>], queries: &[(Row, Row)], ) { let tree = lowered(src); - let mut ex = explain::explain(&tree, shapes); + let mut ex = explain::explain(&tree); ex.optimize(); let mut ex_inputs: Vec> = inputs.to_vec(); ex_inputs.push(query_rows(queries)); @@ -537,7 +575,7 @@ fn first_output_row(src: &str, inputs: &[Vec<(Row, Row)>]) -> (Row, Row) { fn corgi_agrees_on_reach_explanation() { let inputs = vec![gen_edges(50, 55), vec![(row(&[0]), Value::unit())]]; let q = first_output_row(REACH_ROW, &inputs); - assert_explained_backends_agree(REACH_ROW, REACH_SHAPES, &inputs, &[q]); + assert_explained_backends_agree(REACH_ROW, &inputs, &[q]); } /// Transitive closure: iteration with `distinct` rather than `min`. @@ -545,7 +583,7 @@ fn corgi_agrees_on_reach_explanation() { fn corgi_agrees_on_tc_explanation() { let inputs = vec![gen_edges(20, 22)]; let q = first_output_row(TC_ROW, &inputs); - assert_explained_backends_agree(TC_ROW, TC_SHAPES, &inputs, &[q]); + assert_explained_backends_agree(TC_ROW, &inputs, &[q]); } /// The list ops through the rewrite: `flatmap`'s reverse rule (intro + explode) @@ -554,14 +592,14 @@ fn corgi_agrees_on_tc_explanation() { fn corgi_agrees_on_flatmap_explanation() { let inputs = vec![gen_edges(20, 22)]; let q = first_output_row(FLATMAP_ROW, &inputs); - assert_explained_backends_agree(FLATMAP_ROW, FLATMAP_SHAPES, &inputs, &[q]); + assert_explained_backends_agree(FLATMAP_ROW, &inputs, &[q]); } #[test] fn corgi_agrees_on_collect_explanation() { let inputs = vec![gen_edges(20, 22)]; let q = first_output_row(COLLECT_ROW, &inputs); - assert_explained_backends_agree(COLLECT_ROW, COLLECT_SHAPES, &inputs, &[q]); + assert_explained_backends_agree(COLLECT_ROW, &inputs, &[q]); } /// `enter_at` through the rewrite, at depth 1 so the negation trigger is absent. @@ -580,7 +618,7 @@ const REACH_ENTER_AT: &str = r#" fn corgi_agrees_on_enter_at_explanation() { let inputs = vec![gen_edges(50, 55), vec![(row(&[0]), Value::unit())]]; let q = first_output_row(REACH_ENTER_AT, &inputs); - assert_explained_backends_agree(REACH_ENTER_AT, REACH_SHAPES, &inputs, &[q]); + assert_explained_backends_agree(REACH_ENTER_AT, &inputs, &[q]); } /// A `min` reducer in an iterative loop, at depth 1. @@ -598,7 +636,7 @@ const MIN_LOOP: &str = r#" fn corgi_agrees_on_min_loop_explanation() { let inputs = vec![gen_edges(50, 55)]; let q = first_output_row(MIN_LOOP, &inputs); - assert_explained_backends_agree(MIN_LOOP, SCC_SHAPES, &inputs, &[q]); + assert_explained_backends_agree(MIN_LOOP, &inputs, &[q]); } /// SCC's shape — depth-2 nesting, `min`, three joins, a filtered feedback — with @@ -647,14 +685,14 @@ const SCC_ONE_SCOPE_NEGATED: &str = r#" fn corgi_agrees_on_explained_scc_one_scope() { let inputs = vec![gen_edges(50, 55)]; let q = first_output_row(SCC_ONE_SCOPE, &inputs); - assert_explained_backends_agree(SCC_ONE_SCOPE, SCC_SHAPES, &inputs, &[q]); + assert_explained_backends_agree(SCC_ONE_SCOPE, &inputs, &[q]); } #[test] fn explained_scc_one_scope_negated() { let inputs = vec![gen_edges(50, 55)]; let q = first_output_row(SCC_ONE_SCOPE_NEGATED, &inputs); - assert_explained_backends_agree(SCC_ONE_SCOPE_NEGATED, SCC_SHAPES, &inputs, &[q]); + assert_explained_backends_agree(SCC_ONE_SCOPE_NEGATED, &inputs, &[q]); } /// Iterative + `enter_at` + negated feedback: the real SCC. @@ -662,7 +700,7 @@ fn explained_scc_one_scope_negated() { fn corgi_agrees_on_scc_explanation() { let inputs = vec![gen_edges(50, 55)]; let q = first_output_row(SCC_ROW, &inputs); - assert_explained_backends_agree(SCC_ROW, SCC_SHAPES, &inputs, &[q]); + assert_explained_backends_agree(SCC_ROW, &inputs, &[q]); } /// Two queries at once, so the query input carries more than one envelope. @@ -672,24 +710,23 @@ fn corgi_agrees_on_two_query_explanation() { let p = optimized(SCC_ROW); let qs: Vec<(Row, Row)> = export_rows(&p, &inputs, "result").into_iter().take(2).collect(); assert_eq!(qs.len(), 2, "expected at least two scc edges to query"); - assert_explained_backends_agree(SCC_ROW, SCC_SHAPES, &inputs, &qs); + assert_explained_backends_agree(SCC_ROW, &inputs, &qs); } -/// Declared sources give the rewrite its field counts and the query input its shape, and the -/// rewritten program installs with every shape known: the demand matches the undeclared run. +/// Declared sources give the query input its shape, and the rewritten program +/// installs with every shape known. The rewrite itself needs no shapes: the +/// demand matches the undeclared run. #[test] fn declared_sources_shape_the_rewrite() { + use corgi::Shape; let undeclared = REACH_ROW .replace("input 0 : ((int, int) ; ()) |", "input 0 |") .replace("input 1 : ((int) ; ()) |", "input 1 |"); - let tree = lowered(REACH_ROW); - assert_eq!(explain::source_arities(&tree, (9, 9)), REACH_SHAPES); - let ex = explain::explain(&tree, REACH_SHAPES); + let ex = explain::explain(&lowered(REACH_ROW)); let query = ex.root.imports.iter().find(|imp| imp.name == "query").unwrap(); - let int = corgi::Shape::Int; - assert_eq!(query.shape, Some((corgi::Shape::Prod(vec![int.clone()]), corgi::Shape::Prod(vec![int])))); + assert_eq!(query.shape, Some((Shape::Prod(vec![Shape::Int]), Shape::Prod(vec![Shape::Unit, Shape::Int])))); // So does a rewrite through nested iterative scopes. - for ex in [ex, explain::explain(&lowered(SCC_ROW), SCC_SHAPES)] { + for ex in [ex, explain::explain(&lowered(SCC_ROW))] { let (_, problems) = interactive::shapes::infer(&ex, &|_| None); assert!(problems.unknown.is_empty() && problems.conflicts.is_empty(), "{problems:?}"); } @@ -697,8 +734,8 @@ fn declared_sources_shape_the_rewrite() { let inputs = vec![gen_edges(50, 55), vec![(row(&[0]), Value::unit())]]; for (k, v) in export_rows(&optimized(REACH_ROW), &inputs, "result") { assert_eq!( - demand_for(REACH_ROW, REACH_SHAPES, &inputs, &k, &v), - demand_for(&undeclared, REACH_SHAPES, &inputs, &k, &v), + demand_for(REACH_ROW, &inputs, &k, &v), + demand_for(&undeclared, &inputs, &k, &v), ); } }