Correlate parameters if they originate from the same logical result set (#633)

Author: simolus3

packages/service-core-tests/src/tests/register-data-storage-parameter-tests.ts

@@ -1067,4 +1067,87 @@ streams:
     expect(foundLookupA).toBeTruthy();
     expect(foundLookupB).toBeTruthy();
   });
+
+  test('sync streams preserve duplicate downstream lookups with different provenance', async () => {
+    await using factory = await generateStorageFactory();
+    const syncRules = await factory.updateSyncRules(
+      updateSyncRulesFromYaml(
+        `
+config:
+  edition: 3
+streams:
+  stream:
+    auto_subscribe: true
+    query: |
+      SELECT a.*
+      FROM a, b, c
+      WHERE a.x = b.x
+        AND a.z = c.z
+        AND b.y = c.y
+        AND c.u = auth.user_id()
+    `,
+        {
+          storageVersion
+        }
+      )
+    );
+    const bucketStorage = factory.getInstance(syncRules);
+    const sync_rules = syncRules.parsed(test_utils.PARSE_OPTIONS).hydratedSyncRules();
+
+    await using writer = await bucketStorage.createWriter(test_utils.BATCH_OPTIONS);
+    const tableB = await test_utils.resolveTestTable(writer, 'b', ['id'], config);
+    const tableC = await test_utils.resolveTestTable(writer, 'c', ['id'], config);
+
+    await writer.markAllSnapshotDone('1/1');
+    await writer.save({
+      sourceTable: tableB,
+      tag: storage.SaveOperationTag.INSERT,
+      after: {
+        id: 'b1',
+        y: 'shared-y',
+        x: 'x-from-shared-y'
+      },
+      afterReplicaId: test_utils.rid('b1')
+    });
+    await writer.save({
+      sourceTable: tableC,
+      tag: storage.SaveOperationTag.INSERT,
+      after: {
+        id: 'c1',
+        u: 'user1',
+        y: 'shared-y',
+        z: 'z1'
+      },
+      afterReplicaId: test_utils.rid('c1')
+    });
+    await writer.save({
+      sourceTable: tableC,
+      tag: storage.SaveOperationTag.INSERT,
+      after: {
+        id: 'c2',
+        u: 'user1',
+        y: 'shared-y',
+        z: 'z2'
+      },
+      afterReplicaId: test_utils.rid('c2')
+    });
+    await writer.commit('1/1');
+
+    const checkpoint = await bucketStorage.getCheckpoint();
+    const parameters = new RequestParameters(new JwtPayload({ sub: 'user1' }), {});
+    const querier = sync_rules.getBucketParameterQuerier({
+      ...test_utils.querierOptions(parameters)
+    }).querier;
+
+    const buckets = await querier.queryDynamicBucketDescriptions({
+      async getParameterSets(lookups) {
+        return checkpoint.getParameterSets(lookups, 1000);
+      }
+    });
+
+    expect(buckets.map((bucket) => bucket.bucket).sort()).toStrictEqual([
+      expect.stringMatching(/stream.*\["x-from-shared-y","z1"\]$/),
+      expect.stringMatching(/stream.*\["x-from-shared-y","z2"\]$/)
+    ]);
+  });
 }

packages/sync-rules/src/compiler/equality.ts

@@ -1,3 +1,5 @@
+import { SqliteParameterValue } from '../types.js';
+
 /**
  * Stable value-based hashcodes for JavaScript. The sync streams compiler uses hashmaps derived from this to efficiently
  * de-duplicate equivalent expressions and lookups.
@@ -71,6 +73,43 @@ export class StableHasher {
   };
 
   static readonly defaultListEquality = listEquality(this.defaultEquality);
+
+  static readonly parameterValueEquality: Equality<SqliteParameterValue> = (() => {
+    const buf = new DataView(new ArrayBuffer(8));
+
+    return {
+      equals: function (a: SqliteParameterValue, b: SqliteParameterValue): boolean {
+        // Allowed values are numbers, string, and bigint. All of them compare correctly with ===
+        return a === b;
+      },
+      hash: function (hasher: StableHasher, value: SqliteParameterValue): void {
+        switch (typeof value) {
+          case 'string':
+            hasher.addString(value);
+            break;
+          case 'number':
+            const normalized = value || 0; // Ensure 0 and -0 have the same hash code.
+            buf.setFloat64(0, value, true);
+            hasher.addHash(buf.getUint32(0, true));
+            hasher.addHash(buf.getUint32(4, true));
+            break;
+          case 'bigint':
+            // Most bigints we're dealing with fit in 64 bits. Truncating is fine, we're building hashes anyway.
+            buf.setBigInt64(0, value, true);
+            hasher.addHash(buf.getUint32(0, true));
+            hasher.addHash(buf.getUint32(4, true));
+          case 'boolean':
+            hasher.addHash(value ? 0 : 1);
+            break;
+          case 'symbol':
+          case 'undefined':
+          case 'object':
+          case 'function':
+            throw new Error(`Not a parameter value: ${value}`);
+        }
+      }
+    };
+  })();
 }
 
 /**