refactor(zql): move splitting of edits up to source

packages/zql/src/builder/builder.test.ts

@@ -677,6 +677,50 @@ test('self-join edit', () => {
         },
         "type": "remove",
       },
+      {
+        "node": {
+          "relationships": {
+            "recruiter": [
+              {
+                "relationships": {},
+                "row": {
+                  "id": 1,
+                  "name": "aaron",
+                  "recruiterID": null,
+                },
+              },
+            ],
+          },
+          "row": {
+            "id": 4,
+            "name": "matt",
+            "recruiterID": 1,
+          },
+        },
+        "type": "add",
+      },
+      {
+        "node": {
+          "relationships": {
+            "recruiter": [
+              {
+                "relationships": {},
+                "row": {
+                  "id": 1,
+                  "name": "aaron",
+                  "recruiterID": null,
+                },
+              },
+            ],
+          },
+          "row": {
+            "id": 4,
+            "name": "matt",
+            "recruiterID": 1,
+          },
+        },
+        "type": "remove",
+      },
       {
         "node": {
           "relationships": {

packages/zql/src/builder/builder.ts

@@ -170,16 +170,35 @@ function buildPipelineInternal(
   if (!source) {
     throw new Error(`Source not found: ${ast.table}`);
   }
-  const conn = source.connect(must(ast.orderBy), ast.where);
+  ast = uniquifyCorrelatedSubqueryConditionAliases(ast);
+
+  const csqsFromCondition = gatherCorrelatedSubqueryQueriesFromCondition(
+    ast.where,
+  );
+  const splitEditKeys: Set<string> = partitionKey
+    ? new Set(partitionKey)
+    : new Set();
+  for (const csq of csqsFromCondition) {
+    for (const key of csq.correlation.parentField) {
+      splitEditKeys.add(key);
+    }
+  }
+  if (ast.related) {
+    for (const csq of ast.related) {
+      for (const key of csq.correlation.parentField) {
+        splitEditKeys.add(key);
+      }
+    }
+  }
+  const conn = source.connect(must(ast.orderBy), ast.where, splitEditKeys);
   let end: Input = delegate.decorateInput(conn, `${name}:source(${ast.table})`);
   const {fullyAppliedFilters} = conn;
-  ast = uniquifyCorrelatedSubqueryConditionAliases(ast);
 
   if (ast.start) {
     end = delegate.decorateInput(new Skip(end, ast.start), `${name}:skip)`);
   }
 
-  for (const csq of gatherCorrelatedSubqueryQueriesFromCondition(ast.where)) {
+  for (const csq of csqsFromCondition) {
     end = applyCorrelatedSubQuery(csq, delegate, end, name);
   }
 

packages/zql/src/ivm/exists.ts

@@ -76,7 +76,10 @@ export class Exists implements Operator {
     this.#relationshipName = relationshipName;
     this.#input.setOutput(this);
     this.#storage = storage as ExistsStorage;
-    assert(this.#input.getSchema().relationships[relationshipName]);
+    assert(
+      this.#input.getSchema().relationships[relationshipName],
+      `Input schema missing ${relationshipName}`,
+    );
     this.#not = type === 'NOT EXISTS';
     this.#parentJoinKey = parentJoinKey;