From b8fb7265f6c60781fa5b960db1011e7b1acffa64 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Wed, 12 Aug 2026 23:03:05 -0700 Subject: [PATCH] [sql] Give the streams that leave a recursive circuit ids of their own MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The compiler assigned the same ids to streams inside the recursive fragment and the same streams exported outside the fragment: ```rust let (s6, s9) = circuit.recursive(|circuit, (unused_0, s3)| { ... let s6 = ...join...; s6.set_persistent_id(Some("ed5fc63e…")); // inside the scope Ok((s6, s9)) }).unwrap(); s6.set_persistent_id(Some("ed5fc63e…")); // exported — same id ``` However these are different streams with different integrals as can be seen from the circuit: ``` ┌───────────────────────────────────────────────────────────────┐ │ │ i │ ┌───┐ |-s6 inside the scope │ ────┼──►δ0─────────►│ │ V ┌────────┐ ┌───────────────┐ │ ┌───────────┐ s6 exported │ │ f ├─────►│distinct├──┬───►│integrate_trace├───┼──►│consolidate├───────► │ ┌──────►│ │ └────────┘ │ └───────────────┘ │ └───────────┘ │ │ └───┘ │ │ │ │ │ │ │ │ │ │ │ │ ┌────┐ │ │ │ └───────┤z^-1│◄────────────────┘ │ │ └────┘ │ │ │ └───────────────────────────────────────────────────────────────┘ ``` Suffix the exported stream's id with '.export', in both the single-file and the multi-crate writer, mirroring the '.delay' suffix that already distinguishes a recursive input. The checkpoint then holds .shard.accintegral for the inner trace and .export.shard.accintegral for the outer one. Bump RECURSIVE_STATE_VERSION with it: the layout of what a recursive view stores changes, so the ids in the dataflow graph have to change too, or the pipeline manager would diff the views as unchanged while the runtime finds their state missing, and refuse to start with UnexpectedBootstrap. The metadata test's golden ids move for the same reason. Signed-off-by: Leonid Ryzhyk (cherry picked from commit cce7456a1836a2be7e5f722084007f8be2426fe6) --- .../compiler/backend/MerkleOuter.java | 2 +- .../compiler/backend/rust/ToRustVisitor.java | 24 ++++++++++++++++++- .../rust/multi/NestedOperatorWriter.java | 24 ++++++++++++++++++- .../metadataTests-generateDFRecursive.json | 4 ++-- 4 files changed, 49 insertions(+), 5 deletions(-) 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 2a6d55dc652..a9aeaf2a037 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 @@ -42,7 +42,7 @@ public class MerkleOuter extends CircuitVisitor { * same ids, so it reports the recursive views as modified and bootstraps them. * *

Bump this string whenever the runtime changes the way it stores that state. */ - public static final String RECURSIVE_STATE_VERSION = "recursive-state-v1"; + public static final String RECURSIVE_STATE_VERSION = "recursive-state-v2"; public final Map operatorHash; public final boolean includeInputs; 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 464dd8c68cd..45a691b8a76 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 @@ -521,7 +521,7 @@ public VisitDecision preorder(DBSPNestedOperator operator) { for (int i = 0; i < operator.outputCount(); i++) { OutputPort port = operator.internalOutputs.get(i); if (port != null) { - this.computeHash(port.operator); + this.computeExportedHash(port.operator); this.tagStream(port.operator.to(DBSPSimpleOperator.class)); } } @@ -621,6 +621,28 @@ void tagStream(DBSPOperator operator) { .append(".set_persistent_id(hash);"); } + /** Emit the id of a stream that leaves a recursive circuit. + * + *

It must not be the operator's own id: that id already names the stream + * inside the scope, and the two are separate streams whose operators keep + * separate state. Sharing it makes an inner trace and an outer one write + * the same file, and whichever restores second reads a batch that was + * written with the other's layout. */ + void computeExportedHash(DBSPOperator operator) { + if (this.preferHash) + return; + this.builder.append("let hash = "); + HashString hash = OperatorHash.getHash(operator, true); + if (hash == null) { + this.builder.append("None;").newline(); + } else { + this.builder.append("Some(concat!(") + .append(hash.toQuotedString()) + .append(", \".export\"));") + .newline(); + } + } + void computeHash(DBSPOperator operator) { if (this.preferHash) // Hash is received as an argument 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 c7358aaf975..cf99d30f837 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 @@ -115,6 +115,28 @@ String inputName(int inputNo) { /** Emit the code that gives the stream named {@code name} the persistent id of * {@code operator}. */ + /** Name the stream that leaves the recursive circuit. + * + *

It must not be the operator's own id: that id already names the stream + * inside the scope, and the two are separate streams whose operators keep + * separate state. Sharing it makes an inner trace and an outer one write + * the same file, and whichever restores second reads a batch that was + * written with the other's layout. */ + private void setExportedPersistentId(DBSPOperator operator, String name) { + HashString hash = OperatorHash.getHash(operator, true); + if (hash == null) { + this.builder().append("let hash = None;").newline(); + } else { + this.builder().append("let hash = Some(concat!(") + .append(hash.toQuotedString()) + .append(", \".export\"));") + .newline(); + } + this.builder().append(name) + .append(".set_persistent_id(hash);") + .newline(); + } + private void setPersistentId(DBSPOperator operator, String name) { this.builder().append("let hash = "); HashString hash = OperatorHash.getHash(operator, true); @@ -267,7 +289,7 @@ public void write(DBSPCompiler compiler) { for (int i = 0; i < operator.outputCount(); i++) { OutputPort port = operator.internalOutputs.get(i); if (port != null) - this.setPersistentId(port.operator, port.getName(false)); + this.setExportedPersistentId(port.operator, port.getName(false)); } this.builder().append("if let Some(region) = region { circuit.close_region(region.clone()) };").newline(); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDFRecursive.json b/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDFRecursive.json index 24df341a121..f57629a6455 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDFRecursive.json +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDFRecursive.json @@ -442,7 +442,7 @@ "final": 10 }, "positions": [], - "persistent_id": "58795ce08f08d78c6cf97082e9dfcc5fce43825c44bc763377874648bf074d08" + "persistent_id": "71231f996af603d4d548e6aa4a7b1fe5ee9bfccf3fadadbb5a554d2c37824de5" } }, "s8": { "operation": "inspect", @@ -456,7 +456,7 @@ "positions": [ {"start_line_number":4,"start_column":1,"end_line_number":21,"end_column":1} ], - "persistent_id": "08b99a30ddf054bdadc7c1e3588e18db716ca20bdccad6761e24181bc61ccc63" + "persistent_id": "52e7d68b35ea322c8f02f0fe4fe0e06c8341278c3314cf8f08d20cbaac09b156" }, "s9": { "operation": "constant", "inputs": [],