Fix provenance for lookups with multiple inputs

.changeset/calm-lookups-correlate.md

@@ -0,0 +1,5 @@
+---
+'@powersync/service-sync-rules': patch
+---
+
+Preserve multi-column parameter correlation when a deduplicated lookup is reached through multiple provenance paths.

packages/sync-rules/src/sync_plan/evaluator/parameter_evaluator.ts

@@ -495,7 +495,8 @@ class FullInstantiator extends PartialInstantiator<InstantiationInput> {
 
       interface PendingLookup {
         lookup: ScopedParameterLookup;
-        provenance: { origin: VirtualSourceRow[]; symbol: symbol }[];
+        provenancePaths: VirtualSourceRow[][];
+        resultSet: symbol;
       }
 
       // It's possible that we'll have the same logical lookup with multiple provenance values. For instance, if the
@@ -516,35 +517,40 @@ class FullInstantiator extends PartialInstantiator<InstantiationInput> {
           if (old == null) {
             return {
               lookup: ScopedParameterLookup.normalized(scope, UnscopedParameterLookup.normalized(directValues)),
-              provenance: [{ origin: provenance, symbol: Symbol(`lookup ${stage}.${index}`) }]
+              provenancePaths: [provenance],
+              resultSet: Symbol(`lookup ${stage}.${index}`)
             };
           } else {
-            old.provenance.push({ origin: provenance, symbol: Symbol(`lookup ${stage}.${index}`) });
+            old.provenancePaths.push(provenance);
             return old;
           }
         });
       }
 
-      const lookupsToProvenance = new Map<ScopedParameterLookup, { origin: VirtualSourceRow[]; symbol: symbol }[]>();
-      for (const [_, { lookup, provenance }] of pendingLookups.entries) {
-        lookupsToProvenance.set(lookup, provenance);
+      const lookupsToProvenance = new Map<ScopedParameterLookup, PendingLookup>();
+      for (const [_, pending] of pendingLookups.entries) {
+        lookupsToProvenance.set(pending.lookup, pending);
       }
 
       const outputs = await this.input.source.getParameterSets(
         [...lookupsToProvenance.keys()],
         `Stream ${this.evaluators.stream.name} evaluating parameter on ${resolvedLookup.sourceTable.tablePattern}`
       );
 
-      // Stream parameters generate an output row like {0: <expr>, 1: <expr>, ...}.
       const values = outputs.flatMap(({ lookup, rows }) => {
-        return lookupsToProvenance.get(lookup)!.flatMap(({ symbol, origin }) => {
+        const { provenancePaths, resultSet } = lookupsToProvenance.get(lookup)!;
+        return provenancePaths.flatMap((origin, provenanceIndex) => {
           return rows.map((row, rowid) => {
             const length = Object.entries(row).length;
             const asArray: ParameterValueWithRow[] = [];
 
             for (let i = 0; i < length; i++) {
+              // Stream parameters generate an output row like {0: <expr>, 1: <expr>, ...}.
               const value = row[i.toString()] as SqliteParameterValue;
-              const directOrigin = length > 1 ? { resultSet: symbol, row: rowid } : undefined;
+
+              // All paths share one result set because the lookup was deduplicated. Include the path index in the row
+              // identity because the same output rows are instantiated once for every path.
+              const directOrigin = length > 1 ? { resultSet, row: provenanceIndex * rows.length + rowid } : undefined;
 
               asArray.push({
                 value,

packages/sync-rules/test/src/sync_plan/evaluator/evaluator.test.ts

@@ -751,6 +751,64 @@ streams:
     ]);
   });
 
+  syncTest('preserves correlation across provenance paths for a deduplicated lookup', async ({ sync }) => {
+    const desc = sync.prepareSyncStreams(`
+config:
+  edition: 3
+
+streams:
+  stream:
+      auto_subscribe: true
+      query: |
+        SELECT a.*
+        FROM a, b, c
+        WHERE a.x = b.x
+          AND a.y = b.y
+          AND b.k = c.k
+          AND c.user_id = auth.user_id()
+`);
+
+    const { querier, errors } = desc.getBucketParameterQuerier({
+      globalParameters: requestParameters({ sub: 'user1' }),
+      hasDefaultStreams: true,
+      streams: {}
+    });
+    expect(errors).toStrictEqual([]);
+    let requestedLookups = 0;
+
+    const dynamicBuckets = await querier.queryDynamicBucketDescriptions({
+      async getParameterSets(lookups, debugDefinition): Promise<ParameterLookupRows[]> {
+        expect(lookups).toHaveLength(1);
+        requestedLookups++;
+
+        if (debugDefinition.endsWith(' on c')) {
+          // Both c rows produce the same b lookup key. PowerSync must query b
+          // once while retaining the two provenance paths that reached it.
+          return [{ lookup: lookups[0], rows: [{ '0': 'shared-k' }, { '0': 'shared-k' }] }];
+        } else if (debugDefinition.endsWith(' on b')) {
+          // Each b row is one correlated (x, y) pair. Copies of this result set
+          // for different provenance paths must not make its columns independent.
+          return [
+            {
+              lookup: lookups[0],
+              rows: [
+                { '0': 'X1', '1': 'Y1' },
+                { '0': 'X2', '1': 'Y2' }
+              ]
+            }
+          ];
+        }
+
+        throw new Error(`Unexpected lookup: ${debugDefinition}`);
+      }
+    });
+
+    // One on c (deduplicated), one on b.
+    expect(requestedLookups).toStrictEqual(2);
+    const parameters = new Set(dynamicBuckets.map(({ bucket }) => bucket.slice(bucket.indexOf('['))));
+    expect(parameters).toStrictEqual(new Set(['["Y1","X1"]', '["Y2","X2"]']));
+  });
+
   syncTest('preserves correlation across duplicate lookup output rows', async ({ sync }) => {
     const desc = sync.prepareSyncStreams(`
 config: