Skip to content

[sql] Give the streams that leave a recursive circuit ids of their own #6852

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ryzhyk merged 1 commit into from
Aug 13, 2026
Merged

[sql] Give the streams that leave a recursive circuit ids of their own #6852

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ...ompiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/MerkleOuter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ public class MerkleOuter extends CircuitVisitor {
* same ids, so it reports the recursive views as modified and bootstraps them.
*
* <p>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<Long, HashString> operatorHash;
public final boolean includeInputs;
Expand Down
24 changes: 23 additions & 1 deletion .../SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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));
}
}
Expand Down Expand Up @@ -621,6 +621,28 @@ void tagStream(DBSPOperator operator) {
.append(".set_persistent_id(hash);");
}

/** Emit the id of a stream that leaves a recursive circuit.
*
* <p>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) {

@mihaibudiu mihaibudiu Aug 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If this function is only used once maybe it can be inlined

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
Expand Down
24 changes: 23 additions & 1 deletion .../src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/multi/NestedOperatorWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*
* <p>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);
Expand Down Expand Up @@ -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();
Expand Down
4 changes: 2 additions & 2 deletions sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDFRecursive.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -442,7 +442,7 @@
"final": 10
},
"positions": [],
"persistent_id": "58795ce08f08d78c6cf97082e9dfcc5fce43825c44bc763377874648bf074d08"
"persistent_id": "71231f996af603d4d548e6aa4a7b1fe5ee9bfccf3fadadbb5a554d2c37824de5"
}
}, "s8": {
"operation": "inspect",
Expand All @@ -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": [],
Expand Down
Loading