Refactor parameter lookups to make the names configurable.

Author: rkistner

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

@@ -3,6 +3,7 @@ import { ParameterLookup, RequestParameters } from '@powersync/service-sync-rule
 import { expect, test } from 'vitest';
 import * as test_utils from '../test-utils/test-utils-index.js';
 import { TEST_TABLE } from './util.js';
+import { ParameterLookupScope } from '@powersync/service-sync-rules/src/HydrationState.js';
 
 /**
  * @example
@@ -15,6 +16,8 @@ import { TEST_TABLE } from './util.js';
  * ```
  */
 export function registerDataStorageParameterTests(generateStorageFactory: storage.TestStorageFactory) {
+  const MYBUCKET_1: ParameterLookupScope = { lookupName: 'mybucket', queryId: '1' };
+
   test('save and load parameters', async () => {
     await using factory = await generateStorageFactory();
     const syncRules = await factory.updateSyncRules({
@@ -57,7 +60,7 @@ bucket_definitions:
     });
 
     const checkpoint = await bucketStorage.getCheckpoint();
-    const parameters = await checkpoint.getParameterSets([ParameterLookup.normalized('mybucket', '1', ['user1'])]);
+    const parameters = await checkpoint.getParameterSets([ParameterLookup.normalized(MYBUCKET_1, ['user1'])]);
     expect(parameters).toEqual([
       {
         group_id: 'group1a'
@@ -105,15 +108,15 @@ bucket_definitions:
     });
     const checkpoint2 = await bucketStorage.getCheckpoint();
 
-    const parameters = await checkpoint2.getParameterSets([ParameterLookup.normalized('mybucket', '1', ['user1'])]);
+    const parameters = await checkpoint2.getParameterSets([ParameterLookup.normalized(MYBUCKET_1, ['user1'])]);
     expect(parameters).toEqual([
       {
         group_id: 'group2'
       }
     ]);
 
     // Use the checkpoint to get older data if relevant
-    const parameters2 = await checkpoint1.getParameterSets([ParameterLookup.normalized('mybucket', '1', ['user1'])]);
+    const parameters2 = await checkpoint1.getParameterSets([ParameterLookup.normalized(MYBUCKET_1, ['user1'])]);
     expect(parameters2).toEqual([
       {
         group_id: 'group1'
@@ -182,8 +185,8 @@ bucket_definitions:
     // association of `list1`::`todo2`
     const checkpoint = await bucketStorage.getCheckpoint();
     const parameters = await checkpoint.getParameterSets([
-      ParameterLookup.normalized('mybucket', '1', ['list1']),
-      ParameterLookup.normalized('mybucket', '1', ['list2'])
+      ParameterLookup.normalized(MYBUCKET_1, ['list1']),
+      ParameterLookup.normalized(MYBUCKET_1, ['list2'])
     ]);
 
     expect(parameters.sort((a, b) => (a.todo_id as string).localeCompare(b.todo_id as string))).toEqual([
@@ -230,17 +233,11 @@ bucket_definitions:
 
     const checkpoint = await bucketStorage.getCheckpoint();
 
-    const parameters1 = await checkpoint.getParameterSets([
-      ParameterLookup.normalized('mybucket', '1', [314n, 314, 3.14])
-    ]);
+    const parameters1 = await checkpoint.getParameterSets([ParameterLookup.normalized(MYBUCKET_1, [314n, 314, 3.14])]);
     expect(parameters1).toEqual([TEST_PARAMS]);
-    const parameters2 = await checkpoint.getParameterSets([
-      ParameterLookup.normalized('mybucket', '1', [314, 314n, 3.14])
-    ]);
+    const parameters2 = await checkpoint.getParameterSets([ParameterLookup.normalized(MYBUCKET_1, [314, 314n, 3.14])]);
     expect(parameters2).toEqual([TEST_PARAMS]);
-    const parameters3 = await checkpoint.getParameterSets([
-      ParameterLookup.normalized('mybucket', '1', [314n, 314, 3])
-    ]);
+    const parameters3 = await checkpoint.getParameterSets([ParameterLookup.normalized(MYBUCKET_1, [314n, 314, 3])]);
     expect(parameters3).toEqual([]);
   });
 
@@ -294,7 +291,7 @@ bucket_definitions:
     const checkpoint = await bucketStorage.getCheckpoint();
 
     const parameters1 = await checkpoint.getParameterSets([
-      ParameterLookup.normalized('mybucket', '1', [1152921504606846976n])
+      ParameterLookup.normalized(MYBUCKET_1, [1152921504606846976n])
     ]);
     expect(parameters1).toEqual([TEST_PARAMS]);
   });
@@ -335,7 +332,7 @@ bucket_definitions:
     const querier = sync_rules.getBucketParameterQuerier(test_utils.querierOptions(parameters)).querier;
 
     const lookups = querier.parameterQueryLookups;
-    expect(lookups).toEqual([ParameterLookup.normalized('by_workspace', '1', ['u1'])]);
+    expect(lookups).toEqual([ParameterLookup.normalized({ lookupName: 'by_workspace', queryId: '1' }, ['u1'])]);
 
     const parameter_sets = await checkpoint.getParameterSets(lookups);
     expect(parameter_sets).toEqual([{ workspace_id: 'workspace1' }]);
@@ -408,7 +405,7 @@ bucket_definitions:
     const querier = sync_rules.getBucketParameterQuerier(test_utils.querierOptions(parameters)).querier;
 
     const lookups = querier.parameterQueryLookups;
-    expect(lookups).toEqual([ParameterLookup.normalized('by_public_workspace', '1', [])]);
+    expect(lookups).toEqual([ParameterLookup.normalized({ lookupName: 'by_public_workspace', queryId: '1' }, [])]);
 
     const parameter_sets = await checkpoint.getParameterSets(lookups);
     parameter_sets.sort((a, b) => JSON.stringify(a).localeCompare(JSON.stringify(b)));
@@ -510,8 +507,8 @@ bucket_definitions:
 
     const lookups = querier.parameterQueryLookups;
     expect(lookups).toEqual([
-      ParameterLookup.normalized('by_workspace', '1', []),
-      ParameterLookup.normalized('by_workspace', '2', ['u1'])
+      ParameterLookup.normalized({ lookupName: 'by_workspace', queryId: '1' }, []),
+      ParameterLookup.normalized({ lookupName: 'by_workspace', queryId: '2' }, ['u1'])
     ]);
 
     const parameter_sets = await checkpoint.getParameterSets(lookups);
@@ -561,7 +558,7 @@ bucket_definitions:
 
     const checkpoint = await bucketStorage.getCheckpoint();
 
-    const parameters = await checkpoint.getParameterSets([ParameterLookup.normalized('mybucket', '1', ['user1'])]);
+    const parameters = await checkpoint.getParameterSets([ParameterLookup.normalized(MYBUCKET_1, ['user1'])]);
     expect(parameters).toEqual([]);
   });
 

packages/service-core-tests/src/tests/register-parameter-compacting-tests.ts

@@ -40,7 +40,7 @@ bucket_definitions:
       await batch.commit('1/1');
     });
 
-    const lookup = ParameterLookup.normalized('test', '1', ['t1']);
+    const lookup = ParameterLookup.normalized({ lookupName: 'test', queryId: '1' }, ['t1']);
 
     const checkpoint1 = await bucketStorage.getCheckpoint();
     const parameters1 = await checkpoint1.getParameterSets([lookup]);
@@ -151,7 +151,7 @@ bucket_definitions:
         await batch.commit('3/1');
       });
 
-      const lookup = ParameterLookup.normalized('test', '1', ['u1']);
+      const lookup = ParameterLookup.normalized({ lookupName: 'test', queryId: '1' }, ['u1']);
 
       const checkpoint1 = await bucketStorage.getCheckpoint();
       const parameters1 = await checkpoint1.getParameterSets([lookup]);

packages/service-core/test/src/sync/BucketChecksumState.test.ts

@@ -504,7 +504,7 @@ bucket_definitions:
 
     const line = (await state.buildNextCheckpointLine({
       base: storage.makeCheckpoint(1n, (lookups) => {
-        expect(lookups).toEqual([ParameterLookup.normalized('by_project', '1', ['u1'])]);
+        expect(lookups).toEqual([ParameterLookup.normalized({ lookupName: 'by_project', queryId: '1' }, ['u1'])]);
         return [{ id: 1 }, { id: 2 }];
       }),
       writeCheckpoint: null,
@@ -565,7 +565,7 @@ bucket_definitions:
     // Now we get a new line
     const line2 = (await state.buildNextCheckpointLine({
       base: storage.makeCheckpoint(2n, (lookups) => {
-        expect(lookups).toEqual([ParameterLookup.normalized('by_project', '1', ['u1'])]);
+        expect(lookups).toEqual([ParameterLookup.normalized({ lookupName: 'by_project', queryId: '1' }, ['u1'])]);
         return [{ id: 1 }, { id: 2 }, { id: 3 }];
       }),
       writeCheckpoint: null,

packages/sync-rules/src/BucketParameterQuerier.ts

@@ -1,4 +1,5 @@
 import { ResolvedBucket } from './BucketDescription.js';
+import { ParameterLookupScope } from './HydrationState.js';
 import { RequestedStream } from './SqlSyncRules.js';
 import { RequestParameters, SqliteJsonRow, SqliteJsonValue } from './types.js';
 import { normalizeParameterValue } from './utils.js';
@@ -92,15 +93,15 @@ export class ParameterLookup {
   // bucket definition name, parameter query index, ...lookup values
   readonly values: SqliteJsonValue[];
 
-  static normalized(bucketDefinitionName: string, queryIndex: string, values: SqliteJsonValue[]): ParameterLookup {
-    return new ParameterLookup([bucketDefinitionName, queryIndex, ...values.map(normalizeParameterValue)]);
+  static normalized(scope: ParameterLookupScope, values: SqliteJsonValue[]): ParameterLookup {
+    return new ParameterLookup([scope.lookupName, scope.queryId, ...values.map(normalizeParameterValue)]);
   }
 
   /**
    *
    * @param values must be pre-normalized (any integer converted into bigint)
    */
-  constructor(values: SqliteJsonValue[]) {
+  private constructor(values: SqliteJsonValue[]) {
     this.values = values;
   }
 }

packages/sync-rules/src/HydrationState.ts

@@ -6,7 +6,7 @@ export interface BucketSourceState {
   bucketPrefix: string;
 }
 
-export interface BucketParameterLookupSourceState {
+export interface ParameterLookupScope {
   /** The lookup name + queryid is used to reference the parameter lookup record. */
   lookupName: string;
   queryId: string;
@@ -20,7 +20,7 @@ export interface BucketParameterLookupSourceState {
  */
 export interface HydrationState<
   T extends BucketSourceState = BucketSourceState,
-  U extends BucketParameterLookupSourceState = BucketParameterLookupSourceState
+  U extends ParameterLookupScope = ParameterLookupScope
 > {
   /**
    * Given a bucket data source definition, get the bucket prefix to use for it.
@@ -30,7 +30,7 @@ export interface HydrationState<
   /**
    * Given a bucket parameter lookup definition, get the persistence name to use.
    */
-  getParameterLookupState(source: BucketParameterLookupSourceDefinition): U;
+  getParameterLookupScope(source: BucketParameterLookupSourceDefinition): U;
 }
 
 /**
@@ -44,7 +44,7 @@ export const DEFAULT_HYDRATION_STATE: HydrationState = {
       bucketPrefix: source.defaultBucketPrefix
     };
   },
-  getParameterLookupState(source) {
+  getParameterLookupScope(source) {
     return {
       lookupName: source.defaultLookupName,
       queryId: source.defaultQueryId
@@ -67,7 +67,7 @@ export class VersionedHydrationState implements HydrationState {
     };
   }
 
-  getParameterLookupState(source: BucketParameterLookupSourceDefinition): BucketParameterLookupSourceState {
+  getParameterLookupScope(source: BucketParameterLookupSourceDefinition): ParameterLookupScope {
     // No transformations applied here
     return {
       lookupName: source.defaultLookupName,
@@ -85,7 +85,7 @@ export class BucketIdTransformerHydrationState implements HydrationState {
     };
   }
 
-  getParameterLookupState(source: BucketParameterLookupSourceDefinition): BucketParameterLookupSourceState {
+  getParameterLookupScope(source: BucketParameterLookupSourceDefinition): ParameterLookupScope {
     // No transformations applied here
     return {
       lookupName: source.defaultLookupName,

packages/sync-rules/src/SqlParameterQuery.ts

@@ -44,7 +44,7 @@ import {
 } from './types.js';
 import { filterJsonRow, getBucketId, isJsonValue, isSelectStatement, normalizeParameterValue } from './utils.js';
 import { DetectRequestParameters } from './validators.js';
-import { HydrationState, resolveHydrationState } from './HydrationState.js';
+import { ParameterLookupScope, HydrationState, resolveHydrationState } from './HydrationState.js';
 
 export interface SqlParameterQueryOptions {
   sourceTable: TablePattern;
@@ -350,20 +350,24 @@ export class SqlParameterQuery
   createParameterQuerierSource(params: CreateSourceParams): BucketParameterQuerierSource {
     const hydrationState = resolveHydrationState(params);
     const bucketPrefix = hydrationState.getBucketSourceState(this.querierDataSource).bucketPrefix;
+    const lookupState = hydrationState.getParameterLookupScope(this);
+
     return {
       pushBucketParameterQueriers: (result: PendingQueriers, options: GetQuerierOptions) => {
-        const q = this.getBucketParameterQuerier(options.globalParameters, ['default'], bucketPrefix);
+        const q = this.getBucketParameterQuerier(options.globalParameters, ['default'], bucketPrefix, lookupState);
         result.queriers.push(q);
       }
     };
   }
 
   createParameterLookupSource(params: CreateSourceParams): BucketParameterLookupSource {
     // FIXME: Use HydrationState for lookups.
+    const hydrationState = resolveHydrationState(params);
+    const lookupState = hydrationState.getParameterLookupScope(this);
     return {
       evaluateParameterRow: (sourceTable: SourceTableInterface, row: SqliteRow): EvaluatedParametersResult[] => {
         if (this.tableSyncsParameters(sourceTable)) {
-          return this.evaluateParameterRow(row);
+          return this.evaluateParameterRow(lookupState, row);
         } else {
           return [];
         }
@@ -374,16 +378,16 @@ export class SqlParameterQuery
   /**
    * Given a replicated row, results an array of bucket parameter rows to persist.
    */
-  evaluateParameterRow(row: SqliteRow): EvaluatedParametersResult[] {
+  evaluateParameterRow(scope: ParameterLookupScope, row: SqliteRow): EvaluatedParametersResult[] {
     const tables = {
       [this.table.nameInSchema]: row
     };
     try {
       const filterParameters = this.filter.filterRow(tables);
       let result: EvaluatedParametersResult[] = [];
       for (let filterParamSet of filterParameters) {
-        let lookup: SqliteJsonValue[] = [this.descriptorName, this.queryId];
-        lookup.push(
+        let lookupValues: SqliteJsonValue[] = [];
+        lookupValues.push(
           ...this.inputParameters.map((param) => {
             return normalizeParameterValue(param.filteredRowToLookupValue(filterParamSet));
           })
@@ -393,7 +397,7 @@ export class SqlParameterQuery
 
         const role: EvaluatedParameters = {
           bucketParameters: data.map((row) => filterJsonRow(row)),
-          lookup: new ParameterLookup(lookup)
+          lookup: ParameterLookup.normalized(scope, lookupValues)
         };
         result.push(role);
       }
@@ -458,12 +462,12 @@ export class SqlParameterQuery
    *
    * Each lookup is [bucket definition name, parameter query index, ...lookup values]
    */
-  getLookups(parameters: RequestParameters): ParameterLookup[] {
+  getLookups(scope: ParameterLookupScope, parameters: RequestParameters): ParameterLookup[] {
     if (!this.expandedInputParameter) {
-      let lookup: SqliteJsonValue[] = [this.descriptorName, this.queryId];
+      let lookupValues: SqliteJsonValue[] = [];
 
       let valid = true;
-      lookup.push(
+      lookupValues.push(
         ...this.inputParameters.map((param): SqliteJsonValue => {
           // Scalar value
           const value = param.parametersToLookupValue(parameters);
@@ -479,7 +483,7 @@ export class SqlParameterQuery
       if (!valid) {
         return [];
       }
-      return [new ParameterLookup(lookup)];
+      return [ParameterLookup.normalized(scope, lookupValues)];
     } else {
       const arrayString = this.expandedInputParameter.parametersToLookupValue(parameters);
 
@@ -498,10 +502,10 @@ export class SqlParameterQuery
 
       return values
         .map((expandedValue) => {
-          let lookup: SqliteJsonValue[] = [this.descriptorName, this.queryId];
+          let lookupValues: SqliteJsonValue[] = [];
           let valid = true;
           const normalizedExpandedValue = normalizeParameterValue(expandedValue);
-          lookup.push(
+          lookupValues.push(
             ...this.inputParameters.map((param): SqliteJsonValue => {
               if (param == this.expandedInputParameter) {
                 // Expand array value
@@ -523,7 +527,7 @@ export class SqlParameterQuery
             return null;
           }
 
-          return new ParameterLookup(lookup);
+          return ParameterLookup.normalized(scope, lookupValues);
         })
         .filter((lookup) => lookup != null) as ParameterLookup[];
     }
@@ -532,9 +536,10 @@ export class SqlParameterQuery
   getBucketParameterQuerier(
     requestParameters: RequestParameters,
     reasons: BucketInclusionReason[],
-    bucketPrefix: string
+    bucketPrefix: string,
+    scope: ParameterLookupScope
   ): BucketParameterQuerier {
-    const lookups = this.getLookups(requestParameters);
+    const lookups = this.getLookups(scope, requestParameters);
     if (lookups.length == 0) {
       // This typically happens when the query is pre-filtered using a where clause
       // on the parameters, and does not depend on the database state.