From 960a07578a4b18e993774b87e7292d9106a6f5f9 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 14 Nov 2026 08:48:00 -0500 Subject: [PATCH] DDIR: join fragments as a separate, opt-in step from surface text to plain DDP (spike) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The same join fragments as #1085, held apart from the core language. `surface::expand` finds each fragment in the text and replaces its source range with plain DDP, leaving every other byte unchanged; the core parser then reads the result. The core lexer, parser, syntax tree, lowering and printers are master-next's, except that `lambda_in` becomes crate-visible so the scanner finds lambda boundaries as the lexer does. `ddir-plan --fragments` and the server's `load … --fragments` opt in; everything else reads plain DDP, and saved expanded text loads without the flag. Errors keep their messages, prefixed with the line and column in the surface text. The expansion is #1085's: written join order, key-slot reads, direct first joins, each test as a plain filter at its earliest join, variants through `enlist unweave`, and constant later keys rejected. Eight tests keep the essentials; the emitted plans match #1085's byte for byte. Co-Authored-By: GPT-6 Sol (Codex) Claude-Session: https://claude.ai/code/session_01H6wB4H99TLw4tqu4x48pXL --- interactive/server/src/cmd.rs | 61 ++- interactive/src/bin/ddir-plan.rs | 5 +- interactive/src/lib.rs | 2 + interactive/src/parse/pipe.rs | 2 +- interactive/src/surface.rs | 671 +++++++++++++++++++++++++++++++ interactive/tests/fragments.rs | 215 ++++++++++ 6 files changed, 932 insertions(+), 24 deletions(-) create mode 100644 interactive/src/surface.rs create mode 100644 interactive/tests/fragments.rs diff --git a/interactive/server/src/cmd.rs b/interactive/server/src/cmd.rs index 63721c515..a4296163e 100644 --- a/interactive/server/src/cmd.rs +++ b/interactive/server/src/cmd.rs @@ -373,6 +373,7 @@ fn max_feed_bytes() -> usize { struct PendingLoad { plan: bool, + fragments: bool, timings: bool, reqid: ReqId, id_hint: String, @@ -492,19 +493,12 @@ impl LineParser { if let Some(err) = done.poisoned { return Some((done.reqid, Err(err))); } - return Some(( - done.reqid, - Ok(if done.plan { - Cmd::Plan { name: done.id_hint, program: done.body, timings: done.timings } - } else { - Cmd::Load { - id_hint: done.id_hint, - bindings: done.bindings, - program: done.body, - explain: done.explain, - } - }), - )); + let program = if done.fragments { interactive::surface::expand(&done.body) } else { Ok(done.body) }; + return Some((done.reqid, program.map(|program| if done.plan { + Cmd::Plan { name: done.id_hint, program, timings: done.timings } + } else { + Cmd::Load { id_hint: done.id_hint, bindings: done.bindings, program, explain: done.explain } + }))); } } if pl.poisoned.is_none() { @@ -550,6 +544,7 @@ impl LineParser { ParseOutcome::Err(e) => Some((reqid, Err(e))), ParseOutcome::BeginLoad { plan, + fragments, timings, id_hint, bindings, @@ -557,6 +552,7 @@ impl LineParser { } => { self.pending_load = Some(PendingLoad { plan, + fragments, timings, reqid, id_hint, @@ -621,6 +617,7 @@ enum ParseOutcome { Err(String), BeginLoad { plan: bool, + fragments: bool, timings: bool, id_hint: String, bindings: BTreeMap, @@ -664,7 +661,7 @@ fn parse_cmd(cmd: &str, args: &[&str], catalog: &Catalog) -> ParseOutcome { Err(error) => ParseOutcome::Err(format!("plan: cannot read {path:?}: {error}")), }, None => ParseOutcome::BeginLoad { - plan: true, timings, id_hint: name.into(), bindings: BTreeMap::new(), explain: None, + plan: true, fragments: false, timings, id_hint: name.into(), bindings: BTreeMap::new(), explain: None, }, } }, @@ -673,7 +670,7 @@ fn parse_cmd(cmd: &str, args: &[&str], catalog: &Catalog) -> 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`, `explain=debug`, `name=binding`. + // ` end-load`. Options: `--fragments`, `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()); @@ -685,15 +682,17 @@ fn parse_cmd(cmd: &str, args: &[&str], catalog: &Catalog) -> ParseOutcome { }; let mut bindings = BTreeMap::new(); let mut explain = None; + let mut fragments = false; for tok in options { match *tok { + "--fragments" => { fragments = true; continue; } "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`, `explain=debug`, or `name=binding`", + "load: argument {:?} must be `--fragments`, `explain`, `explain=debug`, or `name=binding`", tok )); }; @@ -706,16 +705,15 @@ fn parse_cmd(cmd: &str, args: &[&str], catalog: &Catalog) -> ParseOutcome { // A file the server reads itself is not an upload, so the program-size // gate (an intake defense) does not apply to it. Some(path) => match std::fs::read_to_string(path) { - Ok(program) => ParseOutcome::Cmd(Cmd::Load { - id_hint, - bindings, - program, - explain, - }), + Ok(program) => match if fragments { interactive::surface::expand(&program) } else { Ok(program) } { + Ok(program) => ParseOutcome::Cmd(Cmd::Load { id_hint, bindings, program, explain }), + Err(error) => ParseOutcome::Err(error), + }, Err(error) => ParseOutcome::Err(format!("load: cannot read {path:?}: {error}")), }, None => ParseOutcome::BeginLoad { plan: false, + fragments, timings: false, id_hint, bindings, @@ -1006,6 +1004,25 @@ mod tests { assert!(matches!(p.feed("plan p --typo begin"), Some((_, Err(e))) if e.starts_with("plan: expected"))); } + #[test] + fn load_fragments_is_opt_in_for_inline_and_file_text() { + let body = "let a=input 0 : (int,int); output \"out\"=a(x,y);"; + let mut parser = LineParser::new(Catalog::default()); + assert!(parser.feed("load expanded --fragments begin").is_none()); + assert!(parser.feed(body).is_none()); + let (_, parsed) = parser.feed("expanded end-load").unwrap(); + let cmd = parsed.unwrap(); + assert!(matches!(&cmd, Cmd::Load { program, .. } if !program.contains("a(x,y)"))); + prepare(cmd, &Catalog::default()).unwrap(); + let plain = Cmd::Load { id_hint: "plain".into(), bindings: BTreeMap::new(), program: body.into(), explain: None }; + assert!(matches!(prepare(plain, &Catalog::default()), Err(e) if e.contains("Expected Semi, got LParen"))); + let file = std::env::temp_dir().join(format!("surface-load-{}.ddp", std::process::id())); + std::fs::write(&file, body).unwrap(); + let (_, cmd) = parser.feed(&format!("load file from {} --fragments", file.display())).unwrap(); + std::fs::remove_file(file).unwrap(); + prepare(cmd.unwrap(), &Catalog::default()).unwrap(); + } + #[test] fn multiline_load() { let mut p = LineParser::new(Catalog::default()); diff --git a/interactive/src/bin/ddir-plan.rs b/interactive/src/bin/ddir-plan.rs index 4609dbc7b..58d7a3618 100644 --- a/interactive/src/bin/ddir-plan.rs +++ b/interactive/src/bin/ddir-plan.rs @@ -7,6 +7,7 @@ fn run() -> Result<(), String> { let mut options = interactive::plan::PrintOptions::default(); let mut files = Vec::new(); let mut emit_ddp = false; + let mut fragments = false; let mut check_roundtrip = false; let mut stage = interactive::plan::ddp::DdpStage::AsOf; let mut as_of = 0; @@ -19,6 +20,7 @@ fn run() -> Result<(), String> { "--timings" => options.timings = true, "--lambda-source" => options.lambda_source = true, "--check-lambdas" => options.check_lambdas = true, + "--fragments" => fragments = true, "--emit-ddp" => emit_ddp = true, "--emit-ddp-stage" => { let name = args.next().ok_or("--emit-ddp-stage requires parsed, lowered, optimized, projected, or as-of")?; @@ -37,7 +39,7 @@ fn run() -> Result<(), String> { }), } } - if files.is_empty() { return Err("usage: ddir-plan [--as-of T] [--timings] [--passes LIST] [--emit-ddp | --emit-ddp-stage STAGE | --lambda-source] [--check-lambdas] [--check-roundtrip] [name=]file.ddp [[name=]more.ddp ...]".into()); } + if files.is_empty() { return Err("usage: ddir-plan [--fragments] [--as-of T] [--timings] [--passes LIST] [--emit-ddp | --emit-ddp-stage STAGE | --lambda-source] [--check-lambdas] [--check-roundtrip] [name=]file.ddp [[name=]more.ddp ...]".into()); } if emit_ddp && options.lambda_source { return Err("--emit-ddp always checks source against the compiled graph".into()); } if check_roundtrip && !emit_ddp { return Err("--check-roundtrip requires --emit-ddp".into()); } let mut catalog = BTreeMap::new(); @@ -45,6 +47,7 @@ fn run() -> Result<(), String> { let mut out = io::stdout().lock(); for (i, (program, file)) in files.iter().enumerate() { let text = std::fs::read_to_string(file).map_err(|e| format!("{file}: {e}"))?; + let text = if fragments { interactive::surface::expand(&text).map_err(|e| format!("{file}: {e}"))? } else { text }; let report = interactive::plan::report_at(&text, &|name| catalog.get(name).cloned().or_else(|| interactive::server::generated_shape(name)), &passes, as_of).map_err(|e| format!("{file}: {e}"))?; if emit_ddp { // Earlier files supply imports. Only the requested (last) program is emitted. diff --git a/interactive/src/lib.rs b/interactive/src/lib.rs index bd51d2d3b..2dc77af65 100644 --- a/interactive/src/lib.rs +++ b/interactive/src/lib.rs @@ -22,6 +22,8 @@ pub mod scope_ir; // The language: text to a typed scope tree. // `pipe::parse`, the `.ddp` parser, and the statements it returns (`Stmt`, `Expr`). pub mod parse; +// Optional surface text to ordinary DDP text: `expand`. +pub mod surface; // `lower_tree`: parsed statements to a scope tree. pub mod lower; // `infer` and every collection's shape (`CollShape`, `ScopeShapes`, `Problems`). diff --git a/interactive/src/parse/pipe.rs b/interactive/src/parse/pipe.rs index f8873617e..fbbf94742 100644 --- a/interactive/src/parse/pipe.rs +++ b/interactive/src/parse/pipe.rs @@ -179,7 +179,7 @@ impl Iterator for Chars<'_> { /// 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)> { +pub(crate) 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; diff --git a/interactive/src/surface.rs b/interactive/src/surface.rs new file mode 100644 index 000000000..c84ee55e5 --- /dev/null +++ b/interactive/src/surface.rs @@ -0,0 +1,671 @@ +//! Experimental join fragments, lowered from surface text to ordinary DDP text. +//! Only fragment source ranges are replaced; the core language has no knowledge of them. +//! Written join order, existing indexes and first-appearance binding order are preserved. +//! Literal and equality tests run at the first legal checkpoint as plain filters. +//! Variants bind through unweave lists, and may emit the next lookup key directly. +//! Later keys must use an earlier binding; constant-key cross joins belong in plain DDP. +//! This surface form may change or be removed. + +use std::{cell::Cell, collections::BTreeMap, panic::{catch_unwind, AssertUnwindSafe}}; +use crate::{ir::Lambda, parse::Expr}; + +/// Expand experimental fragments before handing the returned text to the core parser. +/// Errors retain the original message, prefixed by the surface line and column. +pub fn expand(text: &str) -> Result { + let at = Cell::new(0); + catch_unwind(AssertUnwindSafe(|| expand_inner(text, &at))).map_err(|panic| { + let message = panic.downcast_ref::().map(String::as_str) + .or_else(|| panic.downcast_ref::<&str>().copied()).unwrap_or("surface expansion panicked"); + let prefix = &text[..at.get().min(text.len())]; + let line = prefix.bytes().filter(|b| *b == b'\n').count() + 1; + let column = prefix.rsplit('\n').next().unwrap().chars().count() + 1; + format!("{line}:{column}: {message}") + }) +} + +#[derive(Clone, Debug, PartialEq)] +enum Token { + Let, Var, Set, Output, Type, Enum, Input, Import, Map, Join, Min, Distinct, Count, + Collect, Arrange, Negate, Filter, LiftIter, FlatMap, Inspect, + Ident(String), Int(i64), Str(String), Lambda, + LParen, RParen, LBrace, RBrace, Comma, Semi, Colon, ColonColon, Eq, + Pipe, Plus, Minus, Star, LBracket, RBracket, Eof, Other(char), +} +struct Span { kind: Token, start: usize, end: usize } +struct Source { tokens: Vec, comments: Vec<(usize, usize)> } + +// The scanner leaves Corgi opaque, using the same lambda boundaries as the core lexer. +fn scan(text: &str, from: usize, to: usize, source: &mut Source, at: &Cell) { + let bytes = text.as_bytes(); + let mut i = from; + while i < to { + let start = i; + at.set(start); + let ch = text[i..].chars().next().unwrap(); + if ch.is_ascii_whitespace() { i += 1; continue; } + if bytes[i..].starts_with(b"--") { + while i < to && bytes[i] != b'\n' { i += 1; } + source.comments.push((start, i)); + continue; + } + if ch == '(' { + source.tokens.push(Span { kind: Token::LParen, start, end: i + 1 }); + if let Some((begin, _, close)) = crate::parse::pipe::lambda_in(text, i + 1).filter(|(_, _, close)| *close < to) { + scan(text, i + 1, begin, source, at); + source.tokens.push(Span { kind: Token::Lambda, start: begin, end: close }); + source.tokens.push(Span { kind: Token::RParen, start: close, end: close + 1 }); + i = close + 1; + } else { i += 1; } + continue; + } + let kind = if ch == '"' { + i += 1; + let begin = i; + while i < to && bytes[i] != b'"' { i += 1; } + assert!(i < to, "Unterminated string literal"); + let s = text[begin..i].to_string(); + i += 1; + Token::Str(s) + } else if ch.is_ascii_digit() { + while i < to && bytes[i].is_ascii_digit() { i += 1; } + Token::Int(text[start..i].parse().unwrap()) + } else if ch.is_ascii_alphabetic() || ch == '_' { + while i < to && (bytes[i].is_ascii_alphanumeric() || bytes[i] == b'_') { i += 1; } + match &text[start..i] { + "let" => Token::Let, "var" => Token::Var, "set" => Token::Set, "output" => Token::Output, + "type" => Token::Type, "enum" => Token::Enum, "input" => Token::Input, "import" => Token::Import, + "map" => Token::Map, "join" => Token::Join, "min" => Token::Min, "distinct" => Token::Distinct, + "count" => Token::Count, "collect" => Token::Collect, "arrange" => Token::Arrange, + "negate" => Token::Negate, "filter" => Token::Filter, "lift_iter" => Token::LiftIter, + "flatmap" => Token::FlatMap, "inspect" => Token::Inspect, + name => Token::Ident(name.into()), + } + } else { + i += ch.len_utf8(); + match ch { + ')' => Token::RParen, '{' => Token::LBrace, '}' => Token::RBrace, + ',' => Token::Comma, ';' => Token::Semi, '=' => Token::Eq, '|' => Token::Pipe, + '+' => Token::Plus, '-' => Token::Minus, '*' => Token::Star, '[' => Token::LBracket, ']' => Token::RBracket, + ':' if bytes.get(i) == Some(&b':') => { i += 1; Token::ColonColon }, + ':' => Token::Colon, + other => Token::Other(other), + } + }; + source.tokens.push(Span { kind, start, end: i }); + } +} + +fn group_end(tokens: &[Span], mut i: usize) -> usize { + let mut depth = 0; + loop { + match tokens[i].kind { + Token::LParen => depth += 1, + Token::RParen => { depth -= 1; if depth == 0 { return i + 1; } } + Token::Eof => return i, + _ => {}, + } + i += 1; + } +} +fn declaration_end(tokens: &[Span], mut i: usize) -> usize { + while !matches!(tokens[i].kind, Token::Semi | Token::Eof) { i += 1; } + if tokens[i].kind == Token::Semi { i + 1 } else { i } +} +fn next_fragment(tokens: &[Span], mut i: usize, at: &Cell) -> Option { + while i + 1 < tokens.len() { + match tokens[i].kind { + Token::Type | Token::Enum => { i = declaration_end(tokens, i); continue; } + Token::Colon if tokens[i + 1].kind == Token::LParen => { + let end = group_end(tokens, i + 1); + if tokens[end].kind == Token::LParen { at.set(tokens[end - 1].start); panic!("a fragment atom names an existing binding"); } + i = end; continue; + } + Token::LParen if tokens[group_end(tokens, i)].kind == Token::LParen => return Some(i), + Token::Input | Token::Import | Token::Int(_) | Token::Str(_) | Token::RParen if tokens[i + 1].kind == Token::LParen => { + at.set(tokens[i].start); + panic!("a fragment atom names an existing binding"); + } + Token::Ident(_) | Token::ColonColon => { + let start = i; + if tokens[i].kind == Token::ColonColon { i += 1; } + if !matches!(tokens[i].kind, Token::Ident(_)) { i += 1; continue; } + i += 1; + while tokens[i].kind == Token::ColonColon { + i += 1; + if !matches!(tokens[i].kind, Token::Ident(_)) { break; } + i += 1; + } + if tokens[i].kind == Token::LParen && (start == 0 || tokens[start - 1].kind != Token::Pipe) { return Some(start); } + continue; + } + _ => i += 1, + } + } + None +} +fn expand_inner(text: &str, at: &Cell) -> String { + let mut source = Source { tokens: Vec::new(), comments: Vec::new() }; + scan(text, 0, text.len(), &mut source, at); + source.tokens.push(Span { kind: Token::Eof, start: text.len(), end: text.len() }); + let tokens = &source.tokens; + let Some(mut start) = next_fragment(tokens, 0, at) else { return text.to_string(); }; + // Type declarations are global. Reuse their existing resolver without parsing collection text. + let mut declarations = text.bytes().map(|b| if b == b'\n' { b } else { b' ' }).collect::>(); + let mut i = 0; + let mut first_decl = None; + while i + 1 < tokens.len() { + if matches!(tokens[i].kind, Token::Type | Token::Enum) { + let end = declaration_end(tokens, i); + let (a, b) = (tokens[i].start, tokens[end.saturating_sub(1)].end); + declarations[a..b].copy_from_slice(&text.as_bytes()[a..b]); + first_decl.get_or_insert(a); + i = end; + } else { i += 1; } + } + at.set(first_decl.unwrap_or(tokens[start].start)); + let (types, scope, _) = crate::parse::pipe::print_context(&String::from_utf8(declarations).unwrap()); + let mut out = String::new(); + let mut copied = 0; + loop { + let begin = tokens[start].start; + at.set(begin); + let mut parser = Parser { tokens, pos: start, scope: &scope, types: &types, at }; + let fragment = parser.fragment(); + let end = tokens[parser.pos - 1].end; + let expr = fragment.expansion(at); + out.push_str(&text[copied..begin]); + out.push('('); + for &(a, b) in &source.comments { + if begin <= a && b <= end { out.push('\n'); out.push_str(&text[a..b]); out.push('\n'); } + } + out.push_str(&expression(&expr)); + out.push(')'); + copied = end; + match next_fragment(tokens, parser.pos, at) { Some(next) => start = next, None => break } + } + out.push_str(&text[copied..]); + out +} + +// Only the ordinary expression kinds produced by this expansion need printing. +fn expression(expr: &Expr) -> String { + fn slot(l: &Lambda) -> String { + let (pat, body) = l.src.strip_prefix("let ").unwrap().split_once(" = input in ").unwrap(); + format!("{pat} -> {body}") + } + match expr { + Expr::Name(n) => n.clone(), + Expr::Path(parts, root) => format!("{}{}", if *root { "::" } else { "" }, parts.join("::")), + Expr::Join(l, r, f) => format!("{}\n | join({}, {})", expression(l), expression(r), slot(f)), + Expr::Map(e, f) | Expr::Filter(e, f) | Expr::FlatMap(e, f) => { + let word = match expr { Expr::Map(..) => "map", Expr::Filter(..) => "filter", _ => "flatmap" }; + format!("{}\n | {word}({})", expression(e), slot(f)) + } + _ => unreachable!("fragment expansion emits names, joins and linear stages"), + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +enum Pattern { + Var(String), Wild, Tuple(Vec), Literal(String), + Variant { name: String, tag: usize, payload: Box }, +} + +impl Pattern { + fn shape_pattern(&self) -> String { + match self { + Self::Tuple(ps) => tuple(&ps.iter().map(Self::shape_pattern).collect::>()), + Self::Variant { name, .. } => name.split("::").next().unwrap().to_string(), + _ => "_".into(), + } + } + fn has_variant(&self) -> bool { + match self { + Self::Variant { .. } => true, + Self::Tuple(ps) => ps.iter().any(Self::has_variant), + _ => false, + } + } + fn constant(&self) -> bool { + match self { Self::Literal(_) => true, Self::Tuple(ps) => ps.iter().all(Self::constant), _ => false } + } + fn check_lists(&self) { + match self { + // Check the whole literal so sibling constants can supply an empty list's shape. + Self::Literal(s) if s != "[]" && s.contains("[]") => { + corgi::parse_ml(s).unwrap_or_else(|e| panic!("DDIR fragment list literal `{s}` needs a complete element shape; use plain DDP with an explicit list shape: {e}")); + } + Self::Tuple(ps) => for p in ps { p.check_lists(); }, + Self::Variant { payload, .. } => payload.check_lists(), + _ => {}, + } + } + fn text(&self) -> String { + match self { + Self::Var(n) | Self::Literal(n) => n.clone(), Self::Wild => "_".into(), + Self::Tuple(ps) => tuple(&ps.iter().map(Self::text).collect::>()), + Self::Variant { name, payload, .. } => format!("{name}({})", payload.text()), + } + } +} + + +struct Atom { source: Expr, key: Pattern, val: Pattern, at: usize, key_at: usize } +struct Fragment { atoms: Vec, scope: corgi::Scope } +struct Parser<'a> { + tokens: &'a [Span], pos: usize, scope: &'a corgi::Scope, + types: &'a [(String, Vec<(String, corgi::Shape)>)], at: &'a Cell, +} +impl Parser<'_> { + fn peek(&self) -> &Token { &self.tokens[self.pos].kind } + fn next(&mut self) -> Token { + let span = &self.tokens[self.pos]; self.at.set(span.start); self.pos += 1; span.kind.clone() + } + fn expect(&mut self, expected: &Token) { + let t = self.next(); assert_eq!(&t, expected, "Expected {:?}, got {:?}", expected, t); + } + fn parse_ident(&mut self) -> String { + match self.next() { Token::Ident(n) => n, t => panic!("Expected ident, got {:?}", t) } + } + fn lambda_scope(&self) -> &corgi::Scope { self.scope } + fn source(&mut self) -> Expr { + if *self.peek() == Token::LParen { + self.next(); + let source = self.source(); + assert!(*self.peek() == Token::RParen, "a fragment atom names an existing binding"); + self.next(); + return source; + } + assert!(matches!(self.peek(), Token::Ident(_) | Token::ColonColon), "a fragment atom names an existing binding"); + let root = *self.peek() == Token::ColonColon; + if root { self.next(); } + let mut parts = vec![self.parse_ident()]; + while *self.peek() == Token::ColonColon { self.next(); parts.push(self.parse_ident()); } + if root || parts.len() > 1 { Expr::Path(parts, root) } else { Expr::Name(parts.remove(0)) } + } + fn fragment(&mut self) -> Fragment { + let mut atoms = Vec::new(); + loop { + let at = self.tokens[self.pos].start; + let source = self.source(); + self.expect(&Token::LParen); + let key_at = self.tokens[self.pos].start; + let key = self.fragment_pattern(); + self.expect(&Token::Comma); + let val = self.fragment_pattern(); + if atoms.is_empty() { key.check_lists(); } + val.check_lists(); + self.expect(&Token::RParen); + atoms.push(Atom { source, key, val, at, key_at }); + if *self.peek() != Token::Star { break; } + self.next(); + } + Fragment { atoms, scope: self.scope.clone() } + } + fn fragment_pattern(&mut self) -> Pattern { + match self.next() { + Token::LParen => { + let mut ps = Vec::new(); + if *self.peek() != Token::RParen { + loop { + ps.push(self.fragment_pattern()); + if *self.peek() != Token::Comma { break; } + self.next(); + } + } + self.expect(&Token::RParen); + Pattern::Tuple(ps) + } + Token::Ident(n) if n == "_" => Pattern::Wild, + Token::Ident(mut n) => { + if *self.peek() == Token::ColonColon { self.next(); n += &format!("::{}", self.parse_ident()); } + let scope = self.lambda_scope(); + let constructor = scope.variant(&n).map(|(tag, ty)| { + let empty = matches!(scope.variants(ty).unwrap()[tag].1.as_ref(), Some(corgi::Shape::Unit)); + (tag, ty.to_string(), empty) + }); + // An ambiguous declared constructor must not silently become a variable either. + let known = constructor.is_ok() || n.contains("::") || self.types.iter().any(|(_, vs)| vs.iter().any(|(v, _)| v == &n)); + if *self.peek() == Token::LParen || known { + let (tag, ty, empty) = constructor.unwrap_or_else(|e| panic!("fragment constructor `{n}`: {e}")); + let payload = if *self.peek() == Token::LParen { + self.next(); + let p = if *self.peek() == Token::RParen { Pattern::Tuple(vec![]) } else { self.fragment_pattern() }; + self.expect(&Token::RParen); + p + } else { + assert!(empty, "fragment constructor `{n}` requires a payload; write `{n}(pattern)`"); + Pattern::Tuple(vec![]) + }; + Pattern::Variant { name: format!("{ty}::{}", n.rsplit("::").next().unwrap()), tag, payload: Box::new(payload) } + } else { + Pattern::Var(n) + } + } + Token::Int(n) => Pattern::Literal(n.to_string()), + Token::Minus => match self.next() { Token::Int(n) => Pattern::Literal(format!("-{n}")), t => panic!("expected integer after -, got {t:?}") }, + Token::Str(s) => Pattern::Literal(format!("[{}]", s.bytes().map(|b| b.to_string()).collect::>().join(", "))), + Token::LBracket => { + let mut xs = Vec::new(); + if *self.peek() != Token::RBracket { + loop { + let x = self.fragment_pattern(); + assert!(x.constant(), "a list pattern is a literal"); + xs.push(x.text()); + if *self.peek() != Token::Comma { break; } + self.next(); + } + } + self.expect(&Token::RBracket); + Pattern::Literal(format!("[{}]", xs.join(", "))) + } + t => panic!("expected fragment pattern, got {t:?}"), + } + } +} +fn tuple(xs: &[String]) -> String { format!("({})", xs.join(", ")) } +// Emit a join's next key directly, as the design specifies. This is syntax +// composition, with no demand analysis or choice of index. +fn project(input: Expr, map: Lambda) -> Expr { + match input { + Expr::Join(left, right, projection) => Expr::Join(left, right, + Lambda::compose("row", &[(&projection, "row".into()), (&map, "_call0".into())], |r| r[1].clone())), + other => Expr::Map(Box::new(other), map), + } +} + +/// Logical fields stay fixed while checkpoints change their physical locations. +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +enum Access { + Slot(usize), Field(Box, usize), Payload(Box, usize), +} + +/// A lookup key's fields are read from k, including constructed variant payloads. +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +enum KeyRead { + Value(String), Field(Box, usize), Payload(Box, usize), +} +impl KeyRead { + fn scalar(&self) -> Option { + match self { + Self::Value(s) => Some(s.clone()), + Self::Field(a, i) => Some(format!("{}.{i}", a.scalar()?)), + Self::Payload(..) => None, + } + } + fn bind(&self, bindings: &mut Vec<(String, String)>, names: &mut BTreeMap) -> String { + if let Some(name) = names.get(self) { return name.clone(); } + match self { + Self::Value(s) => s.clone(), + Self::Field(a, i) => format!("{}.{i}", a.bind(bindings, names)), + Self::Payload(a, tag) => { + let source = a.bind(bindings, names); + let name = format!("key_payload{}", bindings.len()); + bindings.push((format!("{source} enlist unweave .{}", tag + 1), name.clone())); + names.insert(self.clone(), name.clone()); + name + } + } + } +} + +enum Test { Equal(Access, Access), Literal(Access, String) } +#[derive(Clone)] +struct Variant { at: Access, tag: usize, shape: String } +#[derive(Clone, Copy)] +enum Target<'a> { Key(&'a Pattern), Current, Output } +struct State<'a> { + scope: &'a corgi::Scope, + next: usize, + names: Vec, + bound: BTreeMap, + locations: BTreeMap, + key: BTreeMap, + variants: Vec, + tests: Vec, +} +impl State<'_> { + fn lambda(&self, pat: &str, body: String) -> Lambda { Lambda::new(format!("let {pat} = input in {body}"), self.scope) } + fn read(&self, at: &Access) -> String { + if let Some(expr) = self.locations.get(at) { return expr.clone(); } + match at { + Access::Field(a, i) => format!("let field = {} in field.{i}", self.read(a)), + _ => panic!("fragment payload or field is not bound: {at:?}"), + } + } + fn ready(&self, at: &Access) -> bool { + self.locations.contains_key(at) || matches!(at, Access::Field(a, _) if self.ready(a)) + } + fn test(&self, test: &Test) -> String { + match test { + Test::Equal(a, b) => format!("({}, {}) eq", self.read(a), self.read(b)), + Test::Literal(a, lit) if lit == "[]" => format!("({} len, 0) eq", self.read(a)), + Test::Literal(a, lit) => format!("({}, {lit}) eq", self.read(a)), + } + } + fn pattern(&mut self, p: &Pattern, at: Access) { + match p { + Pattern::Wild => {}, + Pattern::Var(n) => { + if let Some(old) = self.bound.get(n) { self.tests.push(Test::Equal(old.clone(), at)); } + else { self.names.push(n.clone()); self.bound.insert(n.clone(), at); } + } + Pattern::Literal(lit) => self.tests.push(Test::Literal(at, lit.clone())), + Pattern::Tuple(ps) => for (i, p) in ps.iter().enumerate() { + self.pattern(p, Access::Field(Box::new(at.clone()), i)); + }, + Pattern::Variant { tag, payload, .. } => { + self.variants.push(Variant { at: at.clone(), tag: *tag, shape: payload.shape_pattern() }); + self.pattern(payload, Access::Payload(Box::new(at), *tag)); + } + } + } + fn ingest(&mut self, p: &Pattern, expr: String) { + if let Pattern::Tuple(ps) = p { + for (i, p) in ps.iter().enumerate() { self.ingest(p, format!("{expr}.{i}")); } + } else if !matches!(p, Pattern::Wild) { + let at = Access::Slot(self.next); + self.next += 1; + self.locations.insert(at.clone(), expr); + self.pattern(p, at); + } + } + fn validate_key(&self, p: &Pattern) { + fn mentions_bound(s: &State<'_>, p: &Pattern) -> bool { + match p { + Pattern::Var(n) => { + assert!(s.bound.contains_key(n), "fragment key variable `{n}` is not bound by an earlier atom"); + true + } + Pattern::Tuple(ps) => ps.iter().fold(false, |found, p| mentions_bound(s, p) || found), + Pattern::Variant { payload, .. } => mentions_bound(s, payload), + Pattern::Literal(_) => false, + Pattern::Wild => panic!("a later fragment key cannot contain a wildcard; use a variable bound by an earlier atom"), + } + } + assert!(mentions_bound(self, p), "a constant fragment key would move every incoming row to one key; write a cross join as plain DDP"); + p.check_lists(); + } + fn key_expr(&self, p: &Pattern) -> String { + match p { + Pattern::Var(n) => self.read(&self.bound[n]), + Pattern::Tuple(ps) => tuple(&ps.iter().map(|p| self.key_expr(p)).collect::>()), + Pattern::Variant { name, payload, .. } => format!("{} inject {name}", self.key_expr(payload)), + Pattern::Literal(s) => s.clone(), + _ => unreachable!("lookup keys were validated"), + } + } + fn key_locations(&self, p: &Pattern, expr: KeyRead, out: &mut BTreeMap) { + match p { + Pattern::Var(n) => { out.entry(self.bound[n].clone()).or_insert(expr); }, + Pattern::Tuple(ps) => for (i, p) in ps.iter().enumerate() { + self.key_locations(p, KeyRead::Field(Box::new(expr.clone()), i), out); + }, + Pattern::Variant { tag, payload, .. } => self.key_locations(payload, KeyRead::Payload(Box::new(expr), *tag), out), + Pattern::Literal(_) => {}, + _ => unreachable!("lookup keys were validated"), + } + } + /// Keep named fields and pending tests, retaining an enclosing sum only until it is bound. + fn required(&self) -> std::collections::BTreeSet { + fn resident(s: &State<'_>, at: &Access) -> Access { + if s.ready(at) || s.key.contains_key(at) { return at.clone(); } + match at { + Access::Field(a, _) | Access::Payload(a, _) => resident(s, a), + _ => panic!("missing fragment field {at:?}"), + } + } + self.inputs(&self.variants).into_iter().map(|at| resident(self, &at)).collect() + } + fn inputs(&self, variants: &[Variant]) -> Vec { + let mut inputs = self.bound.values().cloned().collect::>(); + for test in &self.tests { match test { + Test::Equal(a, b) => { inputs.push(a.clone()); inputs.push(b.clone()); } + Test::Literal(a, _) => inputs.push(a.clone()), + } } + inputs.extend(variants.iter().map(|v| v.at.clone())); + inputs + } + fn row(&mut self, key_expr: String, key: BTreeMap) -> String { + let carried = self.required().into_iter().filter(|a| !key.contains_key(a)).collect::>(); + let values = carried.iter().map(|a| self.read(a)).collect::>(); + let mut locations = key.iter().filter_map(|(a, r)| r.scalar().map(|r| (a.clone(), r))).collect::>(); + for (i, at) in carried.into_iter().enumerate() { locations.insert(at, format!("v.{i}")); } + self.locations = locations; + self.key = key; + format!("({key_expr}, {})", tuple(&values)) + } + fn layout(&mut self, target: Target<'_>) -> String { + match target { + Target::Output => { + let values = self.names.iter().map(|n| self.read(&self.bound[n])).collect::>(); + let key = if values.len() == 1 { values[0].clone() } else { tuple(&values) }; + format!("({key}, ())") + } + Target::Key(p) => { + self.validate_key(p); + let expr = self.key_expr(p); + let mut key = BTreeMap::new(); + self.key_locations(p, KeyRead::Value("k".into()), &mut key); + self.row(expr, key) + } + // Decoded key payloads remain scalar while a following plain filter uses them. + Target::Current => self.row("k".into(), self.key.iter().filter(|(_, r)| r.scalar().is_some()).map(|(a, r)| (a.clone(), r.clone())).collect()), + } + } + fn filters(&mut self, mut expr: Expr, pat: &str) -> Expr { + let mut ready = Vec::new(); + let tests = std::mem::take(&mut self.tests); + for t in tests { + let available = match &t { + Test::Equal(a, b) => self.ready(a) && self.ready(b), + Test::Literal(a, _) => self.ready(a), + }; + if available { ready.push(self.test(&t)); } else { self.tests.push(t); } + } + if !ready.is_empty() { + let test = if ready.len() == 1 { ready.pop().unwrap() } else { format!("{} all", tuple(&ready)) }; + expr = Expr::Filter(Box::new(expr), self.lambda(pat, test)); + } + expr + } + fn capture(&mut self, variants: &[Variant], payload: &Access, prefix: &str, preserve_key: bool) -> (String, String) { + fn nearest(s: &State<'_>, at: Access) -> Access { + if s.locations.contains_key(&at) { return at; } + match at { + Access::Field(a, _) | Access::Payload(a, _) => nearest(s, *a), + _ => panic!("missing fragment capture {at:?}"), + } + } + let mut inputs = self.inputs(variants).into_iter().map(|a| nearest(self, a)).collect::>(); + inputs.remove(payload); + let mut values = inputs.iter().map(|a| self.read(a)).collect::>(); + let mut names = inputs.into_iter().enumerate().map(|(i, a)| { + let name = format!("{prefix}_{i}"); + self.locations.insert(a, name.clone()); + name + }).collect::>(); + if preserve_key { values.insert(0, "k".into()); names.insert(0, "k".into()); } + (tuple(&values), tuple(&names)) + } + fn payload_lists(&mut self, variants: &[Variant], i: usize, target: Target<'_>) -> String { + let v = &variants[i]; + let list = format!("{} enlist unweave .{}", self.read(&v.at), v.tag + 1); + let payload = Access::Payload(Box::new(v.at.clone()), v.tag); + let name = format!("payload{i}"); + self.locations.insert(payload.clone(), name.clone()); + let (values, pat) = self.capture(&variants[i + 1..], &payload, &format!("capture{i}"), matches!(target, Target::Current)); + let last = i + 1 == variants.len(); + let body = if last { self.layout(target) } else { self.payload_lists(variants, i + 1, target) }; + let binder = if v.shape == "_" { name } else { format!("{name} : {}", v.shape) }; + let mapped = format!("({values}, {list}) cap_list map (({pat}, {binder}) -> {body})"); + if last { mapped } else { format!("{mapped} flatten .1") } + } + /// Finish one atom, before the next arrangement. A payload list can emit that next key. + fn checkpoint(&mut self, expr: Expr, target: Target<'_>, pat: &str) -> Expr { + if let Target::Key(p) = target { self.validate_key(p); } + let mut expr = self.filters(expr, pat); + let mut bindings = Vec::new(); + let mut names = BTreeMap::new(); + for (at, read) in &self.key { self.locations.insert(at.clone(), read.bind(&mut bindings, &mut names)); } + let variants = std::mem::take(&mut self.variants); + if variants.is_empty() && bindings.is_empty() { + let body = self.layout(target); + return project(expr, self.lambda(pat, body)); + } + let defer = !self.tests.is_empty(); + // A plain key remains readable by the following filter. Final layouts and + // constructed payload keys wait until that filter has consumed its fields. + let output = if defer && !matches!(target, Target::Key(p) if !p.has_variant()) { Target::Current } else { target }; + let mut list = !variants.is_empty(); + let mut body = if list { self.payload_lists(&variants, 0, output) } else { self.layout(output) }; + for (i, (payloads, name)) in bindings.iter().enumerate().rev() { + let mut context = vec!["k".into(), "v".into()]; + context.extend(bindings[..i].iter().map(|(_, n)| n.clone())); + let context = tuple(&context); + body = format!("({context}, {payloads}) cap_list map (({context}, {name}) -> {body}){}", if list { " flatten .1" } else { "" }); + list = true; + } + expr = Expr::FlatMap(Box::new(expr), self.lambda(pat, body)); + if defer { + expr = self.filters(expr, "(k, v)"); + assert!(self.tests.is_empty(), "fragment tests must be bound at their checkpoint"); + let body = self.layout(target); + expr = project(expr, self.lambda("(k, v)", body)); + } + expr + } +} + +impl Fragment { + fn expansion(&self, at: &Cell) -> Expr { + let mut s = State { scope: &self.scope, next: 0, names: vec![], bound: BTreeMap::new(), + locations: BTreeMap::new(), key: BTreeMap::new(), variants: vec![], tests: vec![] }; + let first = &self.atoms[0]; + s.ingest(&first.key, "k".into()); + s.key = s.locations.iter().map(|(a, expr)| (a.clone(), KeyRead::Value(expr.clone()))).collect(); + s.ingest(&first.val, "v".into()); + let direct = self.atoms.get(1).is_some_and(|next| first.key == next.key); + let mut expr = first.source.clone(); + if !direct { + at.set(self.atoms.get(1).map_or(first.at, |a| a.key_at)); + let target = self.atoms.get(1).map_or(Target::Output, |a| Target::Key(&a.key)); + expr = s.checkpoint(expr, target, &format!("(k : {}, v : {})", first.key.shape_pattern(), first.val.shape_pattern())); + } + for i in 1..self.atoms.len() { + let atom = &self.atoms[i]; + at.set(atom.key_at); + s.validate_key(&atom.key); + s.ingest(&atom.val, "r".into()); + let body = s.row("k".into(), s.key.clone()); + let pat = if i == 1 && direct { + format!("(k : {}, v : {}, r : {})", first.key.shape_pattern(), first.val.shape_pattern(), atom.val.shape_pattern()) + } else { format!("(k, v, r : {})", atom.val.shape_pattern()) }; + expr = Expr::Join(Box::new(expr), Box::new(atom.source.clone()), s.lambda(&pat, body)); + let target = self.atoms.get(i + 1).map_or(Target::Output, |a| Target::Key(&a.key)); + at.set(self.atoms.get(i + 1).map_or(atom.at, |a| a.key_at)); + expr = s.checkpoint(expr, target, "(k, v)"); + } + expr + } +} diff --git a/interactive/tests/fragments.rs b/interactive/tests/fragments.rs new file mode 100644 index 000000000..ab1fd8475 --- /dev/null +++ b/interactive/tests/fragments.rs @@ -0,0 +1,215 @@ +//! Essential surface semantics and source diagnostics. +use interactive::{lower, parse, plan, passes::Config, server::Server, surface}; +fn program(src: &str) -> interactive::scope_ir::Program { + lower::lower_tree(parse::pipe::parse(&surface::expand(src).unwrap())) +} +fn report(src: &str) -> plan::Report { + plan::report(&surface::expand(src).unwrap(), &|_| None, &Config::all()).unwrap() +} +fn rows(src: &str, feeds: Vec<(usize, String)>) -> Vec<(String, String, i64)> { + let p = program(src); + timely::execute_directly(move |worker| { + let mut s = Server::new(); s.set_passes(Config::all()); s.install(worker, "p", &p).unwrap(); + for (i, data) in feeds { s.feed("p", i, &data, None).unwrap(); } + s.tick(worker); s.snapshot(worker, "p::out").unwrap().rows() + }) +} +fn rejects(src: &str, message: &str, token: &str) { + let error = surface::expand(src).unwrap_err(); + let before = &src[..src.rfind(token).unwrap()]; + let line = before.bytes().filter(|c| *c == b'\n').count() + 1; + let col = before.rsplit('\n').next().unwrap().chars().count() + 1; + assert!(error.starts_with(&format!("{line}:{col}: ")) && error.contains(message), "{error}"); +} + + +#[test] +fn joins_multiply_weights_and_handle_retractions() { + let src = r#" + let a = input 0 : (int, int); + let b = input 1 : (int, int); + let c = input 2 : ((int, int), ()); + output "out" = ((a))(x, y) * (b)(y, z) * c((x, z), ()) + | map(((x, y, z), ()) -> ((x, y, z), ())); + "#; + let p = program(src); + timely::execute_directly(move |worker| { + let mut s = Server::new(); + s.set_passes(Config::parse("all").unwrap()); + s.install(worker, "p", &p).unwrap(); + for (i, text) in [(0,"1 val=2 diff=2"),(1,"2 val=3 diff=3"),(2,"1,3 diff=4")] { s.feed("p",i,text,None).unwrap(); } + s.tick(worker); + assert_eq!(s.snapshot(worker,"p::out").unwrap().rows(), vec![("tuple(1,2,3)".into(),"tuple()".into(),24)]); + s.feed("p",0,"1 val=2 diff=-1",None).unwrap(); s.tick(worker); + assert_eq!(s.snapshot(worker,"p::out").unwrap().rows()[0].2,12); + s.feed("p",1,"2 val=3 diff=-3",None).unwrap(); s.tick(worker); + assert!(s.snapshot(worker,"p::out").unwrap().is_empty()); + }); +} + +#[test] +fn first_literal_precedes_the_first_rekey_and_arrangement() { + use interactive::{ir::LinearOp, scope_ir::{Item, Node, Ref}}; + let src = r#" + let message = input 0 : (int, (int,int,int)); + output "out" = message(_, (1, who, parent)) * message(parent, (_, to, _)); + "#; + let p = program(src); + let items = &p.root.items; + let filter = items.iter().position(|i| matches!(i, Item::Op(Node::Linear { ops, .. }) if matches!(ops.as_slice(), [LinearOp::Filter(_)]))).unwrap(); + let join = items.iter().position(|i| matches!(i, Item::Op(Node::Join { .. }))).unwrap(); + let Item::Op(Node::Join { left: Ref::Local(arrange), .. }) = &items[join] else { panic!() }; + let Item::Op(Node::Arrange(Ref::Local(map))) = &items[*arrange] else { panic!() }; + let Item::Op(Node::Linear { input: Ref::Local(input), ops }) = &items[*map] else { panic!() }; + assert_eq!(*input, filter); + assert!(filter < *map && *map < *arrange && *arrange < join); + let LinearOp::Project(lambda) = &ops[0] else { panic!() }; + assert!(!format!("{lambda:?}").contains("v.0"), "tested kind must not reach the arrangement"); + assert_eq!(rows(src, vec![(0, "10 val=tuple(1,2,20)\n11 val=tuple(0,3,20)\n20 val=tuple(0,4,0)".into())]), + vec![("tuple(2,20,4)".into(), "tuple()".into(), 1)]); +} + +#[test] +fn first_atom_tests_do_not_prevent_direct_arrangement_reuse() { + use interactive::scope_ir::{Item, Node, Ref}; + let src = r#" + let a=input 0 : (int,(Option(int),int,int)); + let b=input 1 : (int,int); + let c=input 2 : (int,int); + output "out"=a(x,(Some(y),x,7)) * b(x,z) * c(z,w); + "#; + let report = report(src); + let root = &report.stages[0].program.root; + let left = root.items.iter().find_map(|item| match item { + Item::Op(Node::Join { left, .. }) => Some(left), _ => None, + }).unwrap(); + let Ref::Local(i) = left else { panic!("expected arrangement") }; + assert!(matches!(&root.items[*i], Item::Op(Node::Arrange(Ref::Import(0))))); + let lowered = report.emit_ddp_at(plan::ddp::DdpStage::Lowered).unwrap().text; + assert_eq!(lowered.matches("| filter(").count(), 1); + assert_eq!(lowered.matches("| flatmap(").count(), 1); + assert!(lowered.find("| join(").unwrap() < lowered.find("| filter(").unwrap()); + assert!(lowered.find("| filter(").unwrap() < lowered.find("| flatmap(").unwrap()); + assert!(lowered.find("| flatmap(").unwrap() < lowered.rfind("| join(").unwrap()); + assert_eq!(rows(src, vec![ + (0,"1 val=tuple(inject(1,9),1,7)\n1 val=tuple(inject(0,tuple()),1,7)\n1 val=tuple(inject(1,8),2,7)\n1 val=tuple(inject(1,8),1,6)".into()), + (1,"1 val=2 diff=3".into()),(2,"2 val=4 diff=2".into()), + ]), vec![("tuple(1,9,2,4)".into(),"tuple()".into(),6)]); +} + +#[test] +fn later_literals_follow_their_join_and_leave_before_the_next_arrangement() { + use interactive::{ir::LinearOp, scope_ir::{Item, Node, Ref}}; + let src = r#" + let asked = input 0 : (int,(int,int)); + let by_name = input 1 : (int,int); + let place = input 2 : (int,(int,int)); + output "out" = asked(xname,(rid,yname)) * by_name(xname,x) + * place(x,(7,_)) * by_name(yname,y) * place(y,(7,_)); + "#; + let p = program(src); + let items = &p.root.items; + let joins: Vec<_> = items.iter().enumerate().filter_map(|(i,n)| matches!(n, Item::Op(Node::Join { .. })).then_some(i)).collect(); + let filters: Vec<_> = items.iter().enumerate().filter_map(|(i,n)| matches!(n, Item::Op(Node::Linear { ops, .. }) if matches!(ops.as_slice(), [LinearOp::Filter(_)] )).then_some(i)).collect(); + assert_eq!(filters.len(), 2); + assert!(joins[1] < filters[0] && filters[0] < joins[2] && joins[3] < filters[1]); + let Item::Op(Node::Linear { input, .. }) = &items[filters[0]] else { panic!() }; + assert_eq!(input, &Ref::Local(joins[1])); + let report = report(src); + let shapes = &report.stages[0].shapes; + let Item::Op(Node::Join { left: Ref::Local(arrange), .. }) = &items[joins[2]] else { panic!() }; + let before = shapes.of(&Ref::Local(filters[0])).unwrap(); + let after = shapes.of(&Ref::Local(*arrange)).unwrap(); + let (corgi::Shape::Prod(before), corgi::Shape::Prod(after)) = (&before.val, &after.val) else { panic!() }; + assert_eq!(before.len(), after.len() + 1, "discard the tested kind before arranging"); +} + +#[test] +fn later_keys_combine_bound_variables_with_constants() { + let src = r#" + enum K = Group (int,int) | Empty; + let a = input 0 : (int,()); + let b = input 1 : ((int,int),int); + let c = input 2 : (K,int); + let d = input 3 : ((int,K,()),int); + output "out" = a(rid,()) * b((rid,7),v) * c(Group((rid,v)),w) * d((w,Empty,()),z); + "#; + assert_eq!(rows(src, vec![(0,"1\n2".into()), (1,"1,7 val=10\n2,8 val=20".into()), + (2,"inject(0,tuple(1,10)) val=99".into()), (3,"tuple(99,inject(1,tuple()),tuple()) val=5".into())]), + vec![("tuple(1,10,99,5)".into(), "tuple()".into(), 1)]); + for (key, error, token) in [ + ("7", "a constant fragment key would move every incoming row to one key; write a cross join as plain DDP", "7"), + ("missing", "fragment key variable `missing` is not bound by an earlier atom", "missing"), + ("_", "a later fragment key cannot contain a wildcard", "_"), + ] { + let src = format!("-- π\nlet a=input 0 : (int,int); let b=a;\noutput \"out\"=a(x,y)*b({key},z);"); + rejects(&src, error, token); + } +} + +#[test] +fn variant_payload_rekeys_before_its_join_without_head_or_tag_columns() { + use interactive::{ir::LinearOp, scope_ir::{Item, Node, Ref}}; + let src = r#" + enum Member = User int | Group int; + let members = input 0 : (int, Member); + let groups = input 1 : (int, Member); + output "out" = members(doc, Group(g)) * groups(g, m); + "#; + let p = program(src); + let items = &p.root.items; + let join = items.iter().position(|i| matches!(i, Item::Op(Node::Join { .. }))).unwrap(); + let Item::Op(Node::Join { left: Ref::Local(arrange), .. }) = &items[join] else { panic!() }; + let Item::Op(Node::Arrange(Ref::Local(flatmap))) = &items[*arrange] else { panic!() }; + let Item::Op(Node::Linear { ops, .. }) = &items[*flatmap] else { panic!() }; + let [LinearOp::FlatMap(lambda)] = ops.as_slice() else { panic!() }; + let text = format!("{lambda:?}"); + assert!(text.contains("unweave .2")); + assert!(!text.contains(" head") && !text.contains(" match ("), "{text}"); + assert!(!items.iter().any(|i| matches!(i, Item::Op(Node::Linear { ops, .. }) if ops.iter().any(|op| matches!(op, LinearOp::Filter(_)))))); + assert_eq!(rows(src, vec![(0, "100 val=inject(1,10)\n200 val=inject(0,20)".into()), (1, "10 val=inject(0,3)".into())]), + vec![("tuple(100,10,inject(0,3))".into(), "tuple()".into(), 1)]); + assert_eq!(rows("let a=input 0 : (int,Option(int)); output \"out\"=a(x,None);", + vec![(0,"1 val=inject(0,tuple()) diff=3\n2 val=inject(1,7)".into())]), + vec![("1".into(),"tuple()".into(),3)]); + rejects("let a=input 0 : (int,Option(int)); output \"out\"=a(x,Some);", + "fragment constructor `Some` requires a payload; write `Some(pattern)`", "Some"); + rejects("enum A=Shared; enum B=Shared; let a=input 0 : (int,A); output \"out\"=a(x,Shared);", + "fragment constructor `Shared`: 'Shared' is a variant of", "Shared"); +} + +#[test] +fn fragments_read_their_own_feedback_variable() { + let src = r#" + let edges = input 0 : (int,int); + let roots = input 1 : (int,()); + s: { + set x = roots + (x(src,()) * edges(src,dst) | map(((src,dst),()) -> (dst,()))) | distinct; + } + output "out" = s::x; + "#; + assert_eq!(rows(src, vec![(0,"1 val=2\n2 val=3\n3 val=2\n4 val=5\n9 val=10".into()), (1,"1\n4".into())]), + (1..=5).map(|n| (n.to_string(), "tuple()".into(), 1)).collect::>()); +} + +#[test] +fn splices_preserve_plain_text_and_report_pattern_errors() { + let prefix = "-- π: a(x,y) * b(y,z)\nenum K=SomeK int | Empty;\nlet a=input 0 : (int,Option(int));\nlet plain=a | map((k,v) -> let unused = \"a(x,y) * text\" in (k,v));\n"; + assert_eq!(surface::expand(prefix).unwrap(), prefix); + let suffix = " | map(((x,y,z),()) -> (x,y)); -- untouched\n"; + let text = format!("{prefix}output \"out\" = a(x,Some(y)) -- inside\n * a(x,z){suffix}"); + let expanded = surface::expand(&text).unwrap(); + assert!(expanded.starts_with(prefix) && expanded.ends_with(suffix)); + assert!(expanded.contains("-- inside\n")); + assert!(!expanded.contains("a(x,Some(y))")); + assert!(report(&text).stages.iter().all(|s| s.problems.is_empty())); + for (pattern, message, token) in [ + ("input 0(x,y)", "a fragment atom names an existing binding", "0"), + ("a(x)", "Expected Comma, got RParen", ")"), + ("a(x,[y])", "a list pattern is a literal", "y"), + ("a(x,[[]])", "DDIR fragment list literal `[[]]` needs a complete element shape", "]"), + ] { + let text = format!("let a=input 0 : (int,List(List(int)));\noutput \"π\"={pattern};"); + rejects(&text, message, token); + } +}