From b91c00e605a86b5db24a7b6ef3842f9fbf78700a Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Wed, 23 Sep 2026 08:03:37 -0400 Subject: [PATCH 1/2] Server: check a program before installing its generated sources `install` created generated sources (`clock`, recipes) while resolving imports, before checking the program's exports and remaining imports. A program rejected for a taken export name or an unknown import left those sources installed until a later drop. Now every check runs first. Co-Authored-By: Claude Opus 5.5 (1M context) --- interactive/src/server.rs | 27 ++++++++++++++-------- interactive/tests/server_install.rs | 35 +++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 9 deletions(-) create mode 100644 interactive/tests/server_install.rs diff --git a/interactive/src/server.rs b/interactive/src/server.rs index d835762e8..fe7c0ae30 100644 --- a/interactive/src/server.rs +++ b/interactive/src/server.rs @@ -448,31 +448,40 @@ impl Server { if self.programs.contains_key(name) { return Err(format!("a program named {:?} is already installed", name)); } - // Resolve trace imports against canonical names, installing generated - // sources (e.g. `random:...`) on demand. A name that is neither - // registered nor a recipe is an error. + // Resolve trace imports against canonical names. A name that is not registered must be a + // generated source (`clock`, or a recipe such as `random:...`), installed on demand below. + // Everything is checked before anything is installed, so a rejected program leaves no trace. + let mut generated: Vec = Vec::new(); for imp in &prog.root.imports { if let st::Source::Trace(t) = &imp.from { let key = canonical_source_name(t); if !self.traces.contains_key(&key) { - if key == "clock" { - self.install_clock(worker); - } else if let Some(recipe) = Recipe::parse(&key) { - self.install_generated(worker, &key, recipe); - } else { + if key != "clock" && Recipe::parse(&key).is_none() { return Err(format!( "program {:?} imports unknown trace {:?}; install its producer first", name, t )); } + generated.push(key); } } } for e in &prog.root.exports { - if self.traces.contains_key(&e.name) { + if self.traces.contains_key(&e.name) || generated.contains(&e.name) { return Err(format!("export name {:?} is already published; choose another name or drop its producer", e.name)); } } + for key in generated { + if self.traces.contains_key(&key) { + continue; // imported twice + } + if key == "clock" { + self.install_clock(worker); + } else { + let recipe = Recipe::parse(&key).expect("checked above"); + self.install_generated(worker, &key, recipe); + } + } let import_names: Vec = prog .root diff --git a/interactive/tests/server_install.rs b/interactive/tests/server_install.rs new file mode 100644 index 000000000..5bda347b8 --- /dev/null +++ b/interactive/tests/server_install.rs @@ -0,0 +1,35 @@ +//! `Server::install` checks a program before installing anything for it. + +use interactive::scope_ir::Program; +use interactive::server::Server; +use interactive::{lower, parse}; + +fn program(src: &str) -> Program { + let mut program = lower::lower_tree(parse::pipe::parse(src)); + program.optimize(); + program +} + +#[test] +fn a_rejected_install_leaves_no_generated_sources() { + timely::execute_directly(move |worker| { + let mut server = Server::new(); + let recipe = "random:nodes=8,edges=12"; + server.install(worker, "first", &program(r#"export "taken" = input 0;"#)).unwrap(); + + // The recipe would be generated, but the export is taken. + let clash = program(&format!(r#"let e = import "{recipe}"; export "taken" = e;"#)); + assert!(server.install(worker, "second", &clash).is_err()); + assert!(server.snapshot(worker, recipe).is_err()); + + // The recipe would be generated, but the other import is unknown. + let unknown = program(&format!(r#"let e = import "{recipe}"; let f = import "nope"; export "x" = e + f;"#)); + assert!(server.install(worker, "third", &unknown).is_err()); + assert!(server.snapshot(worker, recipe).is_err()); + + // Accepted, the program installs the recipe as a source. + let fine = program(&format!(r#"let e = import "{recipe}"; export "fresh" = e;"#)); + server.install(worker, "fourth", &fine).unwrap(); + assert!(server.snapshot(worker, recipe).is_ok()); + }); +} From ac071a434f26f6f625542d12350667cbf75289d5 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Wed, 23 Sep 2026 08:10:50 -0400 Subject: [PATCH 2/2] DDIR: one meaning for spread and filter across backends The two backends disagreed on two corners of the language: - A bare `$n` in a projection whose row is not a tuple. The row backend spliced a list's elements (so the tuple's arity varied with the data) and panicked on a scalar; the columnar backend kept either as one field. Both now splice a tuple's fields and keep any other value as one field, which is the only layout a column can hold. - A filter predicate that is not an `Int`. The row backend dropped every row; the columnar backend panicked on an opaque mask error. Both now reject it, naming the shape they got. `Spread` is only produced inside projection tuples, so the row evaluator's list-intro splice goes too. Adds `spread_values` to the corgi/vec gate, and a test that both backends reject a tuple predicate. Co-Authored-By: Claude Opus 5.5 (1M context) --- interactive/src/backend/vec.rs | 5 +++- interactive/src/corgi/logic.rs | 11 +++++-- interactive/src/ir.rs | 31 ++++++++++---------- interactive/src/parse/pipe.rs | 6 ++-- interactive/tests/corgi_backend.rs | 17 +++++++++++ interactive/tests/programs/spread_values.ddp | 6 ++++ 6 files changed, 54 insertions(+), 22 deletions(-) create mode 100644 interactive/tests/programs/spread_values.ddp diff --git a/interactive/src/backend/vec.rs b/interactive/src/backend/vec.rs index 53babf72a..b6e10dd43 100644 --- a/interactive/src/backend/vec.rs +++ b/interactive/src/backend/vec.rs @@ -79,7 +79,10 @@ fn render_linear<'scope>(c: Col<'scope>, ops: Vec, level: usize) -> Co next.push(((nk, nv), t, d)); }, LinearOp::Filter(cond) => { - let keep = { let mut env = vec![k.clone(), v.clone()]; eval(cond, &mut env).truthy() }; + let keep = match eval(cond, &mut vec![k.clone(), v.clone()]) { + Value::Int(n) => n != 0, + other => panic!("a filter predicate must be an Int, got {other:?}"), + }; if keep { next.push(((k, v), t, d)); } }, LinearOp::Negate => { diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs index 7d8a5bef5..6600ce0d6 100644 --- a/interactive/src/corgi/logic.rs +++ b/interactive/src/corgi/logic.rs @@ -255,7 +255,7 @@ pub fn compile( } Term::Int(n) => Ok(b.add(Op::Lit(CValue::u64(vec![*n as u64])), vec![anchor])), Term::Tuple(fields) => { - // A `Spread(t)` child splices `t`'s `Prod` fields in place (the flat-row model). + // A `Spread(t)` child splices `t`'s `Prod` fields in place; any other value is one field. let mut ids: Vec = Vec::new(); for f in fields { match f { @@ -683,9 +683,14 @@ pub fn compile_scalar(term: &Term, kshape: &Shape, vshape: &Shape) -> Res Res> { - compile_over_kv(cond, kshape, vshape) + let g = compile_over_kv(cond, kshape, vshape)?; + match corgi::shape_of(&g, &Shape::Prod(vec![kshape.clone(), vshape.clone()]))? { + Shape::Prim(_) => Ok(g), + other => Err(format!("a filter predicate must be an Int, got {other}")), + } } /// Compile a join projection: key/val Terms over `Var(0)=key`, `Var(1)=val0`, `Var(2)=val1` (with diff --git a/interactive/src/ir.rs b/interactive/src/ir.rs index 61c67c068..46dda7249 100644 --- a/interactive/src/ir.rs +++ b/interactive/src/ir.rs @@ -81,11 +81,12 @@ pub enum Term { Bound(usize), /// Integer literal. Int(i64), - /// Product intro. A `Spread` child splices its tuple's fields in place. + /// Product intro. A `Spread` child splices a tuple's fields in place; any + /// other value is one field. Tuple(Vec), - /// List intro. A `Spread` child splices in place. + /// List intro. List(Vec), - /// Splice marker; only meaningful as a direct child of `Tuple`/`List`. + /// Splice marker; only meaningful as a direct child of `Tuple`. /// Lets a whole input row (`$n`) contribute all its fields. Spread(Box), /// Product/list elimination: index into a `Tuple` or `List`. @@ -248,9 +249,9 @@ pub fn eval(term: &Term, env: &mut Vec) -> Value { Term::Var(i) => env[*i].clone(), Term::Bound(k) => env[env.len() - 1 - *k].clone(), Term::Int(n) => Value::Int(*n), - Term::Tuple(fields) => Value::Tuple(build_seq(fields, env)), - Term::List(fields) => Value::List(build_seq(fields, env)), - Term::Spread(_) => panic!("Spread is only valid as a direct child of Tuple/List"), + Term::Tuple(fields) => Value::Tuple(tuple_fields(fields, env)), + Term::List(fields) => Value::List(fields.iter().map(|f| eval(f, env)).collect()), + Term::Spread(_) => panic!("Spread is only valid as a direct child of Tuple"), Term::Proj(t, i) => { // If the operand is a "place" (a Var/Bound/Proj chain), index into // it by reference and clone only the selected field — avoids deep- @@ -342,17 +343,17 @@ fn eval_ref<'a>(term: &Term, env: &'a [Value]) -> Option<&'a Value> { } } -/// Build a tuple/list element sequence, splicing any `Spread` children. -fn build_seq(fields: &[Term], env: &mut Vec) -> Vec { +/// Build a tuple's fields. A `Spread` child splices a tuple's fields in place (a unit splices +/// nothing); any other value is one field, so a tuple's arity never depends on the data. +fn tuple_fields(fields: &[Term], env: &mut Vec) -> Vec { let mut out = Vec::with_capacity(fields.len()); for f in fields { - if let Term::Spread(inner) = f { - match eval(inner, env) { - Value::Tuple(xs) | Value::List(xs) => out.extend(xs), - other => panic!("Spread of non-aggregate value: {:?}", other), - } - } else { - out.push(eval(f, env)); + match f { + Term::Spread(inner) => match eval(inner, env) { + Value::Tuple(xs) => out.extend(xs), + other => out.push(other), + }, + _ => out.push(eval(f, env)), } } out diff --git a/interactive/src/parse/pipe.rs b/interactive/src/parse/pipe.rs index 350562d5f..2396b6eae 100644 --- a/interactive/src/parse/pipe.rs +++ b/interactive/src/parse/pipe.rs @@ -449,9 +449,9 @@ impl Parser { } // A projection `(k1, k2 ; v1, v2)` builds `key = tuple(k1, k2)` and - // `val = tuple(v1, v2)`. A bare `$n` field splices the whole input row's - // fields (`Spread`), matching the flat-row concatenation of the original - // `[i64]` model; any other field is one (possibly nested) element. + // `val = tuple(v1, v2)`. A bare `$n` field splices the input row's tuple + // fields (`Spread`), or is one field when the row is not a tuple; any other + // field is one (possibly nested) element. fn parse_projection_inner(&mut self) -> Projection { let key = self.parse_field_list_until(&[Token::Semi, Token::RParen]); let val = if *self.peek() == Token::Semi { diff --git a/interactive/tests/corgi_backend.rs b/interactive/tests/corgi_backend.rs index 074db00d6..62fca1f3a 100644 --- a/interactive/tests/corgi_backend.rs +++ b/interactive/tests/corgi_backend.rs @@ -54,6 +54,8 @@ fn inputs_for(prog: &str) -> Vec> { rows(&[&[1, 1, 10], &[1, 2, 20], &[2, 1, 30], &[2, 1, 31], &[9, 9, 90]]), rows(&[&[1, 1, 5], &[2, 1, 6], &[3, 3, 7]]), ], + // spread_values: keys with one and with several values. + "spread_values" => vec![rows(&[&[1, 10], &[1, 20], &[2, 30]])], "signed_min" => vec![rows(&[ &[1, 0], &[1, -1], @@ -132,3 +134,18 @@ fn serializing(n: usize) -> timely::Config { #[test] fn tour() { assert_backends_agree("tour"); } #[test] fn pair_keys() { assert_backends_agree("pair_keys"); } #[test] fn signed_min() { assert_backends_agree("signed_min"); } +#[test] fn spread_values() { assert_backends_agree("spread_values"); } + +/// A filter predicate must be an `Int`: both backends reject a tuple rather than one of them +/// keeping nothing. +#[test] +fn filter_requires_an_int_predicate() { + let mut tree = lower::lower_tree(parse::pipe::parse(r#"export "result" = input 0 | filter($0);"#)); + tree.optimize(); + let inputs = vec![rows(&[&[1, 10], &[0, 20]])]; + for backend in [RenderBackend::Vec, RenderBackend::Corgi] { + let (tree, inputs) = (tree.clone(), inputs.clone()); + let result = std::panic::catch_unwind(move || evaluate(backend, timely::Config::process(1), &tree, &inputs)); + assert!(result.is_err(), "{backend:?} accepted a tuple filter predicate"); + } +} diff --git a/interactive/tests/programs/spread_values.ddp b/interactive/tests/programs/spread_values.ddp new file mode 100644 index 000000000..3c776a1f2 --- /dev/null +++ b/interactive/tests/programs/spread_values.ddp @@ -0,0 +1,6 @@ +-- A bare `$n` splices a tuple's fields; any other value is one field. Here the spread value is a +-- list (after `collect`) and an integer (after `value`), each kept whole. +let rows = input 0 | key($0[0] ; $0[1]); + +export "lists" = rows | collect | key($0 ; $1) | inspect(lists); +export "ints" = rows | value($1[0]) | key($0 ; $1) | inspect(ints);