[DBSP] Implement RecursiveStreams on Vec<Stream<_, _>> by lstwn · Pull Request #6577 · feldera/feldera

lstwn · 2026-07-02T15:20:57Z

This implements recursion for a variadic number of recursive streams which has been impossible with the existing API and happened to be a marked as TODO in the codebase already. Previously you were required to know the number of recursive streams at compile time (with the benefit of supporting heterogeneous stream types). With this change the number of recursive streams can be specified at runtime (via recursive_variadic(); with the downside of requiring the stream types to be homogeneous).

API Changes

I've implemented this as an additive change to the outward facing API such that no existing code is broken with this change. Users who want a variadic amount of recursive streams at runtime simply use the new recursive_variadic function.

Testing and Documentation

The change is covered by both extensive documentation and unit testing. The diff of the unit test is a bit bigger than what it actually is because I restructured it such that the input and output data is shared among both variants (static and variadic) of the two test cases reachable and reachable2, respectively.

Moreover, this PR also fixes doc tests for existing code with rustc >= 1.96.0.

Hope this matches what you've had in mind for the (missing) implementation, too.

mythical-fred

Nice, clean addition — the Vec<Stream> impl mirrors the tuple impl cleanly and the two new tests give solid coverage of the arity-1 and arity-2 cases against their tuple counterparts.

A few things to consider, all soft: (1) the arity checks in the Vec impl are debug_assert_eq!, and the rustdoc even flags that a release build silently constructs a wrong circuit — I'd make those unconditional. (2) the pre-PR reachability2 test used FallbackZSet for the recursive streams; after consolidation the whole file only exercises OrdZSet. (3) small prose/naming/commit-message nits inline.

No blockers. Approving pending your take on the arity assertions.

mythical-fred · 2026-07-02T16:09:50Z

+    }
+
+    fn connect(&self, vars: Self::Feedback) {
+        debug_assert_eq!(self.len(), vars.len());


The three debug_assert_eq! arity checks in this impl (here, line 131, line 153, line 164) are the only guard against a user closure returning a Vec of the wrong length. Your own rustdoc on recursive_variadic calls it out: "panics in debug builds and produces an incorrect circuit otherwise." Silently building a wrong circuit in release is worse than a panic. These checks fire once per circuit build (not per fixpoint iteration), so making them unconditional assert_eq! has effectively zero cost — and it turns a silent correctness bug into a loud one. Please promote all four.

mythical-fred · 2026-07-02T16:09:50Z

+        // Somewhat lame multiple recursion example to test RecursiveStreams impl for
+        // tuples: compute forward and backward reachability at the same time.
+        #[test]
+        fn reachability2() {


Coverage nit: the pre-PR reachability2 typed the two recursive streams as Stream<_, FallbackZSet<Tup2<u64, u64>>> and the use at the top of the module still pulled in FallbackZSet. After the refactor the whole file only exercises OrdZSet, so we've quietly dropped the FallbackZSet code path in the recursion tests. Either flip one of the four reachability* tests back to FallbackZSet<Edge> or add a short comment noting the backing store is covered elsewhere — whichever you prefer.

I've restored reachability2 to use the FallbackZSet instead of OrdZSet. That wasn't on purpose, I do not really know the difference between the two.

mythical-fred · 2026-07-02T16:09:50Z

+    /// 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.


"panics in debug builds and produces an incorrect circuit otherwise" is a self-diagnosed footgun — see the inline comment on the trait impl. If you promote those debug_asserts to asserts, this sentence should read "panics if the returned vector has a different length."

mythical-fred · 2026-07-02T16:09:50Z

+    ///     //  ------ |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):


Typo: non-biparite → non-bipartite.

you have thumbed-up this, but you haven't fixed it
Let me know when this is ready for a review

mythical-fred · 2026-07-02T16:09:50Z

+    /// Ok::<(), DbspError>(())
+    /// ```
+    #[track_caller]
+    pub fn recursive_variadic<F, K, V, B>(


Naming, minor: recursive_variadic reads to me more like "variadic arguments" (heterogeneous, arity known at compile time — i.e. the tuple impl) than "runtime-known homogeneous arity." recursive_dynamic, recursive_vec, or recursive_n would all be clearer to a reader who hasn't read this docstring. Not a blocker — flag for consideration.

I've settled on recursive_dynamic now.

mihaibudiu

This looks fine to me, but @ryzhyk is the definitive authority.
Thank you!

mihaibudiu · 2026-07-02T17:20:39Z

-                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,


I assume these are not changed; reformatting makes it hard to tell.

Their input output behavior is exactly the same as before. I've just restructured them a bit and accidentally used OrdZSet instead of FallbackZSet as pointed out by the AI above but this is now reverted to still use FallbackZSet in reachability2. What is the difference between the two?

The documentation for FallbackZSet explains that it is a wrapper which allows the ZSet to reside either in-memory or on-disk.

mihaibudiu · 2026-07-02T17:22:29Z

-            Ok(())
-        })
-        .unwrap().0;
+        /// The `Vec` counterpart of [`reachability()`]: a single recursive relation


A rewrite of reachability() using recursive_variadic

mihaibudiu · 2026-07-02T17:27:47Z

        .map(|streams| unsafe { S::typed_exports(&streams) })
    }
+
+    /// Like [`ChildCircuit::recursive`], but for a group of mutually recursive


So you have a use case in which you generate programs dynamically in Rust?

Exactly! I'm building an interpreter on top of DBSP.

mihaibudiu · 2026-07-02T17:28:20Z

+    ///
+    /// 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


Why not always panic, as Fred asked?

mihaibudiu · 2026-07-02T21:08:31Z

Our bot wants you to change this:

diff --git a/crates/dbsp/src/operator/dynamic/recursive.rs b/crates/dbsp/src/operator/dynamic/recursive.rs
index 188aa35..ae597b3 100644
--- a/crates/dbsp/src/operator/dynamic/recursive.rs
+++ b/crates/dbsp/src/operator/dynamic/recursive.rs
@@ -135,7 +135,7 @@ where
                 .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());
```

mythical-fred

Nits addressed — assert_eq! in the Vec impl, recursive_dynamic naming, reachability2 restored on FallbackZSet, typo fixed. LGTM.

lstwn · 2026-07-03T16:45:04Z

    ///     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):


@mihaibudiu It is fixed, I think. From my end, I'm ready to get this merged or for a final review!

mihaibudiu · 2026-07-03T16:51:01Z

I think something is broken in the new github UI, the comments look different depending on where you look at them.

mihaibudiu · 2026-07-03T16:52:47Z

I think this is safe to merge, even without @ryzhyk's review.

lstwn added 2 commits July 2, 2026 16:52

Implement RecursiveStreams for Vec<Stream<_>>

86ae2c3

Fix broken doc tests with rustc > 1.96.0

02db767

lstwn changed the title ~~Implement RecursiveStreams on Vec<Stream<_, _>>~~ [DBSP] Implement RecursiveStreams on Vec<Stream<_, _>> Jul 2, 2026

mythical-fred reviewed Jul 2, 2026

View reviewed changes

mihaibudiu approved these changes Jul 2, 2026

View reviewed changes

Address PR feedback

7b913cb

mythical-fred approved these changes Jul 3, 2026

View reviewed changes

lstwn commented Jul 3, 2026

View reviewed changes

mihaibudiu approved these changes Jul 3, 2026

View reviewed changes

mihaibudiu added this pull request to the merge queue Jul 3, 2026

Merged via the queue into feldera:main with commit 71dac83 Jul 3, 2026
1 check passed

lstwn mentioned this pull request Jul 17, 2026

[DBSP] Potential new API for Recursive Streams #6666

Open

Uh oh!

Conversation

lstwn commented Jul 2, 2026

API Changes

Testing and Documentation

Uh oh!

mythical-fred left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mihaibudiu left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mihaibudiu Jul 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mihaibudiu commented Jul 2, 2026

Uh oh!

mythical-fred left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mihaibudiu commented Jul 3, 2026

Uh oh!

mihaibudiu commented Jul 3, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mihaibudiu Jul 3, 2026 •

edited

Loading