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": [],