From 374ad917ed34c439801d7965b21457f6431734f1 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 14 Nov 2026 07:39:24 -0500 Subject: [PATCH 1/5] DDIR: join fragments, `a(x, y) * b(y, z)`, an experimental surface form (spike) A join fragment is a pipeline source made of atoms joined by `*`. Fragments are an experimental surface form, not part of the plan language or a stable interface: they expand to plain DDP before path resolution and lowering, and no later stage or emitted plan contains them. Each atom names an existing binding and gives a pattern for its key and its value. Shared variables become join keys, and literals, variants and repeated variables become tests. A bare constructor such as `None` means its empty payload; an unknown name is a variable. The rows are every bound variable once, in order of first appearance, as the key, with a unit value; a following `map` builds the head. The expansion is fixed by the text, with no planner. Atoms join in written order, each against its binding's own arrangement: one new arrangement per `*`, of the rows so far, and none at the first `*` when the first two keys are written the same. Every later key must mention a variable bound earlier, so no join moves every row to one constant key; a cross join is written as plain DDP. Key variables are read from the join's key slot. Each literal or equality test is a plain `filter` as soon as its fields are bound, and tests never form arrangements. A variant binds its payload through `enlist unweave` in a `flatmap`, which can also produce the next join's key. When nothing is left to bind, the output layout goes into the last join's lambda. Co-Authored-By: GPT-6 Sol (Codex) Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01H6wB4H99TLw4tqu4x48pXL --- interactive/src/lower.rs | 4 + interactive/src/parse/mod.rs | 2 + interactive/src/parse/pipe.rs | 42 ++- interactive/src/parse/pipe/fragment.rs | 480 +++++++++++++++++++++++++ interactive/src/plan/ddp/parsed.rs | 1 + interactive/tests/fragments.rs | 477 ++++++++++++++++++++++++ 6 files changed, 1002 insertions(+), 4 deletions(-) create mode 100644 interactive/src/parse/pipe/fragment.rs create mode 100644 interactive/tests/fragments.rs diff --git a/interactive/src/lower.rs b/interactive/src/lower.rs index 7ff7c21de..4eb4768c5 100644 --- a/interactive/src/lower.rs +++ b/interactive/src/lower.rs @@ -196,6 +196,7 @@ fn resolve_expr(expr: &mut Expr, here: &[String], item: Option<&str>, resolve_expr(r, here, item, scopes, plan); } Expr::Concat(parts) => for part in parts { resolve_expr(part, here, item, scopes, plan); }, + Expr::Fragment(_) => unreachable!("fragments expand before path resolution"), Expr::Input(..) | Expr::Import(_) | Expr::Name(_) | Expr::ResolvedPath { .. } => {}, } } @@ -245,6 +246,7 @@ fn expr_free_names(expr: &Expr, out: &mut BTreeSet) { Expr::Input(..) | Expr::Import(_) => {}, Expr::Name(n) => { out.insert(RefName::Plain(n.clone())); }, Expr::ResolvedPath { .. } => {}, + Expr::Fragment(_) => unreachable!("fragments expand before lowering"), Expr::Path(..) => unreachable!("paths must be resolved before lowering"), Expr::TypedSource(e, _, _) | Expr::Map(e, _) | Expr::Reduce(e, _) | Expr::Filter(e, _) | Expr::Negate(e) | Expr::LiftIter(e) @@ -303,6 +305,7 @@ pub(crate) fn is_arranged(input: &st::Ref, items: &[st::Item]) -> bool { } pub fn lower_tree(mut stmts: Vec) -> st::Program { + crate::parse::pipe::fragment::expand(&mut stmts); let mut scopes = HashMap::new(); // The program root has no source-level name; `::` is its only explicit anchor. let mut visible = HashMap::new(); @@ -381,6 +384,7 @@ impl ScopeLower { .unwrap_or_else(|| panic!("unresolved name `{}`", name)), Expr::ResolvedPath { anchor, target, field } => self.env.get(&RefName::Path(PathKey { anchor: anchor.clone(), target: target.clone(), field: field.clone() })).cloned() .unwrap_or_else(|| panic!("unresolved path `::{}::{}`", target.join("::"), field)), + Expr::Fragment(_) => unreachable!("fragments expand before lowering"), Expr::Path(..) => unreachable!("paths must be resolved before lowering"), Expr::Map(e, p) => { let r = self.lower_expr(e); self.push(st::Node::Linear { input: r, ops: vec![LinearOp::Project(p.clone())] }) }, Expr::Filter(e, c) => { let r = self.lower_expr(e); self.push(st::Node::Linear { input: r, ops: vec![LinearOp::Filter(c.clone())] }) }, diff --git a/interactive/src/parse/mod.rs b/interactive/src/parse/mod.rs index e1ed9b063..793a8cb9b 100644 --- a/interactive/src/parse/mod.rs +++ b/interactive/src/parse/mod.rs @@ -53,6 +53,8 @@ pub enum Expr { Path(Vec, bool), /// A path checked against the scope tree before lowering. ResolvedPath { anchor: Vec, target: Vec, field: String }, + /// A join fragment, expanded to ordinary collection syntax before IR lowering. + Fragment(Box), Map(Box, Lambda), Join(Box, Box, Lambda), /// `stream | half_join(arrangement, strict, projection)`: see `scope_ir::Node::HalfJoin`. diff --git a/interactive/src/parse/pipe.rs b/interactive/src/parse/pipe.rs index f8873617e..3e5a9a334 100644 --- a/interactive/src/parse/pipe.rs +++ b/interactive/src/parse/pipe.rs @@ -62,6 +62,30 @@ //! and each stage applies to everything to its left: `a | map(f) + b | distinct` is //! `((a | map(f)) + b) | distinct`. To add a pipeline, parenthesize it or put it first. //! +//! Join fragments are an experimental surface form, not part of the plan language or a stable +//! interface. They expand to plain DDP before path resolution and lowering; no later stage or +//! emitted plan contains them. They may change or be removed. +//! A fragment is a pipeline source: `a(x, y) * b(y, z) | map(((x, y, z), ()) -> (x, z))`. +//! Atoms name existing bindings (including scope paths); they always have key and value arguments. +//! Every later key must mention at least one variable bound by an earlier atom; it may combine +//! earlier variables with literals, tuples, unit and constructors, but not wildcards or new variables. +//! A constant key would move every incoming row to one key; write a cross join as plain DDP. +//! Atoms keep their existing arrangements: one new arrangement per `*`, of the rows so far; +//! none at the first `*` when the first two keys are written the same. +//! The first key and every value accept variables, `_`, tuples, integer/list/string literals, +//! and constructors with payload patterns, such as `Some(x)` or `Type::Variant(x)`. +//! A bare constructor such as `None` means its empty payload; a constructor needing a payload +//! must supply one. A name unknown as a constructor is a variable. +//! An empty list pattern `[]` tests emptiness; inside a list literal its element shape must +//! come from sibling constants, otherwise DDIR rejects the literal with a shape error. +//! Reusing a variable tests equality. Plain filters run as soon as their fields are bound, +//! before the next rekey; a direct first join precedes its first atom's tests. Variants bind +//! through typed payload lists in `flatmap`, which can also emit the next lookup key. +//! Rows contain each variable once, in first-appearance order, keyed by their tuple (one +//! variable is scalar, none is unit), with unit value. A test-free final layout composes into +//! the last join's projection. Weights multiply; nothing implies distinct. +//! The parsed surface view retains fragments; all emitted plan stages contain plain DDP. +//! //! Statements: `let x = …;`, `set x = …;` (a feedback variable in a named //! scope), `name: { … }` (a nested scope), `output "name" = …;` (a //! program output, root scope only). An optional `var x;` or @@ -117,6 +141,8 @@ use super::*; use crate::ir::Lambda; +pub mod fragment; + #[derive(Debug, Clone, PartialEq)] enum Token { Let, Var, Set, Output, Type, Enum, @@ -126,7 +152,7 @@ enum Token { Lambda { pat: String, body: String }, LParen, RParen, LBrace, RBrace, Comma, Semi, Colon, ColonColon, Eq, - Pipe, Plus, Minus, Eof, + Pipe, Plus, Minus, Star, LBracket, RBracket, Eof, } /// A shape before names in type declarations have been resolved. @@ -266,6 +292,9 @@ fn lex(input: &str) -> Vec { ';' => { chars.next(); tokens.push(Token::Semi); }, '|' => { chars.next(); tokens.push(Token::Pipe); }, '+' => { chars.next(); tokens.push(Token::Plus); }, + '*' => { chars.next(); tokens.push(Token::Star); }, + '[' => { chars.next(); tokens.push(Token::LBracket); }, + ']' => { chars.next(); tokens.push(Token::RBracket); }, '=' => { chars.next(); tokens.push(Token::Eq); }, '"' => { // String literal: the quoted name in `import "..."` / `output "..." = ...`. @@ -546,13 +575,13 @@ impl Parser { /// An atom followed by stages, applied left to right: `| op`, `+ atom` and `- atom`. A run of /// `+`/`-` stages is one `Concat`. fn parse_pipe_expr(&mut self) -> Expr { - let mut expr = self.parse_atom(); + let mut expr = self.parse_product(); let mut concat = false; loop { let part = match self.peek() { Token::Pipe => { self.next(); expr = self.parse_pipe_op(expr); concat = false; continue }, - Token::Plus => { self.next(); self.parse_atom() }, - Token::Minus => { self.next(); Expr::Negate(Box::new(self.parse_atom())) }, + Token::Plus => { self.next(); self.parse_product() }, + Token::Minus => { self.next(); Expr::Negate(Box::new(self.parse_product())) }, _ => break, }; expr = match expr { @@ -564,6 +593,11 @@ impl Parser { expr } + fn parse_product(&mut self) -> Expr { + let source = self.parse_atom(); + if *self.peek() == Token::LParen { self.fragment(source) } else { source } + } + fn parse_atom(&mut self) -> Expr { match self.peek().clone() { Token::Input => { diff --git a/interactive/src/parse/pipe/fragment.rs b/interactive/src/parse/pipe/fragment.rs new file mode 100644 index 000000000..685dff0d9 --- /dev/null +++ b/interactive/src/parse/pipe/fragment.rs @@ -0,0 +1,480 @@ +//! Deliberately ordered, arrangement-preserving source-to-source expansion. +//! Tests and payload bindings run at the first legal checkpoint, before the next rekey. + +use super::*; +use std::collections::BTreeMap; + +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +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()), + } + } +} + +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] +struct Atom { source: Expr, key: Pattern, val: Pattern } + +#[derive(Clone, serde::Serialize, serde::Deserialize)] +pub struct Fragment { + pub(crate) text: String, + atoms: Vec, + // Scope is compilation context, like the scope captured in ordinary parsed lambdas. + #[serde(skip)] + scope: corgi::Scope, +} +impl std::fmt::Debug for Fragment { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { self.text.fmt(f) } +} + +impl Parser { + pub(super) fn fragment(&mut self, source: Expr) -> Expr { + let mut atoms = Vec::new(); + let mut source = source; + loop { + assert!(matches!(source, Expr::Name(_) | Expr::Path(..)), "a fragment atom names an existing binding"); + self.expect(&Token::LParen); + 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 }); + if *self.peek() != Token::Star { break; } + self.next(); + source = self.parse_atom(); + } + let text = atoms.iter().map(|a| { + let name = match &a.source { + Expr::Name(n) => n.clone(), + Expr::Path(ps, root) => format!("{}{}", if *root { "::" } else { "" }, ps.join("::")), + _ => unreachable!(), + }; + format!("{name}({}, {})", a.key.text(), a.val.text()) + }).collect::>().join(" * "); + Expr::Fragment(Box::new(Fragment { text, atoms, scope: self.lambda_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.values().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) -> 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 { + 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]; + 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)); + expr = s.checkpoint(expr, target, "(k, v)"); + } + expr + } +} + +/// Rewrite syntax before scope/path resolution; no arrangement or demand decisions here. +pub(crate) fn expand(stmts: &mut [Stmt]) { + fn expression(e: &mut Expr) { + match e { + Expr::Fragment(f) => { *e = f.expansion(); }, + Expr::TypedSource(e, _, _) | Expr::Map(e, _) | Expr::Reduce(e, _) | Expr::Filter(e, _) + | Expr::Negate(e) | Expr::LiftIter(e) | Expr::FlatMap(e, _) | Expr::Lift(e) + | Expr::JoinFunction(e, _) | Expr::Inspect(e, _) | Expr::Arrange(e) => expression(e), + Expr::Join(l, r, _) | Expr::HalfJoin(l, r, _, _) => { expression(l); expression(r); }, + Expr::Concat(es) => for e in es { expression(e); }, + _ => {}, + } + } + for stmt in stmts { + match stmt { + Stmt::Let(_, e) | Stmt::Set(_, e) | Stmt::Output(_, _, e) => expression(e), + Stmt::Scope(_, body) => expand(body), Stmt::Var(..) => {}, + } + } +} diff --git a/interactive/src/plan/ddp/parsed.rs b/interactive/src/plan/ddp/parsed.rs index 23185e9c4..2e6de9d94 100644 --- a/interactive/src/plan/ddp/parsed.rs +++ b/interactive/src/plan/ddp/parsed.rs @@ -58,6 +58,7 @@ impl Printer<'_> { Expr::Input(InputRef::Next(_, Some(name))) => format!("input {}", quoted(name)?), Expr::Import(name) => format!("import {}", quoted(name)?), Expr::TypedSource(source, key, val) => format!("{} : {}", self.expression(source)?, self.row(&(key.clone(), val.clone()))?), + Expr::Fragment(fragment) => fragment.text.clone(), Expr::Name(name) => name.clone(), Expr::Path(parts, absolute) => format!("{}{}", if *absolute { "::" } else { "" }, parts.join("::")), Expr::ResolvedPath { .. } => return Err("parsed view unexpectedly contains a lowered path".into()), diff --git a/interactive/tests/fragments.rs b/interactive/tests/fragments.rs new file mode 100644 index 000000000..dc299a3da --- /dev/null +++ b/interactive/tests/fragments.rs @@ -0,0 +1,477 @@ +//! Weighted semantics, restrictions, variants, and the two printable syntax views. +use interactive::{lower, parse, plan, passes::Config, server::Server}; + +fn program(src: &str) -> interactive::scope_ir::Program { lower::lower_tree(parse::pipe::parse(src)) } +fn rejects(src: &str, expected: &str) { + let panic = std::panic::catch_unwind(|| program(src)).expect_err("expected a fragment error"); + let message = panic.downcast_ref::().map(String::as_str) + .or_else(|| panic.downcast_ref::<&str>().copied()).expect("a string panic"); + assert!(message.contains(expected), "expected {expected:?}, got {message:?} for {src}"); +} +fn rows(src: &str, feeds: Vec<(usize, String)>) -> Vec<(String, String, i64)> { + let p = program(src); + timely::execute_directly(move |worker| { + let mut server = Server::new(); + server.set_passes(Config::parse("all").unwrap()); + server.install(worker, "p", &p).unwrap(); + for (i, text) in feeds { server.feed("p", i, &text, None).unwrap(); } + server.tick(worker); + server.snapshot(worker, "p::out").unwrap().rows() + }) +} + +#[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 patterns_tests_and_output_order() { + let src = r#" + let a = input 0 : ((int, int), (List(int), int)); + let b = input 1 : (int, int); + output "out" = a((x, x), ("hi", y)) * b(y, -7); + "#; + assert_eq!(rows(src,vec![(0,"1,1 val=tuple(list(104,105),2)\n1,9 val=tuple(list(104,105),2)\n3,3 val=tuple(list(104),2)".into()),(1,"2 val=-7 diff=3\n2 val=7".into())]),vec![("tuple(1,2)".into(),"tuple()".into(),3)]); + assert_eq!(rows("let a = input 0 : (int,int); output \"out\" = a(_, _);",vec![(0,"1 val=2 diff=4".into())]),vec![("tuple()".into(),"tuple()".into(),4)]); + assert_eq!(rows("let a = input 0 : (int,int); output \"out\" = a(x, _);",vec![(0,"1 val=2".into())]),vec![("1".into(),"tuple()".into(),1)]); +} + +#[test] +fn empty_list_pattern_does_not_choose_an_element_shape() { + assert_eq!(rows("let a = input 0 : (int,List((int,int))); output \"out\" = a(x, []);", + vec![(0,"1 val=list()\n2 val=list(tuple(3,4))".into())]), + vec![("1".into(),"tuple()".into(),1)]); +} + +#[test] +fn nested_open_variants_and_early_key_extraction() { + let src = r#" + enum A = Shared int | Empty; + enum B = Shared int | Other; + let a = input 0 : (int, Option(Result((int, List(int)), ()))); + let b = input 1 : (int, A); + let c = input 2 : (A, int); + output "out" = a(_, Some(Ok((x, text)))) * b(x, A::Shared(y)) * c(A::Shared(y), z); + "#; + let got = rows(src,vec![ + (0,"1 val=inject(1,inject(0,tuple(2,list(65))))\n2 val=inject(0,tuple())\n3 val=inject(1,inject(1,tuple()))".into()), + (1,"2 val=inject(0,3)\n2 val=inject(1,tuple())".into()), + (2,"inject(0,3) val=4\ninject(1,tuple()) val=9".into()), + ]); + assert_eq!(got,vec![("tuple(2,list(65),3,4)".into(),"tuple()".into(),1)]); + let report=plan::report(src,&|_|None,&Config::parse("all").unwrap()).unwrap(); + assert!(report.stages.last().unwrap().problems.is_empty()); + let text = report.emit_ddp().unwrap().text; + for tail in text.lines().filter(|line| line.contains("| flatmap(")) { + assert!(!tail.contains(" head"), "{tail}"); + } + let nested_key = r#" + enum Inner = Leaf int | Other; + enum Key = Wrap Inner | Empty; + let a = input 0 : (int, ()); + let b = input 1 : (Key, int); + output "out" = a(y, ()) * b(Key::Wrap(Inner::Leaf(y)), z); + "#; + assert_eq!(rows(nested_key, vec![(0, "1".into()), + (1, "inject(0,inject(0,1)) val=9".into())]), + vec![("tuple(1,9)".into(), "tuple()".into(), 1)]); + let report = plan::report(nested_key, &|_| None, &Config::parse("all").unwrap()).unwrap(); + let text = report.emit_ddp().unwrap().text; + assert!(!text.lines().filter(|line| line.contains("| flatmap(")).any(|line| line.contains(" head")), "{text}"); +} + +#[test] +fn paths_feedback_pipeline_union_and_difference() { + let src = r#" + let a = input 0 : (int, int); + looped: { var x : (int,int); set x = a; } + let b = ::looped::x; + output "out" = (looped::x(x,y) * b(x,z) | map(((x,y,z),()) -> (y,z))) + + (a(x,y) | map(((x,y),()) -> (x,y))) - a; + "#; + assert_eq!(rows(src,vec![(0,"1 val=2".into())]),vec![("2".into(),"2".into(),1)]); +} + +#[test] +fn key_restrictions_and_arity() { + for (atom, error) in [ + ("b(_,z)", "a later fragment key cannot contain a wildcard"), + ("b((x,_),z)", "a later fragment key cannot contain a wildcard"), + ("b(missing,z)", "fragment key variable `missing` is not bound by an earlier atom"), + ("b((x,missing),z)", "fragment key variable `missing` is not bound by an earlier atom"), + ("b(Some(_),z)", "a later fragment key cannot contain a wildcard"), + ] { + let src=format!("let a = input 0 : (int,int); let b = a; output \"out\" = a(x,y) * {atom};"); + rejects(&src, error); + } + for (bad, error) in [("a(x)", "Expected Comma, got RParen"), + ("a(x,y,z)", "Expected RParen, got Comma"), ("a(x,[y])", "a list pattern is a literal")] { + rejects(&format!("let a=input 0 : (int,int); output \"out\"={bad};"), error); + } +} + +#[test] +fn constant_later_keys_explain_cross_joins() { + for key in ["1", "-7", "\"ok\"", "[1,2]", "[]", "[[]]", "()", "(1,())", "Empty", "Empty()", "Group(7)", "(Empty(),[1,2])"] { + let src = format!("enum K=Group int | Empty; let a=input 0 : (int,int); let b=a; output \"out\"=a(x,y)*b({key},z);"); + rejects(&src, "a constant fragment key would move every incoming row to one key; write a cross join as plain DDP"); + } +} + +#[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)]); +} + +#[test] +fn bare_constructors_match_empty_payloads_and_do_not_bind_variables() { + 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)]); + let src = "let a=input 0 : (int,(Option(int),Option(int))); output \"out\"=a(x,(None,None));"; + assert_eq!(rows(src, vec![(0,"1 val=tuple(inject(0,tuple()),inject(0,tuple()))\n2 val=tuple(inject(0,tuple()),inject(1,7))\n3 val=tuple(inject(1,7),inject(1,7))".into())]), + vec![("1".into(), "tuple()".into(), 1)]); + let text = plan::report(src, &|_| None, &Config::all()).unwrap().emit_ddp().unwrap().text; + assert!(!text.contains("| filter("), "repeated None must not become equality: {text}"); + assert_eq!(rows("enum K=Empty | Full int; let a=input 0 : (K,int); output \"out\"=a(K::Empty,v);", + vec![(0,"inject(0,tuple()) val=9\ninject(1,7) val=8".into())]), + vec![("9".into(), "tuple()".into(), 1)]); + assert_eq!(rows("let a=input 0 : (int,int); output \"out\"=a(x,Fresh);", vec![(0,"1 val=2".into())]), + vec![("tuple(1,2)".into(), "tuple()".into(), 1)]); +} + +#[test] +fn bare_constructors_requiring_payloads_report_the_constructor() { + for name in ["Some", "Group", "K::Group", "Ok"] { + let src = format!("enum K=Group int | Empty; let a=input 0 : (int,K); output \"out\"=a(x,{name});"); + rejects(&src, &format!("fragment constructor `{name}` requires a payload; write `{name}(pattern)`")); + } + 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"); +} + +#[test] +fn nested_empty_list_literals_report_a_ddir_error() { + for literal in ["[[]]", "[[[]]]", "[([], 7)]"] { + let src = format!("let a=input 0 : (int,List(List(int))); output \"out\"=a(x,{literal});"); + rejects(&src, &format!("DDIR fragment list literal `{literal}` needs a complete element shape; use plain DDP with an explicit list shape")); + } + for (literal, value) in [("[[1],[2]]", "list(list(1),list(2))"), + ("[[1],[]]", "list(list(1),list())"), ("[[],[1]]", "list(list(),list(1))")] { + let src = format!("let a=input 0 : (int,List(List(int))); output \"out\"=a(x,{literal});"); + assert_eq!(rows(&src, vec![(0,format!("1 val={value}\n2 val=list()"))]), + vec![("1".into(), "tuple()".into(), 1)]); + } +} + +#[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 fragments_are_absent_from_every_emitted_plan_stage() { + use plan::ddp::DdpStage; + let src = "let a=input 0 : (int,int); let b=a; output \"out\"=a(x,7)*b(x,y);"; + let report = plan::report(src, &|_| None, &Config::all()).unwrap(); + assert!(report.emit_ddp_at(DdpStage::Parsed).unwrap().text.contains("a(x, 7) * b(x, y)")); + for stage in [DdpStage::Lowered, DdpStage::Optimized, DdpStage::Projected, DdpStage::AsOf] { + let text = report.emit_ddp_at(stage).unwrap().text; + assert!(!text.contains('*'), "{stage:?}: {text}"); + assert!(text.contains("| join("), "{stage:?}: {text}"); + } + assert!(!report.emit_ddp().unwrap().text.contains('*')); +} + +#[test] +fn parsed_retains_fragments_and_direct_first_tests_follow_the_first_join() { + use plan::ddp::DdpStage; + let src="let a = input 0 : (int,int); let b = a; let c = a; output \"out\" = a(x,7) * b(x,y) * c(y,9);"; + let report=plan::report(src,&|_|None,&Config::parse("all").unwrap()).unwrap(); + let parsed=report.emit_ddp_at(DdpStage::Parsed).unwrap().text; + assert!(parsed.contains("a(x, 7) * b(x, y) * c(y, 9)")); + let lowered=report.emit_ddp_at(DdpStage::Lowered).unwrap().text; + assert!(!lowered.contains('*')); + assert_eq!(lowered.matches("| join(").count(),2); + assert_eq!(lowered.matches("| filter(").count(), 2); + assert!(!lowered.contains("| flatmap(")); + assert!(lowered.find("| join(").unwrap() < lowered.find("| filter(").unwrap()); + assert!(lowered.find("| filter(").unwrap() < lowered.rfind("| join(").unwrap()); + let again=plan::report(&parsed,&|_|None,&Config::parse("all").unwrap()).unwrap(); + assert!(report.compare_final(&again).is_empty()); +} + +#[test] +fn open_result_constructor_key_exposes_corgis_missing_lane_shape() { + let src = "let a=input 0 : (int,()); let b=input 1 : (Result(int,()),int); output \"out\"=a(x,()) * b(Ok(x),y);"; + let report=plan::report(src,&|_|None,&Config::parse("all").unwrap()).unwrap(); + assert!(report.stages[0].problems.iter().any(|s|s.contains("declares no payload shape"))); +} + +#[test] +fn direct_first_join_reads_the_key_and_carries_it_when_rekeyed() { + use interactive::scope_ir::{Item, Node, Ref}; + let src = "let a=input 0 : (int,int); let b=input 1 : (int,int); let c=input 2 : (int,int); output \"out\"=a(x,y) * b(x,z) * c(y,w);"; + let report = plan::report(src, &|_| None, &Config::parse("all").unwrap()).unwrap(); + let root = &report.stages[0].program.root; + let (left, projection) = root.items.iter().find_map(|item| match item { + Item::Op(Node::Join { left, projection, .. }) => Some((left, projection)), _ => None, + }).unwrap(); + let Ref::Local(i) = left else { panic!("expected arrangement") }; + assert!(matches!(&root.items[*i], Item::Op(Node::Arrange(Ref::Import(0))))); + let source = format!("{projection:?}"); + let graph = corgi::parse_ml_in(&source[1..source.len()-1], &corgi::Scope::default()).unwrap(); + let got = corgi::eval_graph(&graph, corgi::Value::Prod(vec![ + corgi::Value::int([10]), corgi::Value::int([20]), corgi::Value::int([30]), + ])); + // y becomes the key. x leaves the key and must survive in the value with z. + assert_eq!(got, corgi::Value::Prod(vec![corgi::Value::int([20]), + corgi::Value::Prod(vec![corgi::Value::int([10]), corgi::Value::int([30])])])); + assert_eq!(rows(src, vec![(0,"10 val=20".into()),(1,"10 val=30".into()),(2,"20 val=40".into())]), + vec![("tuple(10,20,30,40)".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 = plan::report(src, &|_| None, &Config::parse("all").unwrap()).unwrap(); + 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 first_key_variants_and_repeated_variables_are_tested_after_direct_join() { + use interactive::scope_ir::{Item, Node, Ref}; + let src = r#" + enum Key = Group (int,int) | User (int,int); + let a=input 0 : (Key,int); + let b=input 1 : (Key,int); + output "out"=a(Group((x,x)),y) * b(Group((x,x)),z); + "#; + let report=plan::report(src,&|_|None,&Config::parse("all").unwrap()).unwrap(); + 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("| flatmap(").unwrap()); + assert!(lowered.find("| flatmap(").unwrap() < lowered.find("| filter(").unwrap()); + assert_eq!(rows(src,vec![ + (0,"inject(0,tuple(1,1)) val=8 diff=2\ninject(0,tuple(1,2)) val=7\ninject(1,tuple(2,2)) val=6".into()), + (1,"inject(0,tuple(1,1)) val=9 diff=3\ninject(0,tuple(1,2)) val=5\ninject(1,tuple(2,2)) val=4".into()), + ]),vec![("tuple(1,8,9)".into(),"tuple()".into(),6)]); +} + +#[test] +fn nested_variant_lists_and_equalities_build_only_survivors() { + let src = r#" + let a = input 0 : (int, Option(Result((int,int),()))); + output "out" = a(doc, Some(Ok((u,u)))); + "#; + assert_eq!(rows(src, vec![(0, "1 val=inject(1,inject(0,tuple(2,2))) diff=3\n1 val=inject(1,inject(0,tuple(2,9)))\n2 val=inject(0,tuple())\n3 val=inject(1,inject(1,tuple()))".into())]), + vec![("tuple(1,2)".into(), "tuple()".into(), 3)]); + let report = plan::report(src, &|_| None, &Config::parse("all").unwrap()).unwrap(); + let text = report.emit_ddp().unwrap().text; + assert_eq!(text.matches("| flatmap(").count(), 1, "{text}"); + assert!(!text.contains(" head") && !text.contains("[]") && !text.contains(" match ("), "{text}"); + let again = plan::report(&text, &|_| None, &Config::parse("all").unwrap()).unwrap(); + assert!(report.compare_final(&again).is_empty()); +} + +#[test] +fn one_atom_variant_demux() { + let src = r#" + enum Member = User int | Group int; + let grants = input 0 : (int, (Member, int)); + output "out" = grants(doc, (User(u), level)) | map(((doc, u, level), ()) -> (u, (doc, level))); + "#; + let feed = "100 val=tuple(inject(0,1),3)\n100 val=tuple(inject(1,10),2) diff=5\n200 val=tuple(inject(0,2),1) diff=2"; + assert_eq!(rows(src, vec![(0, feed.into())]), + vec![("1".into(), "tuple(100,3)".into(), 1), ("2".into(), "tuple(200,1)".into(), 2)]); +} + +#[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 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 = plan::report(src, &|_| None, &Config::all()).unwrap(); + 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 test_free_chain_has_no_flatmap_or_layout_stage_after_the_last_join() { + use interactive::{ir::LinearOp, scope_ir::{Item, Node}}; + let src = "let a=input 0 : (int,int); let b=a; let c=a; output \"out\"=a(x,y)*b(y,z)*c(z,w);"; + let p = program(src); + assert!(!p.root.items.iter().any(|i| matches!(i, Item::Op(Node::Linear { ops, .. }) if ops.iter().any(|op| matches!(op, LinearOp::FlatMap(_)))))); + assert!(matches!(p.root.items.last(), Some(Item::Op(Node::Join { .. })))); + assert_eq!(rows(src, vec![(0, "1 val=2\n2 val=3\n3 val=4".into())]), + vec![("tuple(1,2,3,4)".into(), "tuple()".into(), 1)]); +} + +#[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)]); +} + +#[test] +fn wildcard_payloads_still_check_their_tuple_shape() { + for (val, error) in [("Some((_,_))", "wants a product of 2 fields, not Int"), + ("Some((_))", "wants a product of 1 field, not Int")] { + let src = format!("let a=input 0 : (int,Option(int)); output \"out\"=a(_, {val});"); + let report = plan::report(&src, &|_| None, &Config::all()).unwrap(); + assert!(report.stages[0].problems.iter().any(|e| e.contains(error)), "{src}: {:?}", report.stages[0].problems); + } +} + +#[test] +fn variants_rekey_before_pending_plain_payload_tests() { + use interactive::{ir::LinearOp, scope_ir::{Export, Item, Node, Ref}}; + let src = r#" + enum Member = User int | Group (int,int); + let members = input 0 : (int,Member); + let groups = input 1 : (int,int); + output "out" = members(doc,Group((g,7))) * groups(g,m); + "#; + let mut p = program(src); + let flatmap = p.root.items.iter().position(|i| matches!(i, Item::Op(Node::Linear { ops, .. }) if matches!(ops.as_slice(), [LinearOp::FlatMap(_)]))).unwrap(); + // Observe the variant stage before its plain filter: its key is already g. + p.root.exports.push(Export { name: "keyed".into(), value: Ref::Local(flatmap), shape: None }); + timely::execute_directly(move |worker| { + let mut s = Server::new(); + s.set_passes(Config::all()); + s.install(worker, "p", &p).unwrap(); + s.feed("p", 0, "100 val=inject(1,tuple(10,7))\n200 val=inject(1,tuple(20,8))\n300 val=inject(0,30)", None).unwrap(); + s.feed("p", 1, "10 val=3\n20 val=4", None).unwrap(); + s.tick(worker); + assert_eq!(s.snapshot(worker, "p::keyed").unwrap().rows(), vec![ + ("10".into(), "tuple(100,7)".into(), 1), + ("20".into(), "tuple(200,8)".into(), 1), + ]); + assert_eq!(s.snapshot(worker, "p::out").unwrap().rows(), + vec![("tuple(100,10,3)".into(), "tuple()".into(), 1)]); + }); +} From eb0f63f8b02d55c062f18be3b7d6e04767fcabb3 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 14 Nov 2026 09:43:39 -0500 Subject: [PATCH 2/5] DDIR: expand join fragments as they are parsed, so no syntax tree holds one The parser now expands each fragment into plain DDP as soon as it reads it. `Expr::Fragment`, its arms in lowering and the printer, and the separate expansion walk before lowering are gone, and even the parsed view shows the expansion. The expansion itself is unchanged. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01H6wB4H99TLw4tqu4x48pXL --- interactive/src/lower.rs | 4 -- interactive/src/parse/mod.rs | 2 - interactive/src/parse/pipe.rs | 6 +-- interactive/src/parse/pipe/fragment.rs | 51 +++----------------------- interactive/src/plan/ddp/parsed.rs | 1 - interactive/tests/fragments.rs | 7 ++-- 6 files changed, 12 insertions(+), 59 deletions(-) diff --git a/interactive/src/lower.rs b/interactive/src/lower.rs index 4eb4768c5..7ff7c21de 100644 --- a/interactive/src/lower.rs +++ b/interactive/src/lower.rs @@ -196,7 +196,6 @@ fn resolve_expr(expr: &mut Expr, here: &[String], item: Option<&str>, resolve_expr(r, here, item, scopes, plan); } Expr::Concat(parts) => for part in parts { resolve_expr(part, here, item, scopes, plan); }, - Expr::Fragment(_) => unreachable!("fragments expand before path resolution"), Expr::Input(..) | Expr::Import(_) | Expr::Name(_) | Expr::ResolvedPath { .. } => {}, } } @@ -246,7 +245,6 @@ fn expr_free_names(expr: &Expr, out: &mut BTreeSet) { Expr::Input(..) | Expr::Import(_) => {}, Expr::Name(n) => { out.insert(RefName::Plain(n.clone())); }, Expr::ResolvedPath { .. } => {}, - Expr::Fragment(_) => unreachable!("fragments expand before lowering"), Expr::Path(..) => unreachable!("paths must be resolved before lowering"), Expr::TypedSource(e, _, _) | Expr::Map(e, _) | Expr::Reduce(e, _) | Expr::Filter(e, _) | Expr::Negate(e) | Expr::LiftIter(e) @@ -305,7 +303,6 @@ pub(crate) fn is_arranged(input: &st::Ref, items: &[st::Item]) -> bool { } pub fn lower_tree(mut stmts: Vec) -> st::Program { - crate::parse::pipe::fragment::expand(&mut stmts); let mut scopes = HashMap::new(); // The program root has no source-level name; `::` is its only explicit anchor. let mut visible = HashMap::new(); @@ -384,7 +381,6 @@ impl ScopeLower { .unwrap_or_else(|| panic!("unresolved name `{}`", name)), Expr::ResolvedPath { anchor, target, field } => self.env.get(&RefName::Path(PathKey { anchor: anchor.clone(), target: target.clone(), field: field.clone() })).cloned() .unwrap_or_else(|| panic!("unresolved path `::{}::{}`", target.join("::"), field)), - Expr::Fragment(_) => unreachable!("fragments expand before lowering"), Expr::Path(..) => unreachable!("paths must be resolved before lowering"), Expr::Map(e, p) => { let r = self.lower_expr(e); self.push(st::Node::Linear { input: r, ops: vec![LinearOp::Project(p.clone())] }) }, Expr::Filter(e, c) => { let r = self.lower_expr(e); self.push(st::Node::Linear { input: r, ops: vec![LinearOp::Filter(c.clone())] }) }, diff --git a/interactive/src/parse/mod.rs b/interactive/src/parse/mod.rs index 793a8cb9b..e1ed9b063 100644 --- a/interactive/src/parse/mod.rs +++ b/interactive/src/parse/mod.rs @@ -53,8 +53,6 @@ pub enum Expr { Path(Vec, bool), /// A path checked against the scope tree before lowering. ResolvedPath { anchor: Vec, target: Vec, field: String }, - /// A join fragment, expanded to ordinary collection syntax before IR lowering. - Fragment(Box), Map(Box, Lambda), Join(Box, Box, Lambda), /// `stream | half_join(arrangement, strict, projection)`: see `scope_ir::Node::HalfJoin`. diff --git a/interactive/src/parse/pipe.rs b/interactive/src/parse/pipe.rs index 3e5a9a334..e778f0331 100644 --- a/interactive/src/parse/pipe.rs +++ b/interactive/src/parse/pipe.rs @@ -63,8 +63,8 @@ //! `((a | map(f)) + b) | distinct`. To add a pipeline, parenthesize it or put it first. //! //! Join fragments are an experimental surface form, not part of the plan language or a stable -//! interface. They expand to plain DDP before path resolution and lowering; no later stage or -//! emitted plan contains them. They may change or be removed. +//! interface. The parser expands each one into plain DDP as it reads it, so no syntax tree, +//! later stage or emitted plan contains one. They may change or be removed. //! A fragment is a pipeline source: `a(x, y) * b(y, z) | map(((x, y, z), ()) -> (x, z))`. //! Atoms name existing bindings (including scope paths); they always have key and value arguments. //! Every later key must mention at least one variable bound by an earlier atom; it may combine @@ -84,7 +84,7 @@ //! Rows contain each variable once, in first-appearance order, keyed by their tuple (one //! variable is scalar, none is unit), with unit value. A test-free final layout composes into //! the last join's projection. Weights multiply; nothing implies distinct. -//! The parsed surface view retains fragments; all emitted plan stages contain plain DDP. +//! Even the parsed view shows the expansion. //! //! Statements: `let x = …;`, `set x = …;` (a feedback variable in a named //! scope), `name: { … }` (a nested scope), `output "name" = …;` (a diff --git a/interactive/src/parse/pipe/fragment.rs b/interactive/src/parse/pipe/fragment.rs index 685dff0d9..0a8269b08 100644 --- a/interactive/src/parse/pipe/fragment.rs +++ b/interactive/src/parse/pipe/fragment.rs @@ -4,7 +4,7 @@ use super::*; use std::collections::BTreeMap; -#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[derive(Clone, Debug, PartialEq, Eq)] enum Pattern { Var(String), Wild, Tuple(Vec), Literal(String), Variant { name: String, tag: usize, payload: Box }, @@ -48,20 +48,10 @@ impl Pattern { } } -#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] struct Atom { source: Expr, key: Pattern, val: Pattern } -#[derive(Clone, serde::Serialize, serde::Deserialize)] -pub struct Fragment { - pub(crate) text: String, - atoms: Vec, - // Scope is compilation context, like the scope captured in ordinary parsed lambdas. - #[serde(skip)] - scope: corgi::Scope, -} -impl std::fmt::Debug for Fragment { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { self.text.fmt(f) } -} +/// A fragment's atoms, expanded as soon as they are parsed: no syntax tree holds a fragment. +struct Fragment { atoms: Vec } impl Parser { pub(super) fn fragment(&mut self, source: Expr) -> Expr { @@ -81,15 +71,7 @@ impl Parser { self.next(); source = self.parse_atom(); } - let text = atoms.iter().map(|a| { - let name = match &a.source { - Expr::Name(n) => n.clone(), - Expr::Path(ps, root) => format!("{}{}", if *root { "::" } else { "" }, ps.join("::")), - _ => unreachable!(), - }; - format!("{name}({}, {})", a.key.text(), a.val.text()) - }).collect::>().join(" * "); - Expr::Fragment(Box::new(Fragment { text, atoms, scope: self.lambda_scope().clone() })) + Fragment { atoms }.expansion(self.lambda_scope()) } fn fragment_pattern(&mut self) -> Pattern { @@ -429,8 +411,8 @@ impl State<'_> { } impl Fragment { - fn expansion(&self) -> Expr { - let mut s = State { scope: &self.scope, next: 0, names: vec![], bound: BTreeMap::new(), + fn expansion(&self, scope: &corgi::Scope) -> Expr { + let mut s = State { 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()); @@ -457,24 +439,3 @@ impl Fragment { expr } } - -/// Rewrite syntax before scope/path resolution; no arrangement or demand decisions here. -pub(crate) fn expand(stmts: &mut [Stmt]) { - fn expression(e: &mut Expr) { - match e { - Expr::Fragment(f) => { *e = f.expansion(); }, - Expr::TypedSource(e, _, _) | Expr::Map(e, _) | Expr::Reduce(e, _) | Expr::Filter(e, _) - | Expr::Negate(e) | Expr::LiftIter(e) | Expr::FlatMap(e, _) | Expr::Lift(e) - | Expr::JoinFunction(e, _) | Expr::Inspect(e, _) | Expr::Arrange(e) => expression(e), - Expr::Join(l, r, _) | Expr::HalfJoin(l, r, _, _) => { expression(l); expression(r); }, - Expr::Concat(es) => for e in es { expression(e); }, - _ => {}, - } - } - for stmt in stmts { - match stmt { - Stmt::Let(_, e) | Stmt::Set(_, e) | Stmt::Output(_, _, e) => expression(e), - Stmt::Scope(_, body) => expand(body), Stmt::Var(..) => {}, - } - } -} diff --git a/interactive/src/plan/ddp/parsed.rs b/interactive/src/plan/ddp/parsed.rs index 2e6de9d94..23185e9c4 100644 --- a/interactive/src/plan/ddp/parsed.rs +++ b/interactive/src/plan/ddp/parsed.rs @@ -58,7 +58,6 @@ impl Printer<'_> { Expr::Input(InputRef::Next(_, Some(name))) => format!("input {}", quoted(name)?), Expr::Import(name) => format!("import {}", quoted(name)?), Expr::TypedSource(source, key, val) => format!("{} : {}", self.expression(source)?, self.row(&(key.clone(), val.clone()))?), - Expr::Fragment(fragment) => fragment.text.clone(), Expr::Name(name) => name.clone(), Expr::Path(parts, absolute) => format!("{}{}", if *absolute { "::" } else { "" }, parts.join("::")), Expr::ResolvedPath { .. } => return Err("parsed view unexpectedly contains a lowered path".into()), diff --git a/interactive/tests/fragments.rs b/interactive/tests/fragments.rs index dc299a3da..98df87f87 100644 --- a/interactive/tests/fragments.rs +++ b/interactive/tests/fragments.rs @@ -213,8 +213,7 @@ fn fragments_are_absent_from_every_emitted_plan_stage() { use plan::ddp::DdpStage; let src = "let a=input 0 : (int,int); let b=a; output \"out\"=a(x,7)*b(x,y);"; let report = plan::report(src, &|_| None, &Config::all()).unwrap(); - assert!(report.emit_ddp_at(DdpStage::Parsed).unwrap().text.contains("a(x, 7) * b(x, y)")); - for stage in [DdpStage::Lowered, DdpStage::Optimized, DdpStage::Projected, DdpStage::AsOf] { + for stage in [DdpStage::Parsed, DdpStage::Lowered, DdpStage::Optimized, DdpStage::Projected, DdpStage::AsOf] { let text = report.emit_ddp_at(stage).unwrap().text; assert!(!text.contains('*'), "{stage:?}: {text}"); assert!(text.contains("| join("), "{stage:?}: {text}"); @@ -223,12 +222,12 @@ fn fragments_are_absent_from_every_emitted_plan_stage() { } #[test] -fn parsed_retains_fragments_and_direct_first_tests_follow_the_first_join() { +fn parsed_shows_the_expansion_and_direct_first_tests_follow_the_first_join() { use plan::ddp::DdpStage; let src="let a = input 0 : (int,int); let b = a; let c = a; output \"out\" = a(x,7) * b(x,y) * c(y,9);"; let report=plan::report(src,&|_|None,&Config::parse("all").unwrap()).unwrap(); let parsed=report.emit_ddp_at(DdpStage::Parsed).unwrap().text; - assert!(parsed.contains("a(x, 7) * b(x, y) * c(y, 9)")); + assert!(!parsed.contains('*') && parsed.contains("| join(")); let lowered=report.emit_ddp_at(DdpStage::Lowered).unwrap().text; assert!(!lowered.contains('*')); assert_eq!(lowered.matches("| join(").count(),2); From ab266881b637f5b7eef9af02fdacd465710840fb Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 14 Nov 2026 09:40:33 -0500 Subject: [PATCH 3/5] DDIR: join fragments reuse the first atom's arrangement only when its key restricts nothing When the first two atoms' keys were written the same, the first join used the first atom's arrangement even if that key held tests, so rows the tests reject were paired before being dropped. With `a(Some(x), v) * b(Some(x), w)` and 10,000 `None` keys on each side, that paired every `None` with every `None`: 2.32 s, against 4.4 ms now. Now every `*` forms a new arrangement of the rows so far, except at the first `*` when the first two keys are the same unrestricted variables. A restricted first key (a literal, a variant, a repeated variable) is tested first, and the surviving rows are arranged under the same key values, so they stay on their workers. Tests on the first atom's value still follow a direct first join. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01H6wB4H99TLw4tqu4x48pXL --- interactive/src/parse/pipe.rs | 2 +- interactive/src/parse/pipe/fragment.rs | 14 +++++++++++++- interactive/tests/fragments.rs | 11 ++++++----- 3 files changed, 20 insertions(+), 7 deletions(-) diff --git a/interactive/src/parse/pipe.rs b/interactive/src/parse/pipe.rs index e778f0331..d829500be 100644 --- a/interactive/src/parse/pipe.rs +++ b/interactive/src/parse/pipe.rs @@ -71,7 +71,7 @@ //! earlier variables with literals, tuples, unit and constructors, but not wildcards or new variables. //! A constant key would move every incoming row to one key; write a cross join as plain DDP. //! Atoms keep their existing arrangements: one new arrangement per `*`, of the rows so far; -//! none at the first `*` when the first two keys are written the same. +//! none at the first `*` when the first two keys are the same unrestricted variables. //! The first key and every value accept variables, `_`, tuples, integer/list/string literals, //! and constructors with payload patterns, such as `Some(x)` or `Type::Variant(x)`. //! A bare constructor such as `None` means its empty payload; a constructor needing a payload diff --git a/interactive/src/parse/pipe/fragment.rs b/interactive/src/parse/pipe/fragment.rs index 0a8269b08..4caf59da4 100644 --- a/interactive/src/parse/pipe/fragment.rs +++ b/interactive/src/parse/pipe/fragment.rs @@ -25,6 +25,17 @@ impl Pattern { _ => false, } } + /// Distinct variables in tuples: a key that restricts nothing, so joining on it needs no test first. + fn unrestricted(&self) -> bool { + fn walk<'a>(p: &'a Pattern, seen: &mut Vec<&'a str>) -> bool { + match p { + Pattern::Var(n) if !seen.contains(&n.as_str()) => { seen.push(n); true } + Pattern::Tuple(ps) => ps.iter().all(|p| walk(p, seen)), + _ => false, + } + } + walk(self, &mut Vec::new()) + } fn constant(&self) -> bool { match self { Self::Literal(_) => true, Self::Tuple(ps) => ps.iter().all(Self::constant), _ => false } } @@ -418,7 +429,8 @@ impl Fragment { 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); + // A restricted first key is tested before a new arrangement, so rows it rejects are never paired. + let direct = first.key.unrestricted() && self.atoms.get(1).is_some_and(|next| first.key == next.key); let mut expr = first.source.clone(); if !direct { let target = self.atoms.get(1).map_or(Target::Output, |a| Target::Key(&a.key)); diff --git a/interactive/tests/fragments.rs b/interactive/tests/fragments.rs index 98df87f87..876093d52 100644 --- a/interactive/tests/fragments.rs +++ b/interactive/tests/fragments.rs @@ -298,7 +298,7 @@ fn first_atom_tests_do_not_prevent_direct_arrangement_reuse() { } #[test] -fn first_key_variants_and_repeated_variables_are_tested_after_direct_join() { +fn first_key_restrictions_are_tested_before_a_new_arrangement() { use interactive::scope_ir::{Item, Node, Ref}; let src = r#" enum Key = Group (int,int) | User (int,int); @@ -312,12 +312,13 @@ fn first_key_variants_and_repeated_variables_are_tested_after_direct_join() { 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))))); + assert!(!matches!(&root.items[*i],Item::Op(Node::Arrange(Ref::Import(0))))); let lowered=report.emit_ddp_at(plan::ddp::DdpStage::Lowered).unwrap().text; + // One flatmap binds the payload before the arrangement; the other reads the rebuilt key after the join. assert_eq!(lowered.matches("| filter(").count(), 1); - assert_eq!(lowered.matches("| flatmap(").count(), 1); - assert!(lowered.find("| join(").unwrap() < lowered.find("| flatmap(").unwrap()); - assert!(lowered.find("| flatmap(").unwrap() < lowered.find("| filter(").unwrap()); + assert_eq!(lowered.matches("| flatmap(").count(), 2); + assert!(lowered.find("| flatmap(").unwrap() < lowered.find("| join(").unwrap()); + assert!(lowered.find("| filter(").unwrap() < lowered.find("| join(").unwrap()); assert_eq!(rows(src,vec![ (0,"inject(0,tuple(1,1)) val=8 diff=2\ninject(0,tuple(1,2)) val=7\ninject(1,tuple(2,2)) val=6".into()), (1,"inject(0,tuple(1,1)) val=9 diff=3\ninject(0,tuple(1,2)) val=5\ninject(1,tuple(2,2)) val=4".into()), From e47e304a0aa4bd639da2c962e02e80ed6dca3e70 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 14 Nov 2026 10:54:34 -0500 Subject: [PATCH 4/5] DDIR: keep an equal restricted first key, and reject empty lists in later keys When the first atom's key is restricted and written the same as the second atom's, the first atom is tested and then keeps its original key, instead of rebuilding it from bound variables. The rebuild used `inject`, which cannot build `Ok`, `Err` or `None` keys when the other lane's shape is unknown, so `a(Ok(x), v) * b(Ok(x), w)` failed to install. The surviving rows keep their key values exactly, and so stay on their workers. An empty list or empty string in a later key, which the expansion would have to build without an element shape, is now a DDIR error at the key. Dead arms of `Pattern::text` are gone, the fragment module is private, and the language reference states the later-key limits. Co-Authored-By: GPT-6 Sol (Codex) Claude-Session: https://claude.ai/code/session_01H6wB4H99TLw4tqu4x48pXL --- interactive/src/parse/pipe.rs | 4 +++- interactive/src/parse/pipe/fragment.rs | 29 ++++++++++++++++++-------- interactive/tests/fragments.rs | 29 ++++++++++++++++++++++++-- 3 files changed, 50 insertions(+), 12 deletions(-) diff --git a/interactive/src/parse/pipe.rs b/interactive/src/parse/pipe.rs index d829500be..b7d31d82c 100644 --- a/interactive/src/parse/pipe.rs +++ b/interactive/src/parse/pipe.rs @@ -69,6 +69,8 @@ //! Atoms name existing bindings (including scope paths); they always have key and value arguments. //! Every later key must mention at least one variable bound by an earlier atom; it may combine //! earlier variables with literals, tuples, unit and constructors, but not wildcards or new variables. +//! Open built-in constructors (`Ok(x)`, `Err(e)`, `None`) cannot be built as later keys, and +//! untyped empty lists (including `""`) cannot appear in them. //! A constant key would move every incoming row to one key; write a cross join as plain DDP. //! Atoms keep their existing arrangements: one new arrangement per `*`, of the rows so far; //! none at the first `*` when the first two keys are the same unrestricted variables. @@ -141,7 +143,7 @@ use super::*; use crate::ir::Lambda; -pub mod fragment; +mod fragment; #[derive(Debug, Clone, PartialEq)] enum Token { diff --git a/interactive/src/parse/pipe/fragment.rs b/interactive/src/parse/pipe/fragment.rs index 4caf59da4..b1970741e 100644 --- a/interactive/src/parse/pipe/fragment.rs +++ b/interactive/src/parse/pipe/fragment.rs @@ -39,6 +39,14 @@ impl Pattern { fn constant(&self) -> bool { match self { Self::Literal(_) => true, Self::Tuple(ps) => ps.iter().all(Self::constant), _ => false } } + fn has_empty_list(&self) -> bool { + match self { + Self::Literal(s) => s == "[]", + Self::Tuple(ps) => ps.iter().any(Self::has_empty_list), + Self::Variant { payload, .. } => payload.has_empty_list(), + _ => false, + } + } fn check_lists(&self) { match self { // Check the whole literal so sibling constants can supply an empty list's shape. @@ -52,9 +60,9 @@ impl Pattern { } fn text(&self) -> String { match self { - Self::Var(n) | Self::Literal(n) => n.clone(), Self::Wild => "_".into(), + Self::Literal(n) => n.clone(), Self::Tuple(ps) => tuple(&ps.iter().map(Self::text).collect::>()), - Self::Variant { name, payload, .. } => format!("{name}({})", payload.text()), + _ => unreachable!("only constants in list literals need text"), } } } @@ -197,7 +205,7 @@ 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 } +enum Target<'a> { Key(&'a Pattern), Keep(&'a Pattern), Current, Output } struct State<'a> { scope: &'a corgi::Scope, next: usize, @@ -268,6 +276,7 @@ impl State<'_> { } } assert!(mentions_bound(self, p), "a constant fragment key would move every incoming row to one key; write a cross join as plain DDP"); + assert!(!p.has_empty_list(), "DDIR fragment key literal `[]` needs a complete element shape; use plain DDP with an explicit list shape"); p.check_lists(); } fn key_expr(&self, p: &Pattern) -> String { @@ -326,9 +335,9 @@ impl State<'_> { let key = if values.len() == 1 { values[0].clone() } else { tuple(&values) }; format!("({key}, ())") } - Target::Key(p) => { + Target::Key(p) | Target::Keep(p) => { self.validate_key(p); - let expr = self.key_expr(p); + let expr = if matches!(target, Target::Keep(_)) { "k".into() } else { self.key_expr(p) }; let mut key = BTreeMap::new(); self.key_locations(p, KeyRead::Value("k".into()), &mut key); self.row(expr, key) @@ -378,7 +387,7 @@ impl State<'_> { 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 (values, pat) = self.capture(&variants[i + 1..], &payload, &format!("capture{i}"), matches!(target, Target::Current | Target::Keep(_))); 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) }; @@ -387,7 +396,7 @@ impl State<'_> { } /// 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); } + if let Target::Key(p) | Target::Keep(p) = target { self.validate_key(p); } let mut expr = self.filters(expr, pat); let mut bindings = Vec::new(); let mut names = BTreeMap::new(); @@ -430,10 +439,12 @@ impl Fragment { s.key = s.locations.iter().map(|(a, expr)| (a.clone(), KeyRead::Value(expr.clone()))).collect(); s.ingest(&first.val, "v".into()); // A restricted first key is tested before a new arrangement, so rows it rejects are never paired. - let direct = first.key.unrestricted() && self.atoms.get(1).is_some_and(|next| first.key == next.key); + let same_key = self.atoms.get(1).is_some_and(|next| first.key == next.key); + let direct = first.key.unrestricted() && same_key; let mut expr = first.source.clone(); if !direct { - let target = self.atoms.get(1).map_or(Target::Output, |a| Target::Key(&a.key)); + // Tests may restrict the key, but an identical lookup must retain its original value. + let target = self.atoms.get(1).map_or(Target::Output, |a| if same_key { Target::Keep(&a.key) } else { 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() { diff --git a/interactive/tests/fragments.rs b/interactive/tests/fragments.rs index 876093d52..8d6a58ebb 100644 --- a/interactive/tests/fragments.rs +++ b/interactive/tests/fragments.rs @@ -1,4 +1,4 @@ -//! Weighted semantics, restrictions, variants, and the two printable syntax views. +//! Weighted fragment semantics, restrictions, placement, and emitted plans. use interactive::{lower, parse, plan, passes::Config, server::Server}; fn program(src: &str) -> interactive::scope_ir::Program { lower::lower_tree(parse::pipe::parse(src)) } @@ -314,7 +314,7 @@ fn first_key_restrictions_are_tested_before_a_new_arrangement() { 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; - // One flatmap binds the payload before the arrangement; the other reads the rebuilt key after the join. + // One flatmap binds the payload before the arrangement; the other reads the preserved key after the join. assert_eq!(lowered.matches("| filter(").count(), 1); assert_eq!(lowered.matches("| flatmap(").count(), 2); assert!(lowered.find("| flatmap(").unwrap() < lowered.find("| join(").unwrap()); @@ -475,3 +475,28 @@ fn variants_rekey_before_pending_plain_payload_tests() { vec![("tuple(100,10,3)".into(), "tuple()".into(), 1)]); }); } + +#[test] +fn identical_restricted_first_keys_keep_their_original_values() { + for (shape, pattern, kept, dropped) in [ + ("Result(int,int)", "Ok(x)", "inject(0,1)", "inject(1,2)"), + ("Result(int,int)", "Err(x)", "inject(1,1)", "inject(0,2)"), + ("(int,Option(int))", "(x,None)", "tuple(1,inject(0,tuple()))", "tuple(2,inject(1,7))"), + ] { + let src = format!("let a=input 0 : ({shape},int); let b=input 1 : ({shape},int); output \"out\"=a({pattern},v)*b({pattern},w);"); + assert_eq!(rows(&src, vec![(0,format!("{kept} val=8 diff=2\n{dropped} val=7")), + (1,format!("{kept} val=9 diff=3\n{dropped} val=5"))]), + vec![("tuple(1,8,9)".into(), "tuple()".into(), 6)]); + let report = plan::report(&src, &|_| None, &Config::all()).unwrap(); + let text = report.emit_ddp_at(plan::ddp::DdpStage::Lowered).unwrap().text; + assert!(!text.contains(" inject "), "{text}"); + } +} + +#[test] +fn empty_lists_and_strings_in_later_keys_report_a_ddir_error() { + for literal in ["[]", "\"\"", "(7,[])", "Some((x,[]))"] { + let src = format!("let a=input 0 : (int,int); let b=input 1 : ((int,List(int)),int); output \"out\"=a(x,y)*b((x,{literal}),z);"); + rejects(&src, "DDIR fragment key literal `[]` needs a complete element shape; use plain DDP with an explicit list shape"); + } +} From 3e684164280b1c36d9b9e96ec6d1b58745ef2698 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Sat, 14 Nov 2026 11:02:02 -0500 Subject: [PATCH 5/5] DDIR: a fragment's key is always the tuple of its bindings A fragment with one binding produced the bare value as its key, `(x, ())`, while two or more produced a tuple, `((x, y), ())`. It now always produces the tuple, `((x), ())` for one binding and `((), ())` for none, so the layout has no special case and matches collections keyed by one-field tuples. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01H6wB4H99TLw4tqu4x48pXL --- interactive/src/parse/pipe.rs | 4 ++-- interactive/src/parse/pipe/fragment.rs | 2 +- interactive/tests/fragments.rs | 12 ++++++------ 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/interactive/src/parse/pipe.rs b/interactive/src/parse/pipe.rs index b7d31d82c..2c54f2ce9 100644 --- a/interactive/src/parse/pipe.rs +++ b/interactive/src/parse/pipe.rs @@ -83,8 +83,8 @@ //! Reusing a variable tests equality. Plain filters run as soon as their fields are bound, //! before the next rekey; a direct first join precedes its first atom's tests. Variants bind //! through typed payload lists in `flatmap`, which can also emit the next lookup key. -//! Rows contain each variable once, in first-appearance order, keyed by their tuple (one -//! variable is scalar, none is unit), with unit value. A test-free final layout composes into +//! Rows contain each variable once, in first-appearance order, keyed by their tuple (`(x)` for +//! one variable, `()` for none), with unit value. A test-free final layout composes into //! the last join's projection. Weights multiply; nothing implies distinct. //! Even the parsed view shows the expansion. //! diff --git a/interactive/src/parse/pipe/fragment.rs b/interactive/src/parse/pipe/fragment.rs index b1970741e..117c61bfb 100644 --- a/interactive/src/parse/pipe/fragment.rs +++ b/interactive/src/parse/pipe/fragment.rs @@ -332,7 +332,7 @@ impl State<'_> { 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) }; + let key = tuple(&values); format!("({key}, ())") } Target::Key(p) | Target::Keep(p) => { diff --git a/interactive/tests/fragments.rs b/interactive/tests/fragments.rs index 8d6a58ebb..318d44df3 100644 --- a/interactive/tests/fragments.rs +++ b/interactive/tests/fragments.rs @@ -53,14 +53,14 @@ fn patterns_tests_and_output_order() { "#; assert_eq!(rows(src,vec![(0,"1,1 val=tuple(list(104,105),2)\n1,9 val=tuple(list(104,105),2)\n3,3 val=tuple(list(104),2)".into()),(1,"2 val=-7 diff=3\n2 val=7".into())]),vec![("tuple(1,2)".into(),"tuple()".into(),3)]); assert_eq!(rows("let a = input 0 : (int,int); output \"out\" = a(_, _);",vec![(0,"1 val=2 diff=4".into())]),vec![("tuple()".into(),"tuple()".into(),4)]); - assert_eq!(rows("let a = input 0 : (int,int); output \"out\" = a(x, _);",vec![(0,"1 val=2".into())]),vec![("1".into(),"tuple()".into(),1)]); + assert_eq!(rows("let a = input 0 : (int,int); output \"out\" = a(x, _);",vec![(0,"1 val=2".into())]),vec![("tuple(1)".into(),"tuple()".into(),1)]); } #[test] fn empty_list_pattern_does_not_choose_an_element_shape() { assert_eq!(rows("let a = input 0 : (int,List((int,int))); output \"out\" = a(x, []);", vec![(0,"1 val=list()\n2 val=list(tuple(3,4))".into())]), - vec![("1".into(),"tuple()".into(),1)]); + vec![("tuple(1)".into(),"tuple()".into(),1)]); } #[test] @@ -157,15 +157,15 @@ fn later_keys_combine_bound_variables_with_constants() { fn bare_constructors_match_empty_payloads_and_do_not_bind_variables() { 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)]); + vec![("tuple(1)".into(), "tuple()".into(), 3)]); let src = "let a=input 0 : (int,(Option(int),Option(int))); output \"out\"=a(x,(None,None));"; assert_eq!(rows(src, vec![(0,"1 val=tuple(inject(0,tuple()),inject(0,tuple()))\n2 val=tuple(inject(0,tuple()),inject(1,7))\n3 val=tuple(inject(1,7),inject(1,7))".into())]), - vec![("1".into(), "tuple()".into(), 1)]); + vec![("tuple(1)".into(), "tuple()".into(), 1)]); let text = plan::report(src, &|_| None, &Config::all()).unwrap().emit_ddp().unwrap().text; assert!(!text.contains("| filter("), "repeated None must not become equality: {text}"); assert_eq!(rows("enum K=Empty | Full int; let a=input 0 : (K,int); output \"out\"=a(K::Empty,v);", vec![(0,"inject(0,tuple()) val=9\ninject(1,7) val=8".into())]), - vec![("9".into(), "tuple()".into(), 1)]); + vec![("tuple(9)".into(), "tuple()".into(), 1)]); assert_eq!(rows("let a=input 0 : (int,int); output \"out\"=a(x,Fresh);", vec![(0,"1 val=2".into())]), vec![("tuple(1,2)".into(), "tuple()".into(), 1)]); } @@ -190,7 +190,7 @@ fn nested_empty_list_literals_report_a_ddir_error() { ("[[1],[]]", "list(list(1),list())"), ("[[],[1]]", "list(list(),list(1))")] { let src = format!("let a=input 0 : (int,List(List(int))); output \"out\"=a(x,{literal});"); assert_eq!(rows(&src, vec![(0,format!("1 val={value}\n2 val=list()"))]), - vec![("1".into(), "tuple()".into(), 1)]); + vec![("tuple(1)".into(), "tuple()".into(), 1)]); } }