DDIR: join fragments, a(x, y) * b(y, z), with one flatmap after the joins (spike)

A join fragment is a pipeline source made of atoms joined by `*`.
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.
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.
A later atom's key must be built from bound variables, so a test never forms a new arrangement.
Key variables are read from the join's key slot, and the first atom joins directly when its key is the next key.

Everything after the last join, the tests and the output layout, is one `flatmap` that returns zero or one rows.
A variant's payload is read from its lane with `enlist unweave`, and literal and equality tests keep or drop the row with a list `filter`, so only the rows kept are built.
Dropping dead bindings is left to the projection pass.

Co-Authored-By: GPT-6 Sol (Codex) <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01H6wB4H99TLw4tqu4x48pXL

Author: 3 people

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<RefName>) {
         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<Stmt>) -> 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())] }) },

interactive/src/parse/mod.rs

@@ -53,6 +53,8 @@ pub enum Expr {
     Path(Vec<String>, bool),
     /// A path checked against the scope tree before lowering.
     ResolvedPath { anchor: Vec<String>, target: Vec<String>, field: String },
+    /// A join fragment, expanded to ordinary collection syntax before IR lowering.
+    Fragment(Box<pipe::fragment::Fragment>),
     Map(Box<Expr>, Lambda),
     Join(Box<Expr>, Box<Expr>, Lambda),
     /// `stream | half_join(arrangement, strict, projection)`: see `scope_ir::Node::HalfJoin`.

interactive/src/parse/pipe.rs

@@ -62,6 +62,20 @@
 //!   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.
 //!
+//! A join 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 is built from earlier variables, tuples, unit, or variant constructors.
+//! It cannot contain a literal, wildcard, or new variable: atoms keep their existing arrangements.
+//! 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)`.
+//! Reusing a variable tests equality. Tests run after all joins, except variant tags needed by
+//! later lookup keys. Rows contain each variable once, in first-appearance order, keyed by their
+//! tuple (one variable is scalar, none is unit), with unit value. Weights multiply; no distinct.
+//! The tests and the output layout after the last join are one `flatmap`: a variant's payload is
+//! read from its lane with `unweave`, and literal and equality tests keep or drop the row with a
+//! list `filter`, so only the rows kept are built.
+//! Expansion precedes IR lowering; parsed DDP retains the fragment and lowered DDP shows it expanded.
+//!
 //! 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 +131,8 @@
 use super::*;
 use crate::ir::Lambda;
 
+pub mod fragment;
+
 #[derive(Debug, Clone, PartialEq)]
 enum Token {
     Let, Var, Set, Output, Type, Enum,
@@ -126,7 +142,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 +282,9 @@ fn lex(input: &str) -> Vec<Token> {
             ';' => { 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 +565,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 +583,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 => {