Support IN operator on parameter data

Author: simolus3

packages/sync-rules/src/streams/filter.ts

@@ -271,12 +271,10 @@ export class Subquery {
         const lookups: ParameterLookup[] = [];
 
         for (const [variant, id] of innerVariants) {
-          const instantiation = variant.findStaticInstantiation(parameters);
-          if (instantiation == null) {
-            continue;
+          const instantiations = variant.findStaticInstantiations(parameters);
+          for (const instantiation of instantiations) {
+            lookups.push(ParameterLookup.normalized(context.streamName, id, instantiation));
           }
-
-          lookups.push(ParameterLookup.normalized(context.streamName, id, instantiation));
         }
 
         return lookups;
@@ -438,7 +436,20 @@ export class CompareRowValueWithStreamParameter extends FilterOperator {
         lookup: {
           type: 'static',
           fromRequest(parameters) {
-            return filters.parametersToLookupValue(parameters);
+            const value = filters.parametersToLookupValue(parameters);
+            if (filters.expands) {
+              if (typeof value != 'string') {
+                return [];
+              }
+              let values: SqliteJsonValue[] = JSON.parse(value);
+              if (!Array.isArray(values)) {
+                return [];
+              }
+
+              return values;
+            } else {
+              return [value];
+            }
           }
         },
         filterRow(options) {

packages/sync-rules/src/streams/from_sql.ts

@@ -76,11 +76,12 @@ class SyncStreamCompiler {
 
     const tools = new SqlTools({
       table: alias,
-      parameterTables: ['subscription_parameters', 'token_parameters'],
+      parameterTables: [],
       valueTables: [alias],
       sql: this.sql,
       schema: querySchema,
       supportsParameterExpressions: true,
+      supportsExpandingParameters: true, // needed for table.column IN (subscription.parameters() -> ...)
       isStream: true
     });
     tools.checkSpecificNameCase(tableRef);
@@ -307,7 +308,7 @@ class SyncStreamCompiler {
 
     const right = tools.compileClause(clause.right);
 
-    // For cases 3-5, we can actually uses SqlTools.compileClause. Case 3 and 4 are handled specially in there and return
+    // For cases 3-5, we can actually use SqlTools.compileClause. Case 3 and 4 are handled specially in there and return
     // a ParameterMatchClause, which we can translate via CompareRowValueWithStreamParameter. Case 5 is either a row-value
     // or a parameter-value clause which we can wrap in EvaluateSimpleCondition.
     const combined = tools.compileInClause(clause.left, left, clause.right, right);
@@ -366,7 +367,7 @@ class SyncStreamCompiler {
     // Create a new tools instance for this - the subquery does not have access to the outer one.
     const tools = new SqlTools({
       table: alias,
-      parameterTables: ['subscription_parameters', 'token_parameters'],
+      parameterTables: [],
       valueTables: [alias],
       sql: this.sql,
       schema: querySchema,

packages/sync-rules/src/streams/functions.ts

@@ -17,11 +17,29 @@ const subscription_parameters: SqlParameterFunction = {
   usesUnauthenticatedRequestParameters: true
 };
 
+const connection_parameters: SqlParameterFunction = {
+  debugName: 'connection.parameters',
+  call(parameters: ParameterValueSet) {
+    return parameters.rawUserParameters;
+  },
+  getReturnType() {
+    return ExpressionType.TEXT;
+  },
+  detail: 'Unauthenticated connection parameters as JSON',
+  documentation:
+    'Returns parameters passed by the client as a JSON string. These parameters are not authenticated - any value can be passed in by the client.',
+  usesAuthenticatedRequestParameters: false,
+  usesUnauthenticatedRequestParameters: true
+};
+
 export const STREAM_FUNCTIONS: Record<string, Record<string, SqlParameterFunction>> = {
   subscription: {
-    parameter: subscription_parameters
+    parameters: subscription_parameters
+  },
+  connection: {
+    parameters: connection_parameters
   },
-  request: {
+  token: {
     user_id: request_user_id,
     jwt: request_jwt
   }

packages/sync-rules/src/streams/parameter.ts

@@ -55,7 +55,14 @@ export interface SubqueryLookups {
  */
 export interface StaticLookup {
   type: 'static';
-  fromRequest(parameters: ParameterValueSet): SqliteValue | null;
+
+  /**
+   * The value this lookup evaluates to for a specific request.
+   *
+   * This is typically a singleton array, e.g. the user's id in `WHERE owner_id = token.user_id()`. To desugar `IN`
+   * queries on parameter data, this can also return multiple values though: `WHERE owner_id IN subscription.parameters() -> 'user_ids'`.
+   */
+  fromRequest(parameters: ParameterValueSet): SqliteValue[];
 }
 
 /**

packages/sync-rules/src/streams/variant.ts

@@ -146,7 +146,10 @@ export class StreamVariant {
     const staticBuckets: ResolvedBucket[] = [];
     if (dynamicParameters.length == 0 && dynamicRequestFilters.length == 0) {
       // When we have no dynamic parameters, the partial evaluation is a full instantiation.
-      staticBuckets.push(this.resolveBucket(stream, instantiation as SqliteJsonValue[], reason));
+      const instantiations = this.cartesianProductOfParameterInstantiations(instantiation as SqliteJsonValue[][]);
+      for (const instantiation of instantiations) {
+        staticBuckets.push(this.resolveBucket(stream, instantiation, reason));
+      }
     }
 
     const variant = this;
@@ -173,7 +176,13 @@ export class StreamVariant {
 
         const perParameterInstantiation: (SqliteJsonValue | BucketParameter)[][] = [];
         for (const parameter of instantiation) {
-          perParameterInstantiation.push([parameter]);
+          if (Array.isArray(parameter)) {
+            // Statically-resolved values
+            perParameterInstantiation.push(parameter);
+          } else {
+            // to be instantiated with dynamic lookup
+            perParameterInstantiation.push([parameter as BucketParameter]);
+          }
         }
 
         for (const lookup of dynamicParameters) {
@@ -189,13 +198,13 @@ export class StreamVariant {
     };
   }
 
-  findStaticInstantiation(params: RequestParameters): SqliteJsonValue[] | null {
+  findStaticInstantiations(params: RequestParameters): SqliteJsonValue[][] {
     if (this.subqueries.length) {
-      return null;
+      return [];
     }
 
     // This will be an array of values (i.e. a total evaluation) because there are no dynamic parameters.
-    return this.partiallyEvaluateParameters(params) as SqliteJsonValue[];
+    return this.partiallyEvaluateParameters(params) as SqliteJsonValue[][];
   }
 
   /**
@@ -239,23 +248,25 @@ export class StreamVariant {
    * Dynamic parameters that depend on subquery results are not replaced.
    * This returns null if there's a {@link StaticRequestFilter} that doesn't match the request.
    */
-  private partiallyEvaluateParameters(params: RequestParameters): (SqliteJsonValue | BucketParameter)[] | null {
+  private partiallyEvaluateParameters(params: RequestParameters): (SqliteJsonValue[] | BucketParameter)[] | null {
     for (const filter of this.requestFilters) {
       if (filter.type == 'static' && !filter.matches(params)) {
         return null;
       }
     }
 
-    const instantiation: (SqliteJsonValue | BucketParameter)[] = [];
+    const instantiation: (SqliteJsonValue[] | BucketParameter)[] = [];
     for (const parameter of this.parameters) {
       const lookup = parameter.lookup;
       if (lookup.type == 'static') {
-        const value = lookup.fromRequest(params);
-        if (isJsonValue(value)) {
-          instantiation.push(value);
-        } else {
+        const values = lookup.fromRequest(params)?.filter(isJsonValue);
+        if (values.length == 0) {
+          // Parameter not instantiable for this request. Since parameters in a single variant form a conjunction, that
+          // means the whole request won't find anything here.
           return null;
         }
+
+        instantiation.push(values);
       } else {
         instantiation.push(parameter);
       }

packages/sync-rules/test/src/streams.test.ts

@@ -318,6 +318,18 @@ describe('streams', () => {
         })
       ).toStrictEqual(['stream|1["b"]']);
     });
+
+    test('on parameter data', async () => {
+      const desc = parseStream("SELECT * FROM comments WHERE issue_id IN (subscription.parameters() -> 'issue_id')");
+
+      expect(evaluateBucketIds(desc, COMMENTS, { id: 'a', issue_id: 'i' })).toStrictEqual(['stream|0["i"]']);
+      expect(
+        await queryBucketIds(desc, {
+          token_parameters: { user_id: 'a' },
+          parameters: { issue_id: ['i1', 'i2'] }
+        })
+      ).toStrictEqual(['stream|0["i1"]', 'stream|0["i2"]']);
+    });
   });
 
   describe('overlap', () => {
@@ -400,6 +412,21 @@ describe('streams', () => {
         )
       ]);
     });
+
+    test('subquery with two columns', () => {
+      const [_, errors] = syncStreamFromSql(
+        's',
+        'select * from comments where issue_id in (select id, owner_id from issues where owner_id = request.user_id())',
+        options
+      );
+
+      expect(errors).toMatchObject([
+        expect.toBeSqlRuleError(
+          'This subquery must return exactly one column',
+          'select id, owner_id from issues where owner_id = request.user_id()'
+        )
+      ]);
+    });
   });
 
   describe('normalization', () => {