feldera · mihaibudiu · Aug 4, 2026 · Aug 3, 2026
diff --git a/python/tests/runtime/test_recursive_view_checkpoint.py b/python/tests/runtime/test_recursive_view_checkpoint.py
@@ -4,9 +4,33 @@
 used to be missing from checkpoints: the restart reported neither an error nor a
 bootstrap, and the view then produced wrong results
 (https://github.com/feldera/feldera/issues/6765).
+
+Two properties of these tests are what make them catch that:
+
+* **The program does not change across the restart.**  When it changes, the
+  bootstrap replay rebuilds the recursive view from replayed input and hides the
+  loss.  That is why the runtime-upgrade tests never caught this: they restart
+  into a program the bootstrap diff marks as modified, and so always bootstrap.
+* **``edges`` is not materialized**, so there is no replay source for it.
+  Nothing can reconstruct the relation from input history; the state either
+  comes from the checkpoint or it is gone.
+
+The decisive assertion is the one that feeds another edge after the restart.
+A view's own contents live outside the recursive scope and come back either
+way; deriving new paths through the older edges is what needs the state inside
+the scope.  Verified by reverting the fix: the closure then grows by the new
+edge alone, missing every path that runs through it.
+
+A scope with more than one recursive view exercises a second failure, in the
+compiler rather than the runtime
+(https://github.com/feldera/feldera/issues/6792): the checkpoint itself fails
+with ``NoPersistentId`` because the ``z^-1`` operator behind a recursive view
+that the scope does not read through the delay was left unnamed.
 """
 
-from feldera import PipelineBuilder
+from typing import Callable
+
+from feldera import Pipeline, PipelineBuilder
 from feldera.runtime_config import RuntimeConfig
 from feldera.testutils import FELDERA_TEST_NUM_HOSTS, FELDERA_TEST_NUM_WORKERS
 from tests import TEST_CLIENT, enterprise_only
@@ -25,8 +49,31 @@
   (SELECT e.a, c.b FROM edges e JOIN closure c ON e.b = c.a);
 """
 
+# The same closure, split over two mutually recursive views.  Breaking the cycle
+# takes a single delay, so the compiler keeps the declaration of one view and
+# drops the other's: the scope receives a recursive stream that no declaration
+# names, and the operator that closes its loop needs a name all the same.
+MUTUAL_SQL = """
+CREATE TABLE edges (a INT NOT NULL, b INT NOT NULL);
+
+DECLARE RECURSIVE VIEW reachable(a INT NOT NULL, b INT NOT NULL);
+DECLARE RECURSIVE VIEW hops(a INT NOT NULL, b INT NOT NULL);
+
+-- Paths of two edges or more.
+CREATE MATERIALIZED VIEW hops AS
+  SELECT e.a, r.b FROM edges e JOIN reachable r ON e.b = r.a;
+
+-- Paths of one edge or more, i.e. the transitive closure.
+CREATE MATERIALIZED VIEW reachable AS
+  (SELECT a, b FROM edges) UNION (SELECT a, b FROM hops);
+"""
 
-def transitive_closure(edges: list[tuple[int, int]]) -> list[tuple[int, int]]:
+Edges = list[tuple[int, int]]
+# Asserts that every view of the program holds what `edges` implies.
+Check = Callable[[Pipeline, Edges], None]
+
+
+def transitive_closure(edges: Edges) -> Edges:
     """Oracle: the transitive closure of `edges`, computed outside Feldera."""
     closure = set(edges)
     while True:
@@ -36,25 +83,31 @@ def transitive_closure(edges: list[tuple[int, int]]) -> list[tuple[int, int]]:
         closure = grown
 
 
-def query_closure(pipeline) -> list[tuple[int, int]]:
+def long_paths(edges: Edges) -> Edges:
+    """Oracle: the pairs that `edges` connects with two edges or more."""
+    closure = set(transitive_closure(edges))
+    return sorted({(a, d) for (a, b) in edges for (c, d) in closure if b == c})
+
+
+def query_pairs(pipeline: Pipeline, view: str) -> Edges:
     return sorted(
-        (row["a"], row["b"]) for row in pipeline.query("SELECT a, b FROM closure;")
+        (row["a"], row["b"]) for row in pipeline.query(f"SELECT a, b FROM {view};")
     )
 
 
-def insert_edges(pipeline, edges: list[tuple[int, int]]) -> None:
+def insert_edges(pipeline: Pipeline, edges: Edges) -> None:
     """Inserts `edges` and waits for the pipeline to finish processing them."""
     values = ", ".join(f"({a}, {b})" for a, b in edges)
     pipeline.execute(f"INSERT INTO edges VALUES {values};", wait=True)
 
 
-@enterprise_only
-@gen_pipeline_name
-def test_recursive_view_survives_restart(pipeline_name: str) -> None:
+def checkpoint_and_restart(pipeline_name: str, sql: str, check: Check) -> None:
+    """Runs `sql`, checkpoints it, restarts it from that checkpoint, and keeps
+    feeding it edges, requiring `check` to hold at every step."""
     pipeline = PipelineBuilder(
         TEST_CLIENT,
         name=pipeline_name,
-        sql=SQL,
+        sql=sql,
         runtime_config=RuntimeConfig(
             workers=FELDERA_TEST_NUM_WORKERS,
             hosts=FELDERA_TEST_NUM_HOSTS,
@@ -69,7 +122,7 @@ def test_recursive_view_survives_restart(pipeline_name: str) -> None:
         # A chain 0 -> 1 -> 2 -> 3.
         edges = [(0, 1), (1, 2), (2, 3)]
         insert_edges(pipeline, edges)
-        assert query_closure(pipeline) == transitive_closure(edges)
+        check(pipeline, edges)
 
         pipeline.checkpoint(wait=True)
         # Stop without clearing storage, so the checkpoint survives.
@@ -78,22 +131,41 @@ def test_recursive_view_survives_restart(pipeline_name: str) -> None:
         # Restart the same program at the same runtime version.
         pipeline.start()
 
-        # The view's contents are stored outside the recursive scope, so a
+        # The views' contents are stored outside the recursive scope, so a
         # mismatch here means the checkpoint was not loaded at all.
-        assert query_closure(pipeline) == transitive_closure(edges)
+        check(pipeline, edges)
 
         # Extending the chain derives paths that run through the edges fed
         # before the restart, which only the recursive scope's restored state
         # supplies.
         edges.append((3, 4))
         insert_edges(pipeline, [(3, 4)])
-        assert query_closure(pipeline) == transitive_closure(edges)
+        check(pipeline, edges)
 
         # Closing a cycle exercises that state again, deriving paths in both
         # directions.
         edges.append((4, 0))
         insert_edges(pipeline, [(4, 0)])
-        assert query_closure(pipeline) == transitive_closure(edges)
+        check(pipeline, edges)
     finally:
         pipeline.stop(force=True)
         pipeline.clear_storage()
+
+
+@enterprise_only
+@gen_pipeline_name
+def test_recursive_view_survives_restart(pipeline_name: str) -> None:
+    def check(pipeline: Pipeline, edges: Edges) -> None:
+        assert query_pairs(pipeline, "closure") == transitive_closure(edges)
+
+    checkpoint_and_restart(pipeline_name, SQL, check)
+
+
+@enterprise_only
+@gen_pipeline_name
+def test_mutually_recursive_views_survive_restart(pipeline_name: str) -> None:
+    def check(pipeline: Pipeline, edges: Edges) -> None:
+        assert query_pairs(pipeline, "reachable") == transitive_closure(edges)
+        assert query_pairs(pipeline, "hops") == long_paths(edges)
+
+    checkpoint_and_restart(pipeline_name, MUTUAL_SQL, check)
diff --git a/...ompiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/MerkleOuter.java b/...ompiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/MerkleOuter.java
@@ -1,6 +1,7 @@
 package org.dbsp.sqlCompiler.compiler.backend;
 
 import org.dbsp.sqlCompiler.circuit.OutputPort;
+import org.dbsp.sqlCompiler.circuit.annotation.CompactName;
 import org.dbsp.sqlCompiler.circuit.annotation.OperatorHash;
 import org.dbsp.sqlCompiler.circuit.operator.DBSPNestedOperator;
 import org.dbsp.sqlCompiler.circuit.operator.DBSPOperator;
@@ -70,10 +71,13 @@ void setHashedString(DBSPOperator operator, String string) {
         if (this.includeInputs && this.recursiveOutputs.contains(operator.getId()))
             string += this.recursiveStateVersion;
         HashString hash = MerkleInner.hash(string);
+        String name = CompactName.getCompactName(operator);
         Logger.INSTANCE.belowLevel(this, 1)
                 .append(this.includeInputs ? "Global " : "")
                 .append("Merkle hash of ")
-                .append(operator.id);
+                .append(operator.id)
+                .append(" ")
+                .append(name != null ? name : "");
         Logger.INSTANCE.belowLevel(this, 2)
                 .append(" from").newline()
                 .append(string).newline();

diff --git a/.../SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java b/.../SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java
@@ -445,8 +445,16 @@ public VisitDecision preorder(DBSPNestedOperator operator) {
             if (decl != null) {
                 this.builder.append(decl.getNodeName(this.preferHash)).append(", ");
             } else {
-                // view is not really recursive
-                this.builder.append("_").append(", ");
+                // view is not really recursive.
+                if (operator.internalOutputs.get(i) != null) {
+                    // It is not used in recursion,
+                    // but it must be an output of the recursive component, and we
+                    // want to assign it a persistent ID.
+                    this.builder.append("unused_").append(i).append(", ");
+                } else {
+                    // This output doesn't even exist
+                    this.builder.append("_, ");
+                }
             }
         }
         this.builder.append("): (");
@@ -471,11 +479,27 @@ public VisitDecision preorder(DBSPNestedOperator operator) {
         for (int i = 0; i < operator.outputCount(); i++) {
             ProgramIdentifier view = operator.outputViews.get(i);
             DBSPViewDeclarationOperator decl = operator.declarationByName.get(view);
+            OutputPort port = operator.internalOutputs.get(i);
             if (decl != null) {
                 this.computeHash(decl);
                 this.tagStream(decl);
-                this.builder.newline();
+            } else if (port != null) {
+                this.builder.append("let hash = ");
+                HashString hash = OperatorHash.getHash(port.operator, true);
+                if (hash == null) {
+                    this.builder.append("None;").newline();
+                } else {
+                    this.builder.append("Some(concat!(")
+                            .append(hash.toQuotedString())
+                            .append(", \".delay\"));")
+                            .newline();
+                }
+                this.builder
+                        .append("unused_")
+                        .append(i)
+                        .append(".set_persistent_id(hash);");
             }
+            this.builder.newline();
         }
 
         for (IDBSPNode node : operator.getAllOperators())
@@ -607,7 +631,7 @@ void computeHash(DBSPOperator operator) {
             this.builder.append("None;").newline();
         } else {
             this.builder.append("Some(")
-                    .append(Utilities.doubleQuote(hash.toString(), true))
+                    .append(hash.toQuotedString())
                     .append(");")
                     .newline();
         }

diff --git a/.../src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/multi/NestedOperatorWriter.java b/.../src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/multi/NestedOperatorWriter.java
@@ -122,7 +122,7 @@ private void setPersistentId(DBSPOperator operator, String name) {
             this.builder().append("None;").newline();
         } else {
             this.builder().append("Some(")
-                    .append(Utilities.doubleQuote(hash.toString(), false))
+                    .append(hash.toQuotedString())
                     .append(");")
                     .newline();
         }
@@ -195,8 +195,16 @@ public void write(DBSPCompiler compiler) {
             if (decl != null) {
                 this.builder().append(decl.getNodeName(false)).append(", ");
             } else {
-                // view is not really recursive
-                this.builder().append("_").append(", ");
+                // view is not really recursive.
+                if (operator.internalOutputs.get(i) != null) {
+                    // It is not used in recursion,
+                    // but it must be an output of the recursive component, and we
+                    // want to assign it a persistent ID.
+                    this.builder().append("unused_").append(i).append(", ");
+                } else {
+                    // This output doesn't even exist
+                    this.builder().append("_, ");
+                }
             }
         }
         this.builder().append("): (");
@@ -219,8 +227,25 @@ public void write(DBSPCompiler compiler) {
         for (int i = 0; i < operator.outputCount(); i++) {
             ProgramIdentifier view = operator.outputViews.get(i);
             DBSPViewDeclarationOperator decl = operator.declarationByName.get(view);
-            if (decl != null)
+            OutputPort port = operator.internalOutputs.get(i);
+            if (decl != null) {
                 this.setPersistentId(decl, decl.getNodeName(false));
+            } else if (port != null) {
+                HashString hash0 = OperatorHash.getHash(port.operator, true);
+                if (hash0 == null) {
+                    this.builder().append("let hash = None;");
+                } else {
+                    this.builder().append("let hash = Some(concat!(")
+                            .append(hash0.toQuotedString())
+                            .append(", \".delay\"));");
+                }
+                this.builder()
+                        .newline()
+                        .append("unused_")
+                        .append(i)
+                        .append(".set_persistent_id(hash);")
+                        .newline();
+            }
         }
 
         for (DBSPOperator node : this.operator.getAllOperators())

diff --git a/...L-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitPostfix.java b/...L-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitPostfix.java
@@ -36,7 +36,7 @@ public void emit(IIndentStream builder) {
             builder.append("circuit.set_balancer_hint(").increase();
             var input = this.strategy.input == 0 ? this.operator.left() : operator.right();
             var inputHash = OperatorHash.getHash(input.node(), true);
-            builder.append(Utilities.doubleQuote(Objects.requireNonNull(inputHash).toString(), true))
+            builder.append(Objects.requireNonNull(inputHash).toQuotedString())
                     .append(",").newline();
             builder.append(this.strategy.toRust()).newline();
             builder.decrease().append(")?;").newline();

diff --git a/...QL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/ToJsonVisitor.java b/...QL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/ToJsonVisitor.java
@@ -177,7 +177,7 @@ void process(DBSPSimpleOperator operator) {
         HashString hash = OperatorHash.getHash(operator, true);
         if (hash != null) {
             this.builder.appendJsonLabelAndColon("persistent_id");
-            this.builder.append(Utilities.doubleQuote(hash.toString(), false)).newline();
+            this.builder.append(hash.toQuotedString()).newline();
         }
         this.builder.decrease().append("}");
     }
@@ -224,7 +224,7 @@ void processWithError(DBSPOperatorWithError operator) {
         HashString hash = OperatorHash.getHash(operator, true);
         if (hash != null) {
             this.builder.appendJsonLabelAndColon("persistent_id");
-            this.builder.append(Utilities.doubleQuote(hash.toString(), false)).newline();
+            this.builder.append(hash.toQuotedString()).newline();
         }
         this.builder.decrease().append("}");
     }
@@ -271,7 +271,7 @@ void processInputMapWithWaterline(DBSPInputMapWithWaterlineOperator operator) {
         HashString hash = OperatorHash.getHash(operator, true);
         if (hash != null) {
             this.builder.appendJsonLabelAndColon("persistent_id");
-            this.builder.append(Utilities.doubleQuote(hash.toString(), false)).newline();
+            this.builder.append(hash.toQuotedString()).newline();
         }
         this.builder.decrease().append("}");
     }

diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/util/HashString.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/util/HashString.java
@@ -22,4 +22,8 @@ public String makeIdentifier(@Nullable String prefix) {
             prefix = "s";
         return prefix + "_" + this.shortString();
     }
+
+    public String toQuotedString() {
+        return Utilities.doubleQuote(this.toString(), false);
+    }
 }