diff --git a/.changeset/calm-lookups-correlate.md b/.changeset/calm-lookups-correlate.md new file mode 100644 index 000000000..55cc536a3 --- /dev/null +++ b/.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. diff --git a/packages/sync-rules/src/sync_plan/evaluator/parameter_evaluator.ts b/packages/sync-rules/src/sync_plan/evaluator/parameter_evaluator.ts index 3408959b9..0a43dd7d8 100644 --- a/packages/sync-rules/src/sync_plan/evaluator/parameter_evaluator.ts +++ b/packages/sync-rules/src/sync_plan/evaluator/parameter_evaluator.ts @@ -495,7 +495,8 @@ class FullInstantiator extends PartialInstantiator { 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,18 +517,19 @@ class FullInstantiator extends PartialInstantiator { 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(); - for (const [_, { lookup, provenance }] of pendingLookups.entries) { - lookupsToProvenance.set(lookup, provenance); + const lookupsToProvenance = new Map(); + for (const [_, pending] of pendingLookups.entries) { + lookupsToProvenance.set(pending.lookup, pending); } const outputs = await this.input.source.getParameterSets( @@ -535,16 +537,20 @@ class FullInstantiator extends PartialInstantiator { `Stream ${this.evaluators.stream.name} evaluating parameter on ${resolvedLookup.sourceTable.tablePattern}` ); - // Stream parameters generate an output row like {0: , 1: , ...}. 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: , 1: , ...}. 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, diff --git a/packages/sync-rules/test/src/sync_plan/evaluator/evaluator.test.ts b/packages/sync-rules/test/src/sync_plan/evaluator/evaluator.test.ts index fd3db1c94..e1ee6837c 100644 --- a/packages/sync-rules/test/src/sync_plan/evaluator/evaluator.test.ts +++ b/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 { + 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: