Server install check; one meaning for spread and filter

interactive/src/backend/vec.rs

@@ -79,7 +79,10 @@ fn render_linear<'scope>(c: Col<'scope>, ops: Vec<LinearOp>, 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 => {

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<usize> = Vec::new();
             for f in fields {
                 match f {
@@ -683,9 +683,14 @@ pub fn compile_scalar(term: &Term, kshape: &Shape, vshape: &Shape) -> Res<Graph<
     }
 }
 
-/// Compile a `Filter` predicate → a mask column (nonzero keeps the row).
+/// Compile a `Filter` predicate → a mask column (nonzero keeps the row). A predicate must be an
+/// `Int`; any other shape is a type error, as it is in the row backend.
 pub fn compile_predicate(cond: &Term, kshape: &Shape, vshape: &Shape) -> Res<Graph<NumOp>> {
-    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

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<Term>),
-    /// List intro. A `Spread` child splices in place.
+    /// List intro.
     List(Vec<Term>),
-    /// 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<Term>),
     /// Product/list elimination: index into a `Tuple` or `List`.
@@ -248,9 +249,9 @@ pub fn eval(term: &Term, env: &mut Vec<Value>) -> 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<Value>) -> Vec<Value> {
+/// 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<Value>) -> Vec<Value> {
     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

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 {

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<String> = 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<String> = prog
             .root

interactive/tests/corgi_backend.rs

@@ -54,6 +54,8 @@ fn inputs_for(prog: &str) -> Vec<Vec<(Value, Value)>> {
             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");
+    }
+}

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);

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());
+    });
+}