diff --git a/python/tests/runtime/test_recursive_view_checkpoint.py b/python/tests/runtime/test_recursive_view_checkpoint.py index ff60e1d6d4b..847b5a3f1d5 100644 --- 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/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/MerkleOuter.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/MerkleOuter.java index 400f2bbd278..2a6d55dc652 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/MerkleOuter.java +++ b/sql-to-dbsp-compiler/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-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java index f54dc1a929e..464dd8c68cd 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java +++ b/sql-to-dbsp-compiler/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/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/multi/NestedOperatorWriter.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/multi/NestedOperatorWriter.java index 99d1721133f..c7358aaf975 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/multi/NestedOperatorWriter.java +++ b/sql-to-dbsp-compiler/SQL-compiler/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/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitPostfix.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitPostfix.java index c70ae12be06..b5d601fd408 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitPostfix.java +++ b/sql-to-dbsp-compiler/SQL-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/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/ToJsonVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/ToJsonVisitor.java index 3a3b0de2c17..ee58808d9ce 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/ToJsonVisitor.java +++ b/sql-to-dbsp-compiler/SQL-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 index 707f34ea95f..d7bd47800a1 100644 --- 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); + } }