From 86ae2c34ab98e1e13b0a3a55a11012e8499c18d1 Mon Sep 17 00:00:00 2001 From: Leo Stewen Date: Thu, 2 Jul 2026 11:36:59 +0200 Subject: [PATCH 1/3] Implement RecursiveStreams for Vec> --- crates/dbsp/src/operator/dynamic/recursive.rs | 446 +++++++++++++----- crates/dbsp/src/operator/recursive.rs | 179 +++++++ 2 files changed, 512 insertions(+), 113 deletions(-) diff --git a/crates/dbsp/src/operator/dynamic/recursive.rs b/crates/dbsp/src/operator/dynamic/recursive.rs index 4888b106b26..188aa356a08 100644 --- a/crates/dbsp/src/operator/dynamic/recursive.rs +++ b/crates/dbsp/src/operator/dynamic/recursive.rs @@ -82,8 +82,8 @@ where ) } - fn connect(&self, vars: Self::Feedback) { - vars.connect(self) + fn connect(&self, var: Self::Feedback) { + var.connect(self) } fn export(self, factories: &Self::Factories) -> Self::Export { @@ -95,7 +95,81 @@ where } } -// TODO: `impl RecursiveStreams for Vec`. +/// Recursion over a group of streams whose size is only known at runtime. +/// +/// The arity of the group (the number of mutually recursive streams) is +/// determined by the length of the `factories` vector passed to +/// [`new`](RecursiveStreams::new). Every other method preserves this arity: +/// the closure driving the recursion must therefore return exactly as many +/// streams as it received. Unlike the tuple implementations, all streams in +/// the group share the same batch type `B`. +impl RecursiveStreams for Vec> +where + C: Circuit, + C::Parent: Circuit, + B: Checkpoint + IndexedZSet + Send + Sync, + Spine: SizeOf, +{ + type Feedback = Vec>; + type Export = Vec>>; + type Output = Vec>; + type Factories = Vec>; + + fn new(circuit: &C, factories: &Self::Factories) -> (Self::Feedback, Self) { + factories + .iter() + .map(|factory| { + let feedback = + DelayedFeedback::with_default(circuit, B::dyn_empty(&factory.input_factories)); + let stream = feedback.stream().clone(); + (feedback, stream) + }) + .unzip() + } + + fn distinct(mut self, factories: &Self::Factories) -> Self { + debug_assert_eq!(self.len(), factories.len()); + + for (stream, factory) in self.iter_mut().zip(factories) { + let persistent_id = stream + .get_persistent_id() + .map(|name| format!("{name}.distinct")); + *stream = + Stream::dyn_distinct(&stream, factory).set_persistent_id(persistent_id.as_deref()); + } + + self + } + + fn connect(&self, vars: Self::Feedback) { + debug_assert_eq!(self.len(), vars.len()); + + for (stream, var) in self.iter().zip(vars) { + var.connect(stream); + } + } + + fn export(self, factories: &Self::Factories) -> Self::Export { + debug_assert_eq!(self.len(), factories.len()); + + self.into_iter() + .zip(factories) + .map(|(stream, factory)| { + Stream::export(&stream.dyn_integrate_trace(&factory.input_factories)) + }) + .collect() + } + + fn consolidate(exports: Self::Export, factories: &Self::Factories) -> Self::Output { + debug_assert_eq!(exports.len(), factories.len()); + + exports + .into_iter() + .zip(factories) + .map(|(stream, factory)| Stream::dyn_consolidate(&stream, &factory.input_factories)) + .collect() + } +} #[allow(clippy::unused_unit)] #[impl_for_tuples(14)] @@ -187,8 +261,7 @@ where #[cfg(test)] mod test { use crate::{ - Circuit, FallbackZSet, Runtime, Stream, operator::Generator, typed_batch::OrdZSet, - utils::Tup2, zset, + Circuit, Runtime, Stream, operator::Generator, typed_batch::OrdZSet, utils::Tup2, zset, }; use std::{ thread, @@ -196,64 +269,6 @@ mod test { vec, }; - #[test] - fn reachability() { - let mut root = Runtime::init_circuit(1, move |circuit| { - // Changes to the edges relation. - let mut edges = vec![ - zset! { Tup2(1, 2) => 1 }, - zset! { Tup2(2, 3) => 1}, - zset! { Tup2(1, 3) => 1}, - zset! { Tup2(3, 1) => 1}, - zset! { Tup2(3, 1) => -1}, - zset! { Tup2(1, 2) => -1}, - zset! { Tup2(2, 4) => 1, Tup2(4, 1) => 1 }, - zset! { Tup2(2, 3) => -1, Tup2(3, 2) => 1 }, - ] - .into_iter(); - - // Expected content of the reachability relation. - let mut outputs = vec![ - zset! { Tup2(1, 2) => 1 }, - zset! { Tup2(1, 2) => 1, Tup2(2, 3) => 1, Tup2(1, 3) => 1 }, - zset! { Tup2(1, 2) => 1, Tup2(2, 3) => 1, Tup2(1, 3) => 1 }, - zset! { Tup2(1, 1) => 1, Tup2(2, 2) => 1, Tup2(3, 3) => 1, Tup2(1, 2) => 1, Tup2(1, 3) => 1, Tup2(2, 3) => 1, Tup2(2, 1) => 1, Tup2(3, 1) => 1, Tup2(3, 2) => 1}, - zset! { Tup2(1, 2) => 1, Tup2(2, 3) => 1, Tup2(1, 3) => 1 }, - zset! { Tup2(2, 3) => 1, Tup2(1, 3) => 1 }, - zset! { Tup2(1, 3) => 1, Tup2(2, 3) => 1, Tup2(2, 4) => 1, Tup2(2, 1) => 1, Tup2(4, 1) => 1, Tup2(4, 3) => 1 }, - zset! { Tup2(1, 1) => 1, Tup2(2, 2) => 1, Tup2(3, 3) => 1, Tup2(4, 4) => 1, - Tup2(1, 2) => 1, Tup2(1, 3) => 1, Tup2(1, 4) => 1, - Tup2(2, 1) => 1, Tup2(2, 3) => 1, Tup2(2, 4) => 1, - Tup2(3, 1) => 1, Tup2(3, 2) => 1, Tup2(3, 4) => 1, - Tup2(4, 1) => 1, Tup2(4, 2) => 1, Tup2(4, 3) => 1 }, - ] - .into_iter(); - - let edges = circuit - .add_source(Generator::new(move || edges.next().unwrap())); - - let paths = circuit.recursive(|child, paths: Stream<_, OrdZSet>>| { - let edges = edges.delta0(child); - - let paths_indexed = paths.map_index(|&Tup2(x, y)| (y, x)); - let edges_indexed = edges.map_index(|Tup2(x, y)| (*x, *y)); - - Ok(edges.plus(&paths_indexed.join(&edges_indexed, |_via, from, to| Tup2(*from, *to)))) - }) - .unwrap(); - - paths.integrate().stream_distinct().inspect(move |ps| { - assert_eq!(*ps, outputs.next().unwrap()); - }); - Ok(()) - }) - .unwrap().0; - - for _ in 0..8 { - root.transaction().unwrap(); - } - } - // See https://github.com/feldera/feldera/issues/4168 #[test] fn issue4168() { @@ -355,15 +370,14 @@ mod test { } } - // Somewhat lame multiple recursion example to test RecursiveStreams impl for - // tuples: compute forward and backward reachability at the same time. - #[test] - fn reachability2() { - type Edges = Stream>>; + mod reachability { + use super::*; + + type Edge = Tup2; - let mut root = Runtime::init_circuit(1, move |circuit| { - // Changes to the edges relation. - let mut edges = vec![ + /// Changes to the edges relation. + fn edges_data() -> Vec> { + vec![ zset! { Tup2(1, 2) => 1 }, zset! { Tup2(2, 3) => 1}, zset! { Tup2(1, 3) => 1}, @@ -373,59 +387,265 @@ mod test { zset! { Tup2(2, 4) => 1, Tup2(4, 1) => 1 }, zset! { Tup2(2, 3) => -1, Tup2(3, 2) => 1 }, ] - .into_iter(); + } - // Expected content of the reachability relation. - let output_vec = vec![ + /// Expected output to the reachable relation. + fn expected_reachable() -> Vec> { + vec![ zset! { Tup2(1, 2) => 1 }, zset! { Tup2(1, 2) => 1, Tup2(2, 3) => 1, Tup2(1, 3) => 1 }, zset! { Tup2(1, 2) => 1, Tup2(2, 3) => 1, Tup2(1, 3) => 1 }, - zset! { Tup2(1, 1) => 1, Tup2(2, 2) => 1, Tup2(3, 3) => 1, Tup2(1, 2) => 1, Tup2(1, 3) => 1, Tup2(2, 3) => 1, Tup2(2, 1) => 1, Tup2(3, 1) => 1, Tup2(3, 2) => 1}, + zset! { Tup2(1, 1) => 1, Tup2(2, 2) => 1, Tup2(3, 3) => 1, + Tup2(1, 2) => 1, Tup2(1, 3) => 1, Tup2(2, 3) => 1, + Tup2(2, 1) => 1, Tup2(3, 1) => 1, Tup2(3, 2) => 1}, zset! { Tup2(1, 2) => 1, Tup2(2, 3) => 1, Tup2(1, 3) => 1 }, zset! { Tup2(2, 3) => 1, Tup2(1, 3) => 1 }, - zset! { Tup2(1, 3) => 1, Tup2(2, 3) => 1, Tup2(2, 4) => 1, Tup2(2, 1) => 1, Tup2(4, 1) => 1, Tup2(4, 3) => 1 }, - zset! { Tup2(1, 1) => 1, Tup2(2, 2) => 1, Tup2(3, 3) => 1, Tup2(4, 4) => 1, - Tup2(1, 2) => 1, Tup2(1, 3) => 1, Tup2(1, 4) => 1, - Tup2(2, 1) => 1, Tup2(2, 3) => 1, Tup2(2, 4) => 1, - Tup2(3, 1) => 1, Tup2(3, 2) => 1, Tup2(3, 4) => 1, - Tup2(4, 1) => 1, Tup2(4, 2) => 1, Tup2(4, 3) => 1 }, - ]; - - let mut outputs = output_vec.clone().into_iter(); - let mut outputs2 = output_vec.into_iter(); - - let edges = circuit - .add_source(Generator::new(move || edges.next().unwrap())); - - let (paths, reverse_paths): (Stream<_, FallbackZSet>>, Stream<_, FallbackZSet>>) = - circuit.recursive(|child, (paths, reverse_paths): (Edges<_>, Edges<_>)| { - let edges = edges.delta0(child); - - let paths_indexed = paths.map_index(|&Tup2(x, y)| (y, x)); - let reverse_paths_indexed = reverse_paths.map_index(|&Tup2(x, y)| (y, x)); - let edges_indexed = edges.map_index(|Tup2(x,y)| (*x, *y)); - let reverse_edges = edges.map(|&Tup2(x, y)| Tup2(y, x)); - let reverse_edges_indexed = reverse_edges.map_index(|Tup2(x,y)| (*x, *y)); - - Ok((edges.plus(&paths_indexed.join(&edges_indexed, |_via, from, to| Tup2(*from, *to))), - reverse_edges.plus(&reverse_paths_indexed.join(&reverse_edges_indexed, |_via, from, to| Tup2(*from, *to))) - )) + zset! { Tup2(1, 3) => 1, Tup2(2, 3) => 1, Tup2(2, 4) => 1, + Tup2(2, 1) => 1, Tup2(4, 1) => 1, Tup2(4, 3) => 1 }, + zset! { Tup2(1, 1) => 1, Tup2(2, 2) => 1, Tup2(3, 3) => 1, + Tup2(4, 4) => 1, Tup2(1, 2) => 1, Tup2(1, 3) => 1, + Tup2(1, 4) => 1, Tup2(2, 1) => 1, Tup2(2, 3) => 1, + Tup2(2, 4) => 1, Tup2(3, 1) => 1, Tup2(3, 2) => 1, + Tup2(3, 4) => 1, Tup2(4, 1) => 1, Tup2(4, 2) => 1, + Tup2(4, 3) => 1 }, + ] + } + + #[test] + fn reachability() { + let edges_data = edges_data(); + let steps = edges_data.len(); + let mut edges = edges_data.into_iter(); + let mut expected_reachable = expected_reachable().into_iter(); + + let (mut handle, _) = Runtime::init_circuit(1, move |circuit| { + let edges = circuit.add_source(Generator::new(move || edges.next().unwrap())); + + let reachable = circuit + .recursive(|child, reachable: Stream<_, OrdZSet>| { + let edges = edges.delta0(child); + let edges_indexed = edges.map_index(|Tup2(x, y)| (*x, *y)); + + let reachable_indexed = reachable.map_index(|&Tup2(x, y)| (y, x)); + + let reachable_next = edges.plus( + &reachable_indexed + .join(&edges_indexed, |_via, from, to| Tup2(*from, *to)), + ); + + Ok(reachable_next) + }) + .unwrap(); + + reachable + .integrate() + .stream_distinct() + .inspect(move |reachable| { + assert_eq!(*reachable, expected_reachable.next().unwrap()); + }); + + Ok(()) }) .unwrap(); - paths.integrate().stream_distinct().inspect(move |ps| { - assert_eq!(*ps, outputs.next().unwrap()); - }); + for _ in 0..steps { + handle.transaction().unwrap(); + } + } - reverse_paths.map(|Tup2(x, y)| Tup2(*y, *x)).integrate().stream_distinct().inspect(move |ps: &OrdZSet<_>| { - assert_eq!(*ps, outputs2.next().unwrap()); - }); - Ok(()) - }) - .unwrap().0; + /// The `Vec` counterpart of [`reachability()`]: a single recursive relation + /// supplied as a one-element vector (arity 1). It must produce exactly the + /// same output as the single-`Stream` implementation. + #[test] + fn reachability_variadic() { + let edges_data = edges_data(); + let steps = edges_data.len(); + let mut edges = edges_data.into_iter(); + let mut expected_reachable = expected_reachable().into_iter(); + + let (mut handle, _) = Runtime::init_circuit(1, move |circuit| { + let edges = circuit.add_source(Generator::new(move || edges.next().unwrap())); + + let mut recursive_streams = circuit + .recursive_variadic( + 1, + |child, mut recursive_streams: Vec>>| { + let edges = edges.delta0(child); + let edges_indexed = edges.map_index(|Tup2(x, y)| (*x, *y)); + + let reachable = &mut recursive_streams[0]; + let reachable_indexed = reachable.map_index(|&Tup2(x, y)| (y, x)); + + let reachable_next = edges.plus( + &reachable_indexed + .join(&edges_indexed, |_via, from, to| Tup2(*from, *to)), + ); + + // We can even reuse the allocated vector and spare us a reallocation. + *reachable = reachable_next; + Ok(recursive_streams) + }, + ) + .unwrap(); + + let reachable = recursive_streams.pop().unwrap(); + + reachable.integrate().stream_distinct().inspect(move |ps| { + assert_eq!(*ps, expected_reachable.next().unwrap()); + }); + + Ok(()) + }) + .unwrap(); - for _ in 0..8 { - root.transaction().unwrap(); + for _ in 0..steps { + handle.transaction().unwrap(); + } + } + + // Somewhat lame multiple recursion example to test RecursiveStreams impl for + // tuples: compute forward and backward reachability at the same time. + #[test] + fn reachability2() { + let edges_data = edges_data(); + let steps = edges_data.len(); + let mut edges = edges_data.into_iter(); + let expected_reachable = expected_reachable(); + let expected_reachable_reverse = expected_reachable.clone(); + let mut expected_reachable = expected_reachable.into_iter(); + let mut expected_reachable_reverse = expected_reachable_reverse.into_iter(); + + let (mut root, _) = Runtime::init_circuit(1, move |circuit| { + let edges = circuit.add_source(Generator::new(move || edges.next().unwrap())); + + let (reachable, reachable_reverse) = circuit + .recursive( + |child, + (reachable, reachable_reverse): ( + Stream<_, OrdZSet>, + Stream<_, OrdZSet>, + )| { + let edges = edges.delta0(child); + + let edges_indexed = edges.map_index(|Tup2(x, y)| (*x, *y)); + let reachable_indexed = reachable.map_index(|&Tup2(x, y)| (y, x)); + let reachable_reverse_indexed = + reachable_reverse.map_index(|&Tup2(x, y)| (y, x)); + let reverse_edges = edges.map(|&Tup2(x, y)| Tup2(y, x)); + let reverse_edges_indexed = + reverse_edges.map_index(|Tup2(x, y)| (*x, *y)); + + let reachable_next = edges.plus( + &reachable_indexed + .join(&edges_indexed, |_via, from, to| Tup2(*from, *to)), + ); + let reachable_reverse_next = reverse_edges.plus( + &reachable_reverse_indexed + .join(&reverse_edges_indexed, |_via, from, to| { + Tup2(*from, *to) + }), + ); + + Ok((reachable_next, reachable_reverse_next)) + }, + ) + .unwrap(); + + reachable.integrate().stream_distinct().inspect(move |ps| { + assert_eq!(*ps, expected_reachable.next().unwrap()); + }); + + reachable_reverse + .map(|Tup2(x, y)| Tup2(*y, *x)) + .integrate() + .stream_distinct() + .inspect(move |ps: &OrdZSet<_>| { + assert_eq!(*ps, expected_reachable_reverse.next().unwrap()); + }); + + Ok(()) + }) + .unwrap(); + + for _ in 0..steps { + root.transaction().unwrap(); + } + } + + /// The `Vec` counterpart of [`reachability2()`]: forward and backward + /// reachability as two recursive relations supplied as a two-element + /// vector (arity 2). It must match the tuple implementation. + #[test] + fn reachability2_variadic() { + let edges_data = edges_data(); + let steps = edges_data.len(); + let mut edges = edges_data.into_iter(); + let expected_reachable = expected_reachable(); + let expected_reachable_reverse = expected_reachable.clone(); + let mut expected_reachable = expected_reachable.into_iter(); + let mut expected_reachable_reverse = expected_reachable_reverse.into_iter(); + + let (mut root, _) = Runtime::init_circuit(1, move |circuit| { + let edges = circuit.add_source(Generator::new(move || edges.next().unwrap())); + + let mut recursive_streams = circuit + .recursive_variadic( + 2, + |child, mut recursive_streams: Vec>>| { + let edges = edges.delta0(child); + + let (reachable, rest) = recursive_streams.split_first_mut().unwrap(); + let reachable_reverse = rest.first_mut().unwrap(); + + let edges_indexed = edges.map_index(|Tup2(x, y)| (*x, *y)); + let reachable_indexed = reachable.map_index(|&Tup2(x, y)| (y, x)); + let reachable_reverse_indexed = + reachable_reverse.map_index(|&Tup2(x, y)| (y, x)); + let reverse_edges = edges.map(|&Tup2(x, y)| Tup2(y, x)); + let reverse_edges_indexed = + reverse_edges.map_index(|Tup2(x, y)| (*x, *y)); + + let reachable_next = edges.plus( + &reachable_indexed + .join(&edges_indexed, |_via, from, to| Tup2(*from, *to)), + ); + + let reachable_reverse_next = reverse_edges.plus( + &reachable_reverse_indexed + .join(&reverse_edges_indexed, |_via, from, to| { + Tup2(*from, *to) + }), + ); + + // We can even reuse the allocated vector and spare us a reallocation. + *reachable = reachable_next; + *reachable_reverse = reachable_reverse_next; + Ok(recursive_streams) + }, + ) + .unwrap(); + + let reachable_reverse = recursive_streams.pop().unwrap(); + let reachable = recursive_streams.pop().unwrap(); + + reachable.integrate().stream_distinct().inspect(move |ps| { + assert_eq!(*ps, expected_reachable.next().unwrap()); + }); + reachable_reverse + .map(|Tup2(x, y)| Tup2(*y, *x)) + .integrate() + .stream_distinct() + .inspect(move |ps: &OrdZSet<_>| { + assert_eq!(*ps, expected_reachable_reverse.next().unwrap()); + }); + + Ok(()) + }) + .unwrap(); + + for _ in 0..steps { + root.transaction().unwrap(); + } } } } diff --git a/crates/dbsp/src/operator/recursive.rs b/crates/dbsp/src/operator/recursive.rs index 52263a0963a..5652d35a396 100644 --- a/crates/dbsp/src/operator/recursive.rs +++ b/crates/dbsp/src/operator/recursive.rs @@ -236,4 +236,183 @@ where }) .map(|streams| unsafe { S::typed_exports(&streams) }) } + + /// Like [`ChildCircuit::recursive`], but for a group of mutually recursive + /// streams whose size is only known at runtime. + /// + /// Whereas [`recursive`](ChildCircuit::recursive) fixes the number of + /// recursive streams at compile time (a single stream or a tuple of + /// streams), this method computes a fixed point over `arity` mutually + /// recursive streams that all share the same key type `K`, value type `V`, + /// and batch type `B`. The `arity` cannot be inferred, because the + /// recursive streams are the feedback Z-sets created *before* the closure + /// runs; it must therefore be supplied explicitly by the caller. + /// + /// The closure `f` receives a vector of `arity` recursive input streams and + /// must return a vector of exactly `arity` output streams, one per recursive + /// relation. Returning a vector of a different length panics in debug + /// builds and produces an incorrect circuit otherwise. + /// + /// # Examples + /// + /// The circuit below computes a two-coloring (red and blue) of a graph. If + /// no node is both red and blue the graph happens to be bipartite. In the + /// first two computation steps the graph is bipartite but the added edge + /// in the third step adds an odd-length cycle which destroys the bipartite + /// property and all nodes are colored red and blue. + /// + /// ``` + /// use dbsp::{ + /// operator::Generator, + /// OrdZSet, Circuit, RootCircuit, Stream, zset, ZWeight, + /// utils::Tup2, Error as DbspError, Runtime, NestedCircuit + /// }; + /// + /// type Edge = Tup2; + /// type Node = usize; + /// + /// const STEPS: usize = 3; + /// + /// let mut init_data = ([ + /// vec![Tup2(0, 1)], + /// vec![], + /// vec![] + /// ] as [Vec>; STEPS]).into_iter(); + /// + /// let mut edges_data = ([ + /// // The first step adds a graph of four nodes: + /// // |0| --> |1| --> |2| --> |3| --> |4| + /// vec![ + /// Tup2(Tup2(0, 1), 1), + /// Tup2(Tup2(1, 2), 1), + /// Tup2(Tup2(2, 3), 1), + /// Tup2(Tup2(3, 4), 1), + /// ], + /// // Now, we have the following graph in total: + /// // |0| --> |1| --> |2| --> |3| --> |4| + /// // ^ | + /// // | | + /// // ------ |5| <----- + /// vec![Tup2(Tup2(2, 5), 1), Tup2(Tup2(5, 0), 1)], + /// // And we introduce an odd-length cycle, rendering the graph + /// // non-biparite anymore (all nodes are red _and_ blue): + /// // |0| --> |1| --> |2| --> |3| --> |4| + /// // ^ | | + /// // | | | + /// // ------ |5| <----- | + /// // | | + /// // --------------------------------- + /// vec![Tup2(Tup2(4, 0), 1)], + /// ] as [Vec>; STEPS]).into_iter(); + /// + /// let mut expected_red_output = ([ + /// zset! { + /// 0 => 1, + /// 2 => 1, + /// 4 => 1, + /// }, + /// zset! {}, + /// zset! { + /// 1 => 1, + /// 3 => 1, + /// 5 => 1, + /// }, + /// ] as [OrdZSet; STEPS]).into_iter(); + /// + /// let mut expected_blue_output = ([ + /// zset! { + /// 1 => 1, + /// 3 => 1, + /// }, + /// zset! { + /// 5 => 1, + /// }, + /// zset! { + /// 0 => 1, + /// 2 => 1, + /// 4 => 1, + /// }, + /// ] as [OrdZSet; STEPS]).into_iter(); + /// + /// let (mut circuit_handle, ((init_input, edges_input), (red_output, blue_output))) = + /// Runtime::init_circuit(2, move |root_circuit| { + /// let (edges, edges_input) = root_circuit.add_input_zset::(); + /// let (init, init_input) = root_circuit.add_input_zset::(); + /// + /// let recursive_streams = root_circuit.recursive_variadic( + /// 2, + /// |child_circuit, mut recursive_streams: Vec>>| { + /// // delta0 fires only at inner step 0, injecting the base case exactly once. + /// let edges = edges.delta0(child_circuit); + /// let init = init.delta0(child_circuit); + /// + /// let red = &recursive_streams[0]; + /// let blue = &recursive_streams[1]; + /// + /// let new_red = blue + /// .map_index(|blue_node| (*blue_node, *blue_node)) + /// .join( + /// &edges.map_index(|Tup2(from, to)| (*from, *to)), + /// |_blue_node, _, new_red_node| *new_red_node, + /// ) + /// .plus(&init); + /// + /// let new_blue = red.map_index(|red_node| (*red_node, *red_node)).join( + /// &edges.map_index(|Tup2(from, to)| (*from, *to)), + /// |_red_node, _, new_blue_node| *new_blue_node, + /// ); + /// + /// recursive_streams[0] = new_red; + /// recursive_streams[1] = new_blue; + /// Ok(recursive_streams) + /// }, + /// )?; + /// + /// let red_output = recursive_streams[0].accumulate_output(); + /// let blue_output = recursive_streams[1].accumulate_output(); + /// + /// Ok(( + /// (init_input, edges_input), + /// (red_output, blue_output), + /// )) + /// })?; + /// + /// for i in 0..STEPS { + /// init_input.append(&mut init_data.next().unwrap()); + /// edges_input.append(&mut edges_data.next().unwrap()); + /// circuit_handle.transaction().unwrap(); + /// assert_eq!(red_output.concat().consolidate(), expected_red_output.next().unwrap()); + /// assert_eq!(blue_output.concat().consolidate(), expected_blue_output.next().unwrap()); + /// } + /// + /// Ok::<(), DbspError>(()) + /// ``` + #[track_caller] + pub fn recursive_variadic( + &self, + arity: usize, + f: F, + ) -> Result>>, SchedulerError> + where + B: Checkpoint + DynIndexedZSet + Send + Sync, + K: DBData + Erase, + V: DBData + Erase, + F: FnOnce( + &IterativeCircuit, + Vec, TypedBatch>>, + ) -> Result< + Vec, TypedBatch>>, + SchedulerError, + >, + { + let factories: Vec> = (0..arity) + .map(|_| DistinctFactories::new::()) + .collect(); + + self.dyn_recursive(&factories, |circuit, streams: Vec>| { + let typed = streams.iter().map(Stream::typed).collect(); + f(circuit, typed).map(|streams| streams.iter().map(Stream::inner).collect()) + }) + .map(|exports| exports.iter().map(Stream::typed).collect()) + } } From 02db767e60120510ec0e6fe123cf2783152ed13a Mon Sep 17 00:00:00 2001 From: Leo Stewen Date: Thu, 2 Jul 2026 16:51:36 +0200 Subject: [PATCH 2/3] Fix broken doc tests with rustc > 1.96.0 --- crates/dbsp/src/operator/recursive.rs | 6 +++--- crates/dbsp/src/tutorial.rs | 13 +++++++------ 2 files changed, 10 insertions(+), 9 deletions(-) diff --git a/crates/dbsp/src/operator/recursive.rs b/crates/dbsp/src/operator/recursive.rs index 5652d35a396..548b69ba5e4 100644 --- a/crates/dbsp/src/operator/recursive.rs +++ b/crates/dbsp/src/operator/recursive.rs @@ -169,7 +169,7 @@ where /// zset_set! { Tup2(4, 5) }, /// // Remove an edge, breaking the cycle. /// zset! { Tup2(1, 2) => -1 }, - /// ] as [_; STEPS]) + /// ] as [OrdZSet>; STEPS]) /// .into_iter(); /// /// let edges = root_circuit @@ -182,7 +182,7 @@ where /// // Add a label to node 2. /// zset_set! { Tup2(2, "l2".to_string()) }, /// zset! { }, - /// ] as [_; STEPS]) + /// ] as [OrdZSet>; STEPS]) /// .into_iter(); /// /// let init_labels = root_circuit @@ -193,7 +193,7 @@ where /// zset! { Tup2(1, "l1".to_string()) => 1, Tup2(2, "l1".to_string()) => 1, Tup2(3, "l1".to_string()) => 1, Tup2(4, "l1".to_string()) => 1 }, /// zset! { Tup2(1, "l2".to_string()) => 1, Tup2(2, "l2".to_string()) => 1, Tup2(3, "l2".to_string()) => 1, Tup2(4, "l2".to_string()) => 1, Tup2(5, "l1".to_string()) => 1, Tup2(5, "l2".to_string()) => 1 }, /// zset! { Tup2(2, "l1".to_string()) => -1, Tup2(3, "l1".to_string()) => -1, Tup2(4, "l1".to_string()) => -1, Tup2(5, "l1".to_string()) => -1 }, - /// ] as [_; STEPS]) + /// ] as [OrdZSet>; STEPS]) /// .into_iter(); /// /// let labels = root_circuit.recursive(|child_circuit, labels: Stream<_, OrdZSet>>| { diff --git a/crates/dbsp/src/tutorial.rs b/crates/dbsp/src/tutorial.rs index 2c4d5887693..a4caafd4f24 100644 --- a/crates/dbsp/src/tutorial.rs +++ b/crates/dbsp/src/tutorial.rs @@ -2056,9 +2056,9 @@ //! //! let (mut circuit_handle, output_handle) = Runtime::init_circuit(1, move |root_circuit| { //! let mut edges_data = ([ -//! zset_set! { Tup3(0_usize, 1_usize, 1_usize), Tup3(1, 2, 1), Tup3(2, 3, 2), Tup3(3, 4, 2) }, +//! zset_set! { Tup3(0, 1, 1), Tup3(1, 2, 1), Tup3(2, 3, 2), Tup3(3, 4, 2) }, //! zset! { Tup3(1, 2, 1) => -1 }, -//! ] as [_; STEPS]) +//! ] as [OrdZSet>; STEPS]) //! .into_iter(); //! //! let edges = root_circuit.add_source(Generator::new(move || edges_data.next().unwrap())); @@ -2119,7 +2119,7 @@ //! Tup4(1, 3, 3, 2) => -1, //! Tup4(1, 4, 5, 3) => -1, //! }, -//! ] as [_; STEPS]) +//! ] as [OrdZSet>; STEPS]) //! .into_iter(); //! //! closure.inspect(move |output| { @@ -2182,7 +2182,8 @@ //! # indexed_zset, //! # operator::{Generator, Min}, //! # utils::{Tup2, Tup3, Tup4}, -//! # zset_set, Circuit, NestedCircuit, OrdIndexedZSet, RootCircuit, Stream, IndexedZSetReader, Runtime +//! # zset_set, Circuit, NestedCircuit, OrdZSet, OrdIndexedZSet, RootCircuit, +//! # Stream, IndexedZSetReader, Runtime //! # }; //! # //! type Accumulator = @@ -2193,9 +2194,9 @@ //! # //! # let (mut circuit_handle, output_handle) = Runtime::init_circuit(1, move |root_circuit| { //! # let mut edges_data = ([ -//! # zset_set! { Tup3(0_usize, 1_usize, 1_usize), Tup3(1, 2, 1), Tup3(2, 3, 2), Tup3(3, 4, 2) }, +//! # zset_set! { Tup3(0, 1, 1), Tup3(1, 2, 1), Tup3(2, 3, 2), Tup3(3, 4, 2) }, //! # zset_set! { Tup3(4, 0, 3)} -//! # ] as [_; STEPS]) +//! # ] as [OrdZSet>; STEPS]) //! # .into_iter(); //! # //! # let edges = root_circuit.add_source(Generator::new(move || edges_data.next().unwrap())); From 7b913cb0cadf586004b7c96d4ce5635755caa102 Mon Sep 17 00:00:00 2001 From: Leo Stewen Date: Fri, 3 Jul 2026 10:52:08 +0200 Subject: [PATCH 3/3] Address PR feedback --- crates/dbsp/src/operator/dynamic/recursive.rs | 39 +++++++++++-------- crates/dbsp/src/operator/recursive.rs | 18 ++++++--- 2 files changed, 35 insertions(+), 22 deletions(-) diff --git a/crates/dbsp/src/operator/dynamic/recursive.rs b/crates/dbsp/src/operator/dynamic/recursive.rs index 188aa356a08..1af7c633a23 100644 --- a/crates/dbsp/src/operator/dynamic/recursive.rs +++ b/crates/dbsp/src/operator/dynamic/recursive.rs @@ -128,21 +128,21 @@ where } fn distinct(mut self, factories: &Self::Factories) -> Self { - debug_assert_eq!(self.len(), factories.len()); + assert_eq!(self.len(), factories.len()); for (stream, factory) in self.iter_mut().zip(factories) { let persistent_id = stream .get_persistent_id() .map(|name| format!("{name}.distinct")); *stream = - Stream::dyn_distinct(&stream, factory).set_persistent_id(persistent_id.as_deref()); + Stream::dyn_distinct(stream, factory).set_persistent_id(persistent_id.as_deref()); } self } fn connect(&self, vars: Self::Feedback) { - debug_assert_eq!(self.len(), vars.len()); + assert_eq!(self.len(), vars.len()); for (stream, var) in self.iter().zip(vars) { var.connect(stream); @@ -150,7 +150,7 @@ where } fn export(self, factories: &Self::Factories) -> Self::Export { - debug_assert_eq!(self.len(), factories.len()); + assert_eq!(self.len(), factories.len()); self.into_iter() .zip(factories) @@ -161,7 +161,7 @@ where } fn consolidate(exports: Self::Export, factories: &Self::Factories) -> Self::Output { - debug_assert_eq!(exports.len(), factories.len()); + assert_eq!(exports.len(), factories.len()); exports .into_iter() @@ -372,6 +372,7 @@ mod test { mod reachability { use super::*; + use crate::FallbackZSet; type Edge = Tup2; @@ -453,11 +454,13 @@ mod test { } } - /// The `Vec` counterpart of [`reachability()`]: a single recursive relation - /// supplied as a one-element vector (arity 1). It must produce exactly the - /// same output as the single-`Stream` implementation. + /// A rewrite of [`reachability()`] using + /// [`recursive_dynamic`](crate::ChildCircuit::recursive_dynamic): + /// A single recursive relation supplied as a one-element vector + /// (arity 1). It must produce exactly the same output as the + /// single-`Stream` implementation. #[test] - fn reachability_variadic() { + fn reachability_dynamic() { let edges_data = edges_data(); let steps = edges_data.len(); let mut edges = edges_data.into_iter(); @@ -467,7 +470,7 @@ mod test { let edges = circuit.add_source(Generator::new(move || edges.next().unwrap())); let mut recursive_streams = circuit - .recursive_variadic( + .recursive_dynamic( 1, |child, mut recursive_streams: Vec>>| { let edges = edges.delta0(child); @@ -522,8 +525,8 @@ mod test { .recursive( |child, (reachable, reachable_reverse): ( - Stream<_, OrdZSet>, - Stream<_, OrdZSet>, + Stream<_, FallbackZSet>, + Stream<_, FallbackZSet>, )| { let edges = edges.delta0(child); @@ -572,11 +575,13 @@ mod test { } } - /// The `Vec` counterpart of [`reachability2()`]: forward and backward - /// reachability as two recursive relations supplied as a two-element - /// vector (arity 2). It must match the tuple implementation. + /// A rewrite of [`reachability2()`] using + /// [`recursive_dynamic`](crate::ChildCircuit::recursive_dynamic): + /// Forward and backward reachability as two recursive relations + /// supplied as a two-element vector (arity 2). It must match the + /// tuple implementation. #[test] - fn reachability2_variadic() { + fn reachability2_dynamic() { let edges_data = edges_data(); let steps = edges_data.len(); let mut edges = edges_data.into_iter(); @@ -589,7 +594,7 @@ mod test { let edges = circuit.add_source(Generator::new(move || edges.next().unwrap())); let mut recursive_streams = circuit - .recursive_variadic( + .recursive_dynamic( 2, |child, mut recursive_streams: Vec>>| { let edges = edges.delta0(child); diff --git a/crates/dbsp/src/operator/recursive.rs b/crates/dbsp/src/operator/recursive.rs index 548b69ba5e4..1538a738682 100644 --- a/crates/dbsp/src/operator/recursive.rs +++ b/crates/dbsp/src/operator/recursive.rs @@ -250,8 +250,16 @@ where /// /// The closure `f` receives a vector of `arity` recursive input streams and /// must return a vector of exactly `arity` output streams, one per recursive - /// relation. Returning a vector of a different length panics in debug - /// builds and produces an incorrect circuit otherwise. + /// relation. + /// + /// Similar to [`recursive`](ChildCircuit::recursive), the underlying + /// circuit also applies an implicit distinct to the output of each + /// recursive step. + /// + /// # Panics + /// + /// Panics if the returned vector from the closure parameter has a different + /// length than the `arity` parameter. /// /// # Examples /// @@ -295,7 +303,7 @@ where /// // ------ |5| <----- /// vec![Tup2(Tup2(2, 5), 1), Tup2(Tup2(5, 0), 1)], /// // And we introduce an odd-length cycle, rendering the graph - /// // non-biparite anymore (all nodes are red _and_ blue): + /// // non-bipartite anymore (all nodes are red _and_ blue): /// // |0| --> |1| --> |2| --> |3| --> |4| /// // ^ | | /// // | | | @@ -339,7 +347,7 @@ where /// let (edges, edges_input) = root_circuit.add_input_zset::(); /// let (init, init_input) = root_circuit.add_input_zset::(); /// - /// let recursive_streams = root_circuit.recursive_variadic( + /// let recursive_streams = root_circuit.recursive_dynamic( /// 2, /// |child_circuit, mut recursive_streams: Vec>>| { /// // delta0 fires only at inner step 0, injecting the base case exactly once. @@ -388,7 +396,7 @@ where /// Ok::<(), DbspError>(()) /// ``` #[track_caller] - pub fn recursive_variadic( + pub fn recursive_dynamic( &self, arity: usize, f: F,