From f09a87703c4d950428654069eea0db75b0864f4f Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Mon, 12 Oct 2026 16:29:02 -0400 Subject: [PATCH] DDIR: child scopes read their parent's arrangements A child scope used to receive every import as rows. An arrangement in the parent was flattened at the boundary, and each join, reduce, or arrange of it inside the child built a new one. Imports now arrive as the parent rendered them, and an arrangement enters the child's region with `Arranged::enter_region`, sharing the trace. Every scope has the same time type, so no `TraceEnter` is needed, and compaction requests need no translation. A reader inside a loop asks for compaction at iteration times, which is sound: each handle's frontier lower bounds the times its reader will ask about, and the trace compacts to the meet of its handles. Sharing happens when the parent item is an arrangement: an explicit `arrange`, or a reduce. `lower` places the arrange for a join in the joining scope, so a child joining a parent's collection still arranges it unless the parent does. Co-Authored-By: Claude Opus 5.5 --- interactive/src/backend/corgi.rs | 4 + interactive/src/backend/mod.rs | 36 ++++++-- interactive/src/backend/vec.rs | 4 + interactive/tests/shared_arrangements.rs | 112 +++++++++++++++++++++++ 4 files changed, 150 insertions(+), 6 deletions(-) create mode 100644 interactive/tests/shared_arrangements.rs 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); +}