fix(zql): execute sibling exists from selective roots

packages/zql/src/ivm/strip-relationships.ts

@@ -0,0 +1,110 @@
+import type {Writable} from '../../../shared/src/writable.ts';
+import {ChangeIndex} from './change-index.ts';
+import {ChangeType} from './change-type.ts';
+import {
+  makeAddChange,
+  makeEditChange,
+  makeRemoveChange,
+  type Change,
+} from './change.ts';
+import type {Node} from './data.ts';
+import {
+  throwOutput,
+  type FetchRequest,
+  type Input,
+  type InputBase,
+  type Operator,
+  type Output,
+} from './operator.ts';
+import type {SourceSchema} from './schema.ts';
+import {type Stream} from './stream.ts';
+
+/**
+ * Removes relationship payloads from a row stream.
+ *
+ * Some physical plans use hidden relationships as proof that a filter matched.
+ * For example, a flipped EXISTS branch first joins from the child table back to
+ * the parent, which naturally attaches the matching child rows to the parent
+ * node. If that branch later feeds a root-level union, those child rows are
+ * only evidence for the branch. They are not part of the query result.
+ *
+ * This operator draws that boundary explicitly:
+ *
+ *   branch proves EXISTS with child rows
+ *              |
+ *              v
+ *   StripRelationships
+ *              |
+ *              v
+ *   plain parent rows enter the union
+ */
+export class StripRelationships implements Operator {
+  readonly #input: Input;
+  readonly #schema: SourceSchema;
+  #output: Output = throwOutput;
+
+  constructor(input: Input) {
+    this.#input = input;
+    this.#schema = {
+      ...input.getSchema(),
+      relationships: {},
+    } satisfies Writable<SourceSchema>;
+    input.setOutput(this);
+  }
+
+  setOutput(output: Output): void {
+    this.#output = output;
+  }
+
+  getSchema(): SourceSchema {
+    return this.#schema;
+  }
+
+  *fetch(req: FetchRequest): Stream<Node | 'yield'> {
+    for (const node of this.#input.fetch(req)) {
+      yield node === 'yield' ? node : stripRelationships(node);
+    }
+  }
+
+  *push(change: Change, _pusher: InputBase): Stream<'yield'> {
+    switch (change[ChangeIndex.TYPE]) {
+      case ChangeType.ADD:
+        yield* this.#output.push(
+          makeAddChange(stripRelationships(change[ChangeIndex.NODE])),
+          this,
+        );
+        return;
+
+      case ChangeType.REMOVE:
+        yield* this.#output.push(
+          makeRemoveChange(stripRelationships(change[ChangeIndex.NODE])),
+          this,
+        );
+        return;
+
+      case ChangeType.EDIT: {
+        const oldNode = stripRelationships(change[ChangeIndex.OLD_NODE]);
+        const newNode = stripRelationships(change[ChangeIndex.NODE]);
+        if (JSON.stringify(oldNode.row) === JSON.stringify(newNode.row)) {
+          return;
+        }
+        yield* this.#output.push(makeEditChange(newNode, oldNode), this);
+        return;
+      }
+
+      case ChangeType.CHILD:
+        return;
+    }
+  }
+
+  destroy(): void {
+    this.#input.destroy();
+  }
+}
+
+function stripRelationships(node: Node): Node {
+  return {
+    row: node.row,
+    relationships: {},
+  };
+}

packages/zql/src/ivm/union-fan-in.ts

@@ -15,6 +15,7 @@ import {
 } from './operator.ts';
 import {
   makeAddEmptyRelationships,
+  mergeEmpty,
   mergeRelationships,
   pushAccumulatedChanges,
 } from './push-accumulated.ts';
@@ -102,8 +103,19 @@ export class UnionFanIn implements Operator {
 
   fetch(req: FetchRequest): Stream<Node | 'yield'> {
     const iterables = this.#inputs.map(input => input.fetch(req));
-    return mergeFetches(iterables, (l, r) =>
-      this.#schema.compareRows(l.row, r.row),
+    const compareRows = req.reverse
+      ? (left: Node, right: Node) =>
+          this.#schema.compareRows(right.row, left.row)
+      : (left: Node, right: Node) =>
+          this.#schema.compareRows(left.row, right.row);
+    const relationshipNames = Object.keys(this.#schema.relationships);
+    return addEmptyRelationshipsToFetch(
+      mergeFetches(
+        iterables,
+        compareRows,
+        mergeNodeRelationships(relationshipNames),
+      ),
+      relationshipNames,
     );
   }
 
@@ -219,24 +231,58 @@ export class UnionFanIn implements Operator {
   }
 }
 
+function mergeNodeRelationships(relationshipNames: string[]) {
+  return (left: Node, right: Node): Node => {
+    const relationships = {
+      ...right.relationships,
+      ...left.relationships,
+    };
+    mergeEmpty(relationships, relationshipNames);
+    return {
+      row: left.row,
+      relationships,
+    };
+  };
+}
+
+function* addEmptyRelationshipsToFetch(
+  fetch: Iterable<Node | 'yield'>,
+  relationshipNames: string[],
+): Stream<Node | 'yield'> {
+  for (const node of fetch) {
+    if (node === 'yield') {
+      yield node;
+      continue;
+    }
+    const relationships = {...node.relationships};
+    mergeEmpty(relationships, relationshipNames);
+    yield {
+      row: node.row,
+      relationships,
+    };
+  }
+}
+
 export function* mergeFetches(
   fetches: Iterable<Node | 'yield'>[],
   comparator: (l: Node, r: Node) => number,
+  mergeEqual?: ((left: Node, right: Node) => Node) | undefined,
 ): IterableIterator<Node | 'yield'> {
   const iterators = fetches.map(i => i[Symbol.iterator]());
   let threw = false;
   try {
     const current: (Node | null)[] = [];
-    let lastNodeYielded: Node | undefined;
-    for (let i = 0; i < iterators.length; i++) {
+    const advance = function* (i: number) {
       const iter = iterators[i];
       let result = iter.next();
-      // yield yields when initializing
       while (!result.done && result.value === 'yield') {
         yield result.value;
         result = iter.next();
       }
       current[i] = result.done ? null : (result.value as Node);
+    };
+    for (let i = 0; i < iterators.length; i++) {
+      yield* advance(i);
     }
     while (current.some(c => c !== null)) {
       const min = current.reduce(
@@ -253,22 +299,35 @@ export function* mergeFetches(
       );
 
       assert(min !== undefined, 'min is undefined');
-      const [minNode, minIndex] = min;
-      const iter = iterators[minIndex];
-      let result = iter.next();
-      while (!result.done && result.value === 'yield') {
-        yield result.value;
-        result = iter.next();
-      }
-      current[minIndex] = result.done ? null : (result.value as Node);
-      if (
-        lastNodeYielded !== undefined &&
-        comparator(lastNodeYielded, minNode) === 0
-      ) {
-        continue;
+      let [nodeToYield, currentIndex] = min;
+      yield* advance(currentIndex);
+
+      while (true) {
+        const equal = current.reduce(
+          (
+            acc: [Node, number] | undefined,
+            c,
+            i,
+          ): [Node, number] | undefined => {
+            if (c === null || comparator(c, nodeToYield) !== 0) {
+              return acc;
+            }
+            return acc ?? [c, i];
+          },
+          undefined,
+        );
+        if (!equal) {
+          break;
+        }
+        const [equalNode, equalIndex] = equal;
+        nodeToYield = mergeEqual
+          ? mergeEqual(nodeToYield, equalNode)
+          : nodeToYield;
+        currentIndex = equalIndex;
+        yield* advance(currentIndex);
       }
-      lastNodeYielded = minNode;
-      yield minNode;
+
+      yield nodeToYield;
     }
   } catch (e) {
     threw = true;

packages/zql/src/ivm/yield.fetch.test.ts

@@ -51,6 +51,31 @@ class YieldSource implements Input {
   destroy(): void {}
 }
 
+class StaticInput implements Input {
+  readonly #schema: SourceSchema;
+  readonly #nodes: readonly Node[];
+
+  constructor(schema: SourceSchema, nodes: readonly Node[]) {
+    this.#schema = schema;
+    this.#nodes = nodes;
+  }
+
+  setOutput(_: Output): void {}
+
+  getSchema(): SourceSchema {
+    return this.#schema;
+  }
+
+  *fetch(req: FetchRequest): Stream<Node | 'yield'> {
+    const nodes = req.reverse ? this.#nodes.toReversed() : this.#nodes;
+    for (const node of nodes) {
+      yield node;
+    }
+  }
+
+  destroy(): void {}
+}
+
 describe('Yield Propagation', () => {
   test('FilterStart/End propagates yield', () => {
     const source = new YieldSource();
@@ -160,13 +185,13 @@ describe('Yield Propagation', () => {
         "yield",
         "yield",
         "yield",
+        "yield",
         {
           "relationships": {},
           "row": {
             "id": "1",
           },
         },
-        "yield",
         {
           "relationships": {},
           "row": {
@@ -400,3 +425,57 @@ describe('Yield Propagation', () => {
     expect(() => catchOp.fetch({})).toThrow(error);
   });
 });
+
+test('UnionFanIn fetch merges relationships from duplicate branch rows', () => {
+  const childSchema: SourceSchema = {
+    ...YIELD_SOURCE_SCHEMA_BASE,
+    tableName: 'child',
+  };
+  const leftRelationship = () =>
+    [{row: {id: 'left-child'}, relationships: {}}] as const;
+  const rightRelationship = () =>
+    [{row: {id: 'right-child'}, relationships: {}}] as const;
+  const parent = new StaticInput(YIELD_SOURCE_SCHEMA_BASE, []);
+  const fanOut = new UnionFanOut(parent);
+  const left = new StaticInput(
+    {
+      ...YIELD_SOURCE_SCHEMA_BASE,
+      relationships: {leftRelationship: childSchema},
+    },
+    [
+      {
+        row: {id: '1'},
+        relationships: {leftRelationship},
+      },
+    ],
+  );
+  const right = new StaticInput(
+    {
+      ...YIELD_SOURCE_SCHEMA_BASE,
+      relationships: {rightRelationship: childSchema},
+    },
+    [
+      {
+        row: {id: '1'},
+        relationships: {rightRelationship},
+      },
+    ],
+  );
+  const fanIn = new UnionFanIn(fanOut, [left, right]);
+  const [node] = [...fanIn.fetch({})];
+
+  expect(node).not.toBe('yield');
+  if (node === undefined || node === 'yield') {
+    throw new Error('Expected one fetched node');
+  }
+  expect(Object.keys(node.relationships).sort()).toEqual([
+    'leftRelationship',
+    'rightRelationship',
+  ]);
+  expect([...node.relationships.leftRelationship()]).toEqual([
+    {row: {id: 'left-child'}, relationships: {}},
+  ]);
+  expect([...node.relationships.rightRelationship()]).toEqual([
+    {row: {id: 'right-child'}, relationships: {}},
+  ]);
+});