diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index b80eea9b7..f21a7b1f6 100644 --- a/interactive/src/backend/corgi.rs +++ b/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 diff --git a/interactive/src/backend/mod.rs b/interactive/src/backend/mod.rs index b9edac7bd..0d0f8d745 100644 --- a/interactive/src/backend/mod.rs +++ b/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, level: usize, shape: Option) -> 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>, shapes: Option<&ScopeShapes>, +) -> Vec> { + render_scope::(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>, + shapes: Option<&ScopeShapes>, ) -> Vec> { let shape_of = |r: &st::Ref| shapes.and_then(|sh| sh.of(r)).map(|c| (c.key, c.val)); let mut var_handles: Vec>> = 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> = child.imports.iter().map(|imp| match &imp.from { - st::Source::Parent(r) => resolve(&items, &imports, &var_cols, r).collection(), + let child_imports: Vec> = 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::(child, region, depth + 1, entered, child_shapes); + let exports = render_scope::(child, region, depth + 1, entered, child_shapes); exports.into_iter().map(|c| c.leave(scope)).collect::>() }); let left: Vec> = exported.into_iter().map(|c| B::leave_dynamic(c, depth + 1)).collect(); diff --git a/interactive/src/backend/vec.rs b/interactive/src/backend/vec.rs index 13497a0df..547814df7 100644 --- a/interactive/src/backend/vec.rs +++ b/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())) } diff --git a/interactive/tests/shared_arrangements.rs b/interactive/tests/shared_arrangements.rs new file mode 100644 index 000000000..5a92689b3 --- /dev/null +++ b/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> { + 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]) -> Vec> { + 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); +}