From 39bf6ad603c3579590980cc4b0b930c691f82b4b Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 10 Oct 2026 21:36:39 -0400 Subject: [PATCH 1/2] Bump corgi to WIP 91df1f0 In corgi's ML frontend, parentheses build a tuple in expressions and patterns, as they do in shapes: `(e)` is a one-field tuple and `()` the unit, and a tuple pattern takes apart exactly its fields. Corgi's effect system is gone: checked ops are words that return a sum, and the plain names are the lossy ops. `Builder::graft` inlines a graph at an existing node. Co-Authored-By: Claude Opus 5.5 --- interactive/Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/interactive/Cargo.toml b/interactive/Cargo.toml index 1f54bd180..3eb08119c 100644 --- a/interactive/Cargo.toml +++ b/interactive/Cargo.toml @@ -14,7 +14,7 @@ workspace = true [dependencies] columnar = { workspace = true } # The columnar kernels for the interpreted backend, pinned by git rev. -corgi = { git = "https://github.com/frankmcsherry/WIP", rev = "9737792dd5db4b99e8299898ff917b174d37778b", features = ["serde"] } +corgi = { git = "https://github.com/frankmcsherry/WIP", rev = "91df1f0ba2a16b48e3ebee5c6e5c77dbb3cf527e", features = ["serde"] } differential-dataflow = { workspace = true } serde = { version = "1.0", features = ["derive"] } smallvec = "1.15.1" From fc3847a25996080d8197cc0ecc3a106c3b42f24b Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 10 Oct 2026 21:43:53 -0400 Subject: [PATCH 2/2] DDIR: slots written as corgi lambdas Any term slot may be a corgi lambda, `pat -> body`, in corgi's ML syntax: it runs `let pat = input in body` with `input` the tuple of the slot's rows, `(key, val)` for a linear op and `(key, left val, right val)` for a join. In `key`, `map`, and `join` the body gives the pair `(key, val)`; in `value`, `filter`, `flatmap`, and `enter_at` it gives the one value. For example, `key(((src, dst), ()) -> ((src), (dst)))` is `key($0[0] ; $0[1])`. The lexer takes a lambda's text whole, since corgi's syntax is not DDIR's: a parenthesized group with a `->` outside nested brackets is a lambda, from after the group's last comma before the arrow. A lambda is `Term::Corgi(src, args)`. The corgi backend parses it once and grafts its graph into the operator's, and `lower` now runs corgi's CSE, so a lambda that gives both key and val runs once. The row backend runs it on one row at a time. Parse and shape errors, including a pattern that does not fit its rows, are reported at install. Co-Authored-By: Claude Opus 5.5 --- interactive/src/corgi/logic.rs | 41 +++++++- interactive/src/explain/mod.rs | 1 + interactive/src/ir.rs | 7 ++ interactive/src/parse/pipe.rs | 107 +++++++++++++++++++-- interactive/tests/corgi_lambdas.rs | 146 +++++++++++++++++++++++++++++ interactive/tests/explain.rs | 26 +++++ 6 files changed, 318 insertions(+), 10 deletions(-) create mode 100644 interactive/tests/corgi_lambdas.rs diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs index 74cb4d275..960ae02fd 100644 --- a/interactive/src/corgi/logic.rs +++ b/interactive/src/corgi/logic.rs @@ -188,7 +188,7 @@ fn mentions_env(t: &Term, depth: usize) -> bool { Term::Var(_) => true, Term::Bound(k) => *k >= depth, Term::Int(_) => false, - Term::Tuple(fs) | Term::List(fs) | Term::Hash(fs) | Term::Call(_, fs) => fs.iter().any(|f| mentions_env(f, depth)), + Term::Tuple(fs) | Term::List(fs) | Term::Hash(fs) | Term::Call(_, fs) | Term::Corgi(_, fs) => fs.iter().any(|f| mentions_env(f, depth)), Term::Spread(inner) | Term::Proj(inner, _) | Term::Unary(_, inner) => mentions_env(inner, depth), Term::Inject { tag, payload, .. } => mentions_env(tag, depth) || mentions_env(payload, depth), Term::Case { scrutinee, arms, default } => { @@ -278,6 +278,17 @@ pub fn compile( }; Ok(b.add(NumOp::Host(corgi::HostOp(kernel)), vec![input])) } + Term::Corgi(src, args) => { + // The program's input is the tuple of its arguments, a `Unit` over the anchor for none. + let input = if args.is_empty() { + b.add(Op::Unit, vec![anchor]) + } else { + let fields = args.iter().map(|a| compile(a, b, env, env_shapes, anchor, None)).collect::>>()?; + b.tuple(fields) + }; + let program = parsed(src)?; + Ok(b.graft(&program, input)) + } Term::Var(i) => env.get(*i).copied().ok_or_else(|| format!("`${i}` is not in scope here")), Term::Bound(k) => { env.len().checked_sub(1 + *k).map(|i| env[i]).ok_or_else(|| format!("binder `^{k}` is not in scope here")) @@ -763,7 +774,8 @@ fn lower(terms: &[&Term], shapes: &[Shape]) -> Res<(Graph, Shape)> { let env: Vec = (0..shapes.len()).map(|i| b.add(Op::Field(i), vec![input])).collect(); let outs = terms.iter().map(|t| compile(t, &mut b, &env, shapes, input, None)).collect::>>()?; let out = if let [one] = outs[..] { one } else { b.tuple(outs) }; - let g = b.finish(out); + // CSE, so a term compiled twice runs once: a lambda's key and val are two fields of one program. + let g = corgi::cse(&b.finish(out)); let output = corgi::shape_of(&g, &Shape::Prod(shapes.to_vec()))?; Ok((g, output)) } @@ -807,6 +819,31 @@ pub fn compile_projection(key: &Term, val: &Term, kshape: &Shape, vshape: &Shape Ok(lower(&[key, val], &[kshape.clone(), vshape.clone()])?.0) } +/// Evaluate the corgi program `src` on one row, as `ir::eval` does for `Term::Corgi`. The row's +/// shape is its own, so a row whose shape it cannot tell alone (an empty list, a variant) panics. +pub fn eval_corgi_row(src: &str, input: DValue) -> DValue { + let shape = shape_of_row(&input).unwrap_or_else(|e| panic!("corgi program `{src}`: {e}")); + let graph = parsed(src).unwrap_or_else(|e| panic!("corgi program `{src}`: {e}")); + let out = corgi::eval_graph(&graph, transcode_owned(vec![input], &shape)); + let out_shape = corgi::shape_of_value(&out); + untranscode(out, &out_shape).pop().expect("one row in, one row out") +} + +/// The graph of the corgi program `src` (`Term::Corgi`), parsed once for the life of the process. +/// `Err` is a parse error. +fn parsed(src: &str) -> Res>> { + use std::collections::HashMap; + use std::sync::{Arc, Mutex, OnceLock}; + static GRAPHS: OnceLock>>>> = OnceLock::new(); + let mut map = GRAPHS.get_or_init(Default::default).lock().unwrap(); + if let Some(g) = map.get(src) { + return Ok(Arc::clone(g)); + } + let g = Arc::new(corgi::parse_ml(src)?); + map.insert(src.to_string(), Arc::clone(&g)); + Ok(g) +} + /// The Float functions without a composition of corgi ops, as host kernels over the float /// column: apply the function as `ir::eval` does. One kernel per function, shared by every call /// site, so CSE merges equal calls. diff --git a/interactive/src/explain/mod.rs b/interactive/src/explain/mod.rs index 9a702aaa2..cfde10fde 100644 --- a/interactive/src/explain/mod.rs +++ b/interactive/src/explain/mod.rs @@ -264,6 +264,7 @@ fn with_rows(t: &Term, rows: &[Term]) -> Term { 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::Corgi(src, ts) => Term::Corgi(src.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)), diff --git a/interactive/src/ir.rs b/interactive/src/ir.rs index eb6f02a56..2a529ecae 100644 --- a/interactive/src/ir.rs +++ b/interactive/src/ir.rs @@ -131,6 +131,9 @@ pub enum Term { /// columnar kernel if one is registered (`register_kernel`), otherwise its /// row body over just the call's arguments. Call(String, Vec), + /// A corgi ML program, run with the tuple of the arguments as its `input`. A slot written + /// `pat -> body` is `let pat = input in body` over the slot's input rows. + Corgi(String, Vec), } /// The sum type an `Inject` builds into. `Declared` carries the full lane shapes of a `type` @@ -327,6 +330,10 @@ pub fn eval(term: &Term, env: &mut Vec) -> Value { } (f.body)(&vals) } + Term::Corgi(src, args) => { + let input = Value::Tuple(args.iter().map(|a| eval(a, env)).collect()); + crate::corgi::logic::eval_corgi_row(src, input) + } Term::Var(i) => env[*i].clone(), Term::Bound(k) => env[env.len() - 1 - *k].clone(), Term::Int(n) => Value::Int(*n), diff --git a/interactive/src/parse/pipe.rs b/interactive/src/parse/pipe.rs index 9caccecef..a611a9113 100644 --- a/interactive/src/parse/pipe.rs +++ b/interactive/src/parse/pipe.rs @@ -92,6 +92,14 @@ //! - Binders: `^k` refers to the k-th enclosing `case`/`fold` binder (de //! Bruijn, innermost = 0); `case` patterns also bind payload fields by name. //! - Conditional: `if(cond, then, els)`. +//! +//! # Corgi lambdas +//! +//! Any slot may instead be a corgi lambda, `pat -> body`, in corgi's ML syntax: it runs +//! `let pat = input in body` with `input` the tuple of the slot's rows, `(key, val)` for a linear +//! op and `(key, left val, right val)` for a join. In `key`, `map`, and `join` the body gives the +//! pair `(key, val)`; in `value`, `filter`, `flatmap`, and `enter_at` it gives the one value. For +//! example, `key(((src, dst), ()) -> ((src), (dst)))` is `key($0[0] ; $0[1])`. use super::*; use crate::ir::{BinOp, Projection, Reducer, SumTy, Term, UnOp}; @@ -101,21 +109,83 @@ enum Token { Let, Var, Export, Type, Case, Fold, Input, Import, Key, Map, Join, Min, Distinct, Count, Collect, Arrange, Negate, Filter, EnterAt, LiftIter, FlatMap, Inspect, Ident(String), Int(i64), Str(String), + /// A slot written as a corgi lambda, `pat -> body`, kept as corgi source. + Lambda { pat: String, body: String }, Dollar, Caret, LParen, RParen, LBrace, RBrace, LBracket, RBracket, Comma, Semi, Colon, ColonColon, Eq, EqEq, NotEq, Lt, LtEq, Gt, GtEq, AndAnd, FatArrow, Pipe, Plus, Minus, Star, Eof, } +/// The source as a stream of chars, with the byte offset of the next one. +#[derive(Clone)] +struct Chars<'a> { src: &'a str, at: usize, peeked: Option } + +impl Chars<'_> { + fn peek(&mut self) -> Option<&char> { self.peeked = self.src[self.at..].chars().next(); self.peeked.as_ref() } +} + +impl Iterator for Chars<'_> { + type Item = char; + fn next(&mut self) -> Option { let c = self.src[self.at..].chars().next()?; self.at += c.len_utf8(); Some(c) } +} + +/// A corgi lambda that ends the parenthesized group starting at byte `from` of `src` (just after +/// its `(`): the group holds a `->` outside any nested bracket. Returns the lambda's start (after +/// the group's last comma before the arrow, or `from`), the arrow, and the group's `)`. +/// +/// Corgi's text is not DDIR's (it has `.2`, float literals, `#` comments), so the lexer takes the +/// lambda whole rather than lexing it. A pattern holds no comma outside parentheses, so the last +/// comma before the arrow separates the lambda from the arguments before it, as in `join`. +fn lambda_in(src: &str, from: usize) -> Option<(usize, usize, usize)> { + let bytes = src.as_bytes(); + let (mut depth, mut start, mut arrow) = (0usize, from, None); + let mut i = from; + while i < bytes.len() { + match bytes[i] { + b'"' => { + i += 1; + while i < bytes.len() && bytes[i] != b'"' { if bytes[i] == b'\\' { i += 1; } i += 1; } + } + b'#' => { while i < bytes.len() && bytes[i] != b'\n' { i += 1; } } + b'-' if bytes.get(i + 1) == Some(&b'-') => { while i < bytes.len() && bytes[i] != b'\n' { i += 1; } } + b'-' if bytes.get(i + 1) == Some(&b'>') && depth == 0 && arrow.is_none() => { arrow = Some(i); i += 1; } + b'(' | b'[' | b'{' => depth += 1, + b')' | b']' | b'}' if depth == 0 => return arrow.map(|a| (start, a, i)), + b')' | b']' | b'}' => depth -= 1, + b',' if depth == 0 && arrow.is_none() => start = i + 1, + _ => {} + } + i += 1; + } + None +} + fn tokenize(input: &str) -> Vec { + let mut tokens = lex(input); + tokens.push(Token::Eof); + tokens +} + +fn lex(input: &str) -> Vec { let mut tokens = Vec::new(); - let mut chars = input.chars().peekable(); + let mut chars = Chars { src: input, at: 0, peeked: None }; while let Some(&ch) = chars.peek() { match ch { ' ' | '\t' | '\n' | '\r' => { chars.next(); }, '-' if chars.clone().nth(1).map_or(false, |c| c == '-') => { while let Some(&c) = chars.peek() { chars.next(); if c == '\n' { break; } } }, - '(' => { chars.next(); tokens.push(Token::LParen); }, + '(' => { + chars.next(); + tokens.push(Token::LParen); + if let Some((start, arrow, close)) = lambda_in(input, chars.at) { + tokens.extend(lex(&input[chars.at..start])); + let (pat, body) = (input[start..arrow].trim().to_string(), input[arrow + 2..close].trim().to_string()); + tokens.push(Token::Lambda { pat, body }); + tokens.push(Token::RParen); + chars.at = close + 1; + } + }, ')' => { chars.next(); tokens.push(Token::RParen); }, '{' => { chars.next(); tokens.push(Token::LBrace); }, '}' => { chars.next(); tokens.push(Token::RBrace); }, @@ -161,7 +231,6 @@ fn tokenize(input: &str) -> Vec { other => panic!("Unexpected character: {:?}", other), } } - tokens.push(Token::Eof); tokens } @@ -455,19 +524,19 @@ impl Parser { Token::Map => { self.next(); let p = self.parse_projection(); Expr::Map(Box::new(lhs), p) }, Token::Ident(name) if name == "value" => { self.next(); self.expect(&Token::LParen); - let val = self.parse_term(); self.expect(&Token::RParen); + let val = self.parse_slot_term(); self.expect(&Token::RParen); Expr::Map(Box::new(lhs), Projection { key: Term::Var(0), val }) }, - Token::Join => { self.next(); self.expect(&Token::LParen); let r = self.parse_join_arg(); self.expect(&Token::Comma); let p = self.parse_projection(); self.expect(&Token::RParen); Expr::Join(Box::new(lhs), Box::new(r), p) }, + Token::Join => { self.next(); self.expect(&Token::LParen); let r = self.parse_join_arg(); self.expect(&Token::Comma); let p = match self.lambda(3) { Some(l) => Self::split(l), None => self.parse_projection() }; self.expect(&Token::RParen); Expr::Join(Box::new(lhs), Box::new(r), p) }, Token::Min => { self.next(); Expr::Reduce(Box::new(lhs), Reducer::Min) }, Token::Distinct => { self.next(); Expr::Reduce(Box::new(lhs), Reducer::Distinct) }, Token::Count => { self.next(); Expr::Reduce(Box::new(lhs), Reducer::Count) }, Token::Collect => { self.next(); Expr::Reduce(Box::new(lhs), Reducer::Collect) }, - Token::FlatMap => { self.next(); self.expect(&Token::LParen); let l = self.parse_term(); self.expect(&Token::RParen); Expr::FlatMap(Box::new(lhs), l) }, + Token::FlatMap => { self.next(); self.expect(&Token::LParen); let l = self.parse_slot_term(); self.expect(&Token::RParen); Expr::FlatMap(Box::new(lhs), l) }, Token::Arrange => { self.next(); Expr::Arrange(Box::new(lhs)) }, Token::Negate => { self.next(); Expr::Negate(Box::new(lhs)) }, - Token::Filter => { self.next(); self.expect(&Token::LParen); let c = self.parse_term(); self.expect(&Token::RParen); Expr::Filter(Box::new(lhs), c) }, - Token::EnterAt => { self.next(); self.expect(&Token::LParen); let f = self.parse_term(); self.expect(&Token::RParen); Expr::EnterAt(Box::new(lhs), f) }, + Token::Filter => { self.next(); self.expect(&Token::LParen); let c = self.parse_slot_term(); self.expect(&Token::RParen); Expr::Filter(Box::new(lhs), c) }, + Token::EnterAt => { self.next(); self.expect(&Token::LParen); let f = self.parse_slot_term(); self.expect(&Token::RParen); Expr::EnterAt(Box::new(lhs), f) }, Token::LiftIter => { self.next(); Expr::LiftIter(Box::new(lhs)) }, Token::Inspect => { self.next(); self.expect(&Token::LParen); let l = self.parse_ident(); self.expect(&Token::RParen); Expr::Inspect(Box::new(lhs), l) }, other => panic!("Expected pipe operator, got {:?}", other), @@ -476,9 +545,31 @@ impl Parser { fn parse_projection(&mut self) -> Projection { self.expect(&Token::LParen); + if let Some(l) = self.lambda(2) { + self.expect(&Token::RParen); + return Self::split(l); + } self.parse_projection_inner() } + /// A slot written as a corgi lambda, over the slot's `arity` input rows (two for a linear + /// op, three for a join): a term running `let pat = input in body` on the tuple of them. + fn lambda(&mut self, arity: usize) -> Option { + let Token::Lambda { pat, body } = self.peek().clone() else { return None }; + self.next(); + Some(Term::Corgi(format!("let {pat} = input in {body}"), (0..arity).map(Term::Var).collect())) + } + + /// The projection a lambda's `(key, val)` result gives. + fn split(lambda: Term) -> Projection { + Projection { key: Term::Proj(Box::new(lambda.clone()), 0), val: Term::Proj(Box::new(lambda), 1) } + } + + /// A linear op's term slot: a corgi lambda, or a term. + fn parse_slot_term(&mut self) -> Term { + self.lambda(2).unwrap_or_else(|| self.parse_term()) + } + // A projection `(k1, k2 ; v1, v2)` builds `key = tuple(k1, k2)` and // `val = tuple(v1, v2)`. A bare `$n` field splices the input row's tuple // fields (`Spread`), or is one field when the row is not a tuple; any other diff --git a/interactive/tests/corgi_lambdas.rs b/interactive/tests/corgi_lambdas.rs new file mode 100644 index 000000000..22e1d95dc --- /dev/null +++ b/interactive/tests/corgi_lambdas.rs @@ -0,0 +1,146 @@ +//! Slots written as corgi lambdas, `pat -> body`, compute what the same slots written as terms do, +//! on both backends. + +use interactive::ir::Value; +use interactive::server::{evaluate, RenderBackend}; +use interactive::{lower, parse}; + +fn tup(fields: &[i64]) -> Value { + Value::Tuple(fields.iter().map(|&n| Value::Int(n)).collect()) +} +fn rows(rs: &[&[i64]]) -> Vec<(Value, Value)> { + rs.iter().map(|f| (tup(f), Value::unit())).collect() +} + +/// Evaluate both programs on both backends and assert all four agree, and have output. +fn assert_same(terms: &str, lambdas: &str, inputs: &[Vec<(Value, Value)>]) { + let tree = |src: &str| { + let mut tree = lower::lower_tree(parse::pipe::parse(src)); + tree.optimize(); + tree + }; + let (terms, lambdas) = (tree(terms), tree(lambdas)); + let want = evaluate(RenderBackend::Vec, timely::Config::process(1), &terms, inputs); + assert!(want.values().all(|rows| !rows.is_empty()), "every export has rows: {want:?}"); + assert_eq!(evaluate(RenderBackend::Vec, timely::Config::process(1), &lambdas, inputs), want, "vec backend"); + for workers in [1, 3] { + let config = || timely::Config::process(workers); + assert_eq!(evaluate(RenderBackend::Corgi, config(), &terms, inputs), want, "corgi backend, terms, {workers} workers"); + assert_eq!(evaluate(RenderBackend::Corgi, config(), &lambdas, inputs), want, "corgi backend, lambdas, {workers} workers"); + } +} + +fn edges() -> Vec<(Value, Value)> { + rows(&[&[1, 2], &[2, 3], &[3, 4], &[5, 6], &[4, 2], &[6, 5]]) +} + +#[test] +fn reach() { + let terms = r#" + let edges = input 0 : ((int, int) ; ()) | key($0[0] ; $0[1]); + let roots = input 1 : ((int) ; ()) | key($0[0] ;); + reach: { + let proposals = reach | join(edges, ($2 ;)); + var reach = roots + proposals | distinct; + } + export "result" = reach::reach; + "#; + let lambdas = r#" + let edges = input 0 : ((int, int) ; ()) | key(((src, dst), ()) -> ((src), (dst))); + let roots = input 1 : ((int) ; ()) | key(((n), ()) -> ((n), ())); + reach: { + let proposals = reach | join(edges, (_, _, dst) -> (dst, ())); + var reach = roots + proposals | distinct; + } + export "result" = reach::reach; + "#; + assert_same(terms, lambdas, &[edges(), rows(&[&[1]])]); +} + +#[test] +fn scc() { + let terms = &interactive::load_program(&format!("{}/examples/programs/scc.ddp", env!("CARGO_MANIFEST_DIR"))); + let lambdas = r#" + let edges = input 0 : ((int, int) ; ()) | key(((src, dst), ()) -> ((src), (dst))); + let trans = edges | key((src, dst) -> (dst, src)); + + outer: { + let scc = edges + trim; + + fwd: { + let nodes = edges | key((_, dst) -> (dst, dst)) | enter_at((_, (n)) -> n); + let labels = proposals + nodes | min; + var proposals = labels | join(scc, (_, label, dst) -> (dst, label)); + } + + let trim_fwd = edges + | join(fwd::labels, ((src), (dst), (l)) -> ((dst), (src, l))) + | join(fwd::labels, (dst, (src, l), (m)) -> (dst, (src, l, m))) + | filter((_, (_, l, m)) -> (l, m) eq) + | key((dst, (src, _, _)) -> (dst, (src))); + + bwd: { + let nodes = trans | key((_, dst) -> (dst, dst)) | enter_at((_, (n)) -> n); + let labels = proposals + nodes | min; + var proposals = labels | join(trim_fwd, (_, label, dst) -> (dst, label)); + } + + let trim_bwd = trans + | join(bwd::labels, ((src), (dst), (l)) -> ((dst), (src, l))) + | join(bwd::labels, (dst, (src, l), (m)) -> (dst, (src, l, m))) + | filter((_, (_, l, m)) -> (l, m) eq) + | key((dst, (src, _, _)) -> (dst, (src))); + + var trim = trim_bwd - edges; + } + + export "result" = outer::scc | map(_ -> ((), ())) | arrange | inspect(total); + "#; + assert_same(terms, lambdas, &[edges()]); +} + +/// The slots that take one term: `value`, `filter`, `flatmap`, `enter_at`. +#[test] +fn term_slots() { + let terms = r#" + let pairs = input 0 : ((int, int) ; ()) | key($0[0] ; $0[1]); + export "value" = pairs | value($1[0] * 10 + $0[0]); + export "filter" = pairs | filter($0[0] < $1[0]); + export "flatmap" = pairs | flatmap(list($0[0], $1[0], $0[0] + $1[0])); + stepped: { + let delayed = pairs | enter_at($1[0]); + var counted = delayed | count; + } + export "enter_at" = stepped::counted; + "#; + let lambdas = r#" + let pairs = input 0 : ((int, int) ; ()) | key(((a, b), ()) -> ((a), (b))); + export "value" = pairs | value(((a), (b)) -> ((b, 10) mul, a) add); + export "filter" = pairs | filter(((a), (b)) -> (a, b) lt); + export "flatmap" = pairs | flatmap(((a), (b)) -> ((a enlist, b enlist) append, (a, b) add enlist) append); + stepped: { + let delayed = pairs | enter_at((_, (b)) -> b); + var counted = delayed | count; + } + export "enter_at" = stepped::counted; + "#; + assert_same(terms, lambdas, &[edges()]); +} + +/// A lambda whose pattern does not fit its slot's rows, or whose body does not parse, is refused +/// at install with corgi's message. +#[test] +fn mismatched_lambdas_are_refused_at_install() { + timely::execute_directly(|worker| { + for (src, says) in [ + (r#"export "a" = input 0 : ((int, int) ; ()) | key(((s), ()) -> ((s), ()));"#, "takes apart a product of 1 field"), + (r#"export "a" = input 0 : ((int, int) ; ()) | filter(((s, d), ()) -> (s, d));"#, "not an integer"), + (r#"export "a" = input 0 : ((int, int) ; ()) | key(((s, d), ()) -> (s, d, ) add);"#, "expected an expression"), + ] { + let mut program = lower::lower_tree(parse::pipe::parse(src)); + program.optimize(); + let error = interactive::server::Server::with_backend(RenderBackend::Corgi).install(worker, "p", &program).unwrap_err(); + assert!(error.contains(says), "{src}: {error}"); + } + }); +} diff --git a/interactive/tests/explain.rs b/interactive/tests/explain.rs index d9b43f598..ecdce57d4 100644 --- a/interactive/tests/explain.rs +++ b/interactive/tests/explain.rs @@ -470,6 +470,32 @@ fn whole_row_explanations_sufficient_small() { assert_eq!(sizes.len(), 10, "expected 5 reached nodes, each in two positions"); } +/// Slots written as corgi lambdas, which read their rows whole: reach, with +/// the always-true filter and the flatmap of `WHOLE_ROW` as lambdas. +const LAMBDAS: &str = r#" + let edges = input 0 : ((int, int) ; ()) | key(((src, dst), ()) -> ((src), (dst))) | filter((_, (d)) -> (d, d) eq); + let roots = input 1 : ((int) ; ()) | key(((n), ()) -> ((n), ())); + reach: { + let proposals = reach | join(edges, (_, _, dst) -> (dst, ())); + var reach = roots + proposals | distinct; + } + export "result" = reach::reach | key((k, _) -> (k, k)) | flatmap(((a), (b)) -> (a enlist, b enlist) append); +"#; + +#[test] +fn lambda_explanations_sufficient_small() { + let inputs = vec![gen_edges(50, 55), vec![(row(&[0]), Value::unit())]]; + let sizes = assert_all_rows_sufficient(LAMBDAS, &inputs); + assert_eq!(sizes.len(), 10, "expected 5 reached nodes, each in two positions"); +} + +#[test] +fn corgi_agrees_on_lambda_explanation() { + let inputs = vec![gen_edges(50, 55), vec![(row(&[0]), Value::unit())]]; + let q = first_output_row(LAMBDAS, &inputs); + assert_explained_backends_agree(LAMBDAS, &inputs, &[q]); +} + /// 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