DDIR: child scopes read their parent's arrangements

interactive/src/backend/corgi.rs

@@ -242,6 +242,10 @@ impl Backend for CorgiBackend {
         )
     }
 
+    fn enter_region<'s, 'r>(a: Self::Arr<'s>, region: timely::dataflow::Scope<'r, Time>) -> Self::Arr<'r> {
+        a.enter_region(region)
+    }
+
     fn as_collection<'s>(a: Self::Arr<'s>) -> Collection<'s, Time, CC> {
         // Each chunk already IS a columnar container: its key/val columns clone by Arc bump,
         // so a chunk becomes a `CorgiContainer` for the price of copying its time lanes and

interactive/src/backend/mod.rs

@@ -43,6 +43,9 @@ pub trait Backend {
     /// `shape` is the input's row shape, where it is known before any data arrives.
     fn linear<'s>(c: Collection<'s, Time, Self::Container>, ops: Vec<LinearOp>, level: usize, shape: Option<st::RowShape>) -> Collection<'s, Time, Self::Container>;
     fn arrange<'s>(c: Collection<'s, Time, Self::Container>) -> Self::Arr<'s>;
+    /// Bring an arrangement into a child region. Regions share the timestamp, so this shares
+    /// the trace rather than rebuilding it.
+    fn enter_region<'s, 'r>(a: Self::Arr<'s>, region: Scope<'r, Time>) -> Self::Arr<'r>;
     fn as_collection<'s>(a: Self::Arr<'s>) -> Collection<'s, Time, Self::Container>;
     /// `shapes` are the two inputs' row shapes, where they are known before any data arrives.
     fn join<'s>(l: Self::Arr<'s>, r: Self::Arr<'s>, projection: &Projection, shapes: Option<(st::RowShape, st::RowShape)>) -> Collection<'s, Time, Self::Container>;
@@ -58,6 +61,12 @@ enum Rendered<'s, B: Backend> {
 }
 
 impl<'s, B: Backend> Rendered<'s, B> {
+    fn enter<'r>(&self, region: Scope<'r, Time>) -> Rendered<'r, B> {
+        match self {
+            Rendered::Collection(c) => Rendered::Collection(c.clone().enter(region)),
+            Rendered::Arrangement(a) => Rendered::Arrangement(B::enter_region(a.clone(), region)),
+        }
+    }
     fn collection(&self) -> Collection<'s, Time, B::Container> {
         match self {
             Rendered::Collection(c) => c.clone(),
@@ -81,7 +90,7 @@ enum RItem<'s, B: Backend> {
 
 fn resolve<'s, B: Backend>(
     items: &[RItem<'s, B>],
-    imports: &[Collection<'s, Time, B::Container>],
+    imports: &[Rendered<'s, B>],
     var_cols: &[Collection<'s, Time, B::Container>],
     r: &st::Ref,
 ) -> Rendered<'s, B> {
@@ -91,7 +100,10 @@ fn resolve<'s, B: Backend>(
             RItem::Op(Rendered::Arrangement(a)) => Rendered::Arrangement(a.clone()),
             RItem::Sub(_) => panic!("Ref::Local points at a child scope"),
         },
-        st::Ref::Import(i) => Rendered::Collection(imports[*i].clone()),
+        st::Ref::Import(i) => match &imports[*i] {
+            Rendered::Collection(c) => Rendered::Collection(c.clone()),
+            Rendered::Arrangement(a) => Rendered::Arrangement(a.clone()),
+        },
         st::Ref::Var(i) => Rendered::Collection(var_cols[*i].clone()),
         st::Ref::ChildExport(i, j) => match &items[*i] {
             RItem::Sub(exports) => Rendered::Collection(exports[*j].clone()),
@@ -111,6 +123,18 @@ pub fn render_tree<'s, B: Backend>(
     depth: usize,
     imports: Vec<Collection<'s, Time, B::Container>>,
     shapes: Option<&ScopeShapes>,
+) -> Vec<Collection<'s, Time, B::Container>> {
+    render_scope::<B>(s, scope, depth, imports.into_iter().map(Rendered::Collection).collect(), shapes)
+}
+
+/// [`render_tree`] at any depth. A child's imports are the parent's items as rendered, so an
+/// arrangement crosses into the child as the same trace, not as rows to arrange again.
+fn render_scope<'s, B: Backend>(
+    s: &st::Scope,
+    scope: Scope<'s, Time>,
+    depth: usize,
+    imports: Vec<Rendered<'s, B>>,
+    shapes: Option<&ScopeShapes>,
 ) -> Vec<Collection<'s, Time, B::Container>> {
     let shape_of = |r: &st::Ref| shapes.and_then(|sh| sh.of(r)).map(|c| (c.key, c.val));
     let mut var_handles: Vec<Option<Variable<'s, Time, B::Container>>> = Vec::new();
@@ -154,17 +178,17 @@ pub fn render_tree<'s, B: Backend>(
                 items.push(RItem::Op(rendered));
             },
             st::Item::Sub(child) => {
-                let child_imports: Vec<Collection<'s, Time, B::Container>> = child.imports.iter().map(|imp| match &imp.from {
-                    st::Source::Parent(r) => resolve(&items, &imports, &var_cols, r).collection(),
+                let child_imports: Vec<Rendered<'s, B>> = child.imports.iter().map(|imp| match &imp.from {
+                    st::Source::Parent(r) => resolve(&items, &imports, &var_cols, r),
                     other => panic!("non-root scope with external source {:?}", other),
                 }).collect();
                 // Each `{}` scope is a real timely region: imports enter it,
                 // the child renders inside, exports leave it structurally —
                 // and then pop the child's dynamic coordinate.
                 let exported = scope.region_named(&child.name, |region| {
-                    let entered: Vec<_> = child_imports.iter().map(|c| c.clone().enter(region)).collect();
+                    let entered: Vec<_> = child_imports.iter().map(|c| c.enter(region)).collect();
                     let child_shapes = shapes.and_then(|sh| match sh.items.get(index) { Some(ItemShapes::Sub(c)) => Some(c), _ => None });
-                    let exports = render_tree::<B>(child, region, depth + 1, entered, child_shapes);
+                    let exports = render_scope::<B>(child, region, depth + 1, entered, child_shapes);
                     exports.into_iter().map(|c| c.leave(scope)).collect::<Vec<_>>()
                 });
                 let left: Vec<Collection<'s, Time, B::Container>> = exported.into_iter().map(|c| B::leave_dynamic(c, depth + 1)).collect();

interactive/src/backend/vec.rs

@@ -137,6 +137,10 @@ impl Backend for VecBackend {
     fn arrange<'s>(c: Collection<'s, Time, Self::Container>) -> Self::Arr<'s> {
         c.arrange_by_key()
     }
+    fn enter_region<'s, 'r>(a: Self::Arr<'s>, region: timely::dataflow::Scope<'r, Time>) -> Self::Arr<'r> {
+        a.enter_region(region)
+    }
+
     fn as_collection<'s>(a: Self::Arr<'s>) -> Collection<'s, Time, Self::Container> {
         a.as_collection(|k, v| (k.clone(), v.clone()))
     }

interactive/tests/shared_arrangements.rs

@@ -0,0 +1,112 @@
+//! An arrangement imported into a child scope is the parent's trace, not a copy.
+//!
+//! Each program runs twice: as written, where children arrange their imports themselves, and
+//! with the parent arranging them, so children read the parent's trace. The answers must agree
+//! at every epoch while edges come and go, on both backends and at several workers. The loops
+//! read the shared traces at iteration times, so this also checks that their compaction
+//! requests are sound for the parent's arrangement.
+use interactive::ir::Value;
+use interactive::scope_ir::{Item, Node, Program, Ref, Scope, Source};
+use interactive::server::{InputUpdate, RenderBackend, Server};
+use interactive::{lower, parse};
+
+fn program(src: &str) -> Program {
+    let mut p = lower::lower_tree(parse::pipe::parse(src));
+    p.optimize();
+    p
+}
+
+/// The number of child-scope imports, anywhere in the tree, whose parent item is an arrangement.
+fn arranged_imports(s: &Scope) -> usize {
+    s.items.iter().map(|item| match item {
+        Item::Op(_) => 0,
+        Item::Sub(c) => {
+            let here = c.imports.iter().filter(|imp| matches!(&imp.from,
+                Source::Parent(Ref::Local(i)) if matches!(&s.items[*i], Item::Op(Node::Arrange(_) | Node::Reduce { .. }))
+            )).count();
+            here + arranged_imports(c)
+        }
+    }).sum()
+}
+
+/// Edge changes per epoch: a few insertions and a few deletions of earlier insertions.
+fn churn(epochs: usize) -> Vec<Vec<InputUpdate>> {
+    let mut state = 0x2545F4914F6CDD1Du64;
+    let mut next = move |n: u64| { state ^= state << 13; state ^= state >> 7; state ^= state << 17; (state % n) as i64 };
+    let mut live: Vec<(i64, i64)> = Vec::new();
+    let edge = |(s, d): (i64, i64), diff| InputUpdate { key: Value::Tuple(vec![Value::Int(s), Value::Int(d)]), val: Value::unit(), diff };
+    (0..epochs).map(|epoch| {
+        let mut batch = Vec::new();
+        for _ in 0..if epoch == 0 { 30 } else { 4 } {
+            let e = (next(16), next(16));
+            live.push(e);
+            batch.push(edge(e, 1));
+        }
+        for _ in 0..if epoch == 0 { 0 } else { 3 } {
+            let e = live.swap_remove(next(live.len() as u64) as usize);
+            batch.push(edge(e, -1));
+        }
+        batch
+    }).collect()
+}
+
+/// The export `result` after each epoch, read at worker 0.
+fn run(backend: RenderBackend, workers: usize, program: &Program, epochs: &[Vec<InputUpdate>]) -> Vec<Vec<(Value, Value, i64)>> {
+    let (program, epochs) = (program.clone(), epochs.to_vec());
+    let guards = timely::execute(timely::Config::process(workers), move |worker| {
+        let mut server = Server::with_backend(backend);
+        server.install(worker, "p", &program).unwrap();
+        let mut seen = Vec::new();
+        for batch in epochs.iter() {
+            let mine = if worker.index() == 0 { batch.clone() } else { Vec::new() };
+            server.feed_batch("p", 0, mine).unwrap();
+            server.tick(worker);
+            let mut rows = server.snapshot(worker, "result").unwrap();
+            rows.sort();
+            seen.push(rows);
+        }
+        seen
+    }).unwrap();
+    guards.join().into_iter().next().unwrap().unwrap()
+}
+
+fn assert_sharing_agrees(written: &str, shared: &str) {
+    let (written, shared) = (program(written), program(shared));
+    assert_eq!(arranged_imports(&written.root), 0);
+    assert!(arranged_imports(&shared.root) > 0, "the shared variant imports no arrangement");
+    let epochs = churn(8);
+    let want = run(RenderBackend::Vec, 1, &written, &epochs);
+    assert!(want.iter().any(|rows| !rows.is_empty()));
+    for backend in [RenderBackend::Vec, RenderBackend::Corgi] {
+        for workers in [1, 3] {
+            assert_eq!(run(backend, workers, &shared, &epochs), want, "{backend:?} at {workers} worker(s)");
+        }
+    }
+}
+
+const REACH: &str = r#"
+    let edges = input 0 : ((int, int) ; ()) | key($0[0] ; $0[1]) EDGES;
+    let roots = edges | key($0[0] ;) | distinct | filter($0[0] < 3);
+    reach: {
+        let proposals = reach | join(edges, ($2 ;));
+        var reach = roots + proposals | distinct;
+    }
+    export "result" = reach::reach;
+"#;
+
+#[test]
+fn reach() {
+    assert_sharing_agrees(&REACH.replace(" EDGES", ""), &REACH.replace(" EDGES", " | arrange"));
+}
+
+#[test]
+fn scc() {
+    let src = interactive::load_program(&format!("{}/examples/programs/scc.ddp", env!("CARGO_MANIFEST_DIR")));
+    let src = src.replace(r#"export "result" = outer::scc | map(;) | arrange | inspect(total);"#, r#"export "result" = outer::scc;"#);
+    let shared = src
+        .replace("| key($0[0] ; $0[1]);", "| key($0[0] ; $0[1]) | arrange;")
+        .replace("let trans = edges | key($1 ; $0);", "let trans = edges | key($1 ; $0) | arrange;")
+        .replace("let scc = edges + trim;", "let scc = edges + trim | arrange;");
+    assert_ne!(src, shared);
+    assert_sharing_agrees(&src, &shared);
+}