Support overlap operator

Author: simolus3

packages/sync-rules/src/sql_filters.ts

@@ -13,6 +13,7 @@ import {
   OPERATOR_JSON_EXTRACT_JSON,
   OPERATOR_JSON_EXTRACT_SQL,
   OPERATOR_NOT,
+  OPERATOR_OVERLAP,
   SQL_FUNCTIONS,
   SqlFunction,
   castOperator,
@@ -466,6 +467,98 @@ export class SqlTools {
     }
   }
 
+  compileOverlapClause(
+    left: Expr,
+    leftFilter: CompiledClause,
+    right: Expr,
+    rightFilter: CompiledClause
+  ): CompiledClause {
+    // Special cases:
+    //  parameterValue IN rowValue
+    //  rowValue IN parameterValue
+    // All others are handled by standard function composition
+
+    const composeType = this.getComposeType(OPERATOR_OVERLAP, [leftFilter, rightFilter], [left, right]);
+    if (composeType.errorClause != null) {
+      return composeType.errorClause;
+    } else if (composeType.argsType != null) {
+      // This is a standard supported configuration, takes precedence over
+      // the special cases below.
+      return this.composeFunction(OPERATOR_OVERLAP, [leftFilter, rightFilter], [left, right]);
+    } else if (isParameterValueClause(leftFilter) && isRowValueClause(rightFilter)) {
+      // token_parameters.value IN table.some_array
+      // bucket.param IN table.some_array
+      const inputParam = this.basicInputParameter(leftFilter);
+
+      return {
+        error: false,
+        inputParameters: [inputParam],
+        unbounded: true,
+        filterRow(tables: QueryParameters): TrueIfParametersMatch {
+          const aValue = rightFilter.evaluate(tables);
+          if (aValue == null) {
+            return MATCH_CONST_FALSE;
+          }
+          const values = JSON.parse(aValue as string);
+          if (!Array.isArray(values)) {
+            throw new Error('Not an array');
+          }
+          return values.map((value) => {
+            return { [inputParam.key]: value };
+          });
+        },
+        usesAuthenticatedRequestParameters: leftFilter.usesAuthenticatedRequestParameters,
+        usesUnauthenticatedRequestParameters: leftFilter.usesUnauthenticatedRequestParameters
+      } satisfies ParameterMatchClause;
+    } else if (
+      this.supportsExpandingParameters &&
+      isRowValueClause(leftFilter) &&
+      isParameterValueClause(rightFilter)
+    ) {
+      // table.some_value && token_parameters.some_array
+      // This expands into "OR(table_some_value = <value>)" for each value of both arrays.
+      // We only support one such filter per query
+      const key = `${rightFilter.key}[*]`;
+
+      const inputParam: InputParameter = {
+        key: key,
+        expands: true,
+        filteredRowToLookupValue: (filterParameters) => {
+          return filterParameters[key];
+        },
+        parametersToLookupValue: (parameters) => {
+          return rightFilter.lookupParameterValue(parameters);
+        }
+      };
+
+      return {
+        error: false,
+        inputParameters: [inputParam],
+        unbounded: false,
+        filterRow(tables: QueryParameters): TrueIfParametersMatch {
+          const value = leftFilter.evaluate(tables);
+          if (!isJsonValue(value)) {
+            // Cannot persist, e.g. BLOB
+            return MATCH_CONST_FALSE;
+          }
+
+          const values = JSON.parse(value as string);
+          if (!Array.isArray(values)) {
+            throw new Error('Not an array');
+          }
+          return values.map((value) => {
+            return { [inputParam.key]: value };
+          });
+        },
+        usesAuthenticatedRequestParameters: rightFilter.usesAuthenticatedRequestParameters,
+        usesUnauthenticatedRequestParameters: rightFilter.usesUnauthenticatedRequestParameters
+      } satisfies ParameterMatchClause;
+    } else {
+      // Not supported, return the error previously computed
+      return this.error(composeType.error!, composeType.errorExpr);
+    }
+  }
+
   parameterMatchClause(staticFilter: RowValueClause, otherFilter: ParameterValueClause) {
     const inputParam = this.basicInputParameter(otherFilter);
 

packages/sync-rules/src/sql_functions.ts

@@ -743,23 +743,47 @@ export function evaluateOperator(op: string, a: SqliteValue, b: SqliteValue): Sq
       return sqliteBool(sqliteBool(a) && sqliteBool(b));
     case 'OR':
       return sqliteBool(sqliteBool(a) || sqliteBool(b));
-    case 'IN':
+    case 'IN': {
       if (a == null || b == null) {
         return null;
       }
-      if (typeof b != 'string') {
-        throw new Error('IN is only supported on JSON arrays');
+      const bParsed = checkJsonArray(b, 'IN is only supported on JSON arrays');
+      return sqliteBool(bParsed.includes(a));
+    }
+    case '&&': {
+      // a && b evaluates to true iff they're both arrays and have a non-empty intersection.
+      if (a == null || b == null) {
+        return null;
       }
-      const bParsed = JSON.parse(b);
-      if (!Array.isArray(bParsed)) {
-        throw new Error('IN is only supported on JSON arrays');
+
+      const aParsed = checkJsonArray(a, '&& is only supported on JSON arrays');
+      const bParsed = checkJsonArray(a, '&& is only supported on JSON arrays');
+
+      for (const elementInA in aParsed) {
+        if (bParsed.includes(elementInA)) {
+          return sqliteBool(true);
+        }
       }
-      return sqliteBool(bParsed.includes(a));
+      return sqliteBool(false);
+    }
     default:
       throw new Error(`Operator not supported: ${op}`);
   }
 }
 
+export function checkJsonArray(value: SqliteValue, errorMessage: string): any[] {
+  if (typeof value != 'string') {
+    throw new Error(errorMessage);
+  }
+
+  const parsed = JSON.parse(value);
+  if (!Array.isArray(parsed)) {
+    throw new Error(value);
+  }
+
+  return parsed;
+}
+
 export function getOperatorReturnType(op: string, left: ExpressionType, right: ExpressionType) {
   switch (op) {
     case '=':
@@ -799,6 +823,7 @@ export function getOperatorReturnType(op: string, left: ExpressionType, right: E
     case 'OR':
       return ExpressionType.INTEGER;
     case 'IN':
+    case '&&':
       return ExpressionType.INTEGER;
     default:
       return ExpressionType.NONE;
@@ -928,6 +953,8 @@ export const OPERATOR_NOT: SqlFunction = {
 
 export const OPERATOR_IN = getOperatorFunction('IN');
 
+export const OPERATOR_OVERLAP = getOperatorFunction('&&');
+
 export function castOperator(castTo: string | undefined): SqlFunction | null {
   if (castTo == null || !CAST_TYPES.has(castTo)) {
     return null;

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

@@ -4,7 +4,7 @@ import { TablePattern } from '../TablePattern.js';
 import { ParameterMatchClause, ParameterValueClause, RowValueClause, SqliteJsonValue } from '../types.js';
 import { isJsonValue, normalizeParameterValue } from '../utils.js';
 import { SqlTools } from '../sql_filters.js';
-import { OPERATOR_NOT } from '../sql_functions.js';
+import { checkJsonArray, OPERATOR_NOT } from '../sql_functions.js';
 import { ParameterLookup } from '../BucketParameterQuerier.js';
 
 import { StreamVariant } from './variant.js';
@@ -107,6 +107,10 @@ export class Subquery {
       const id = context.queryCounter.toString();
       context.queryCounter++;
 
+      if (variant.parameters.length == 0) {
+        throw new Error('Unsupported subquery without parameter, must depend on request parameters');
+      }
+
       return [variant, id] satisfies [StreamVariant, string];
     });
 
@@ -175,38 +179,36 @@ export class InOperator extends FilterOperator {
 
   compile(context: BucketCompilationContext): void {
     const subqueryEvaluator = this.right.compileEvaluator(context);
-
     const filter = this.left;
-    // Something like `SELECT * FROM comments WHERE issue_id IN (SELECT id FROM issue WHERE owner_id = request.user())`
-    // This groups rows into buckets identified by comments.issue_id, which happens in filterRow.
-    // When a user connects, we need to resolve all the issue ids they own. This happens with an indirection:
-    //  1. In the subquery evaluator, we create an index from owner_id to issue ids.
-    //  2. When we have users, we use that index to find issue ids dynamically, with which we can build the buckets
-    //     to sync.
-    context.currentVariant.parameters.push({
-      lookup: {
-        type: 'in',
-        subquery: subqueryEvaluator
-      },
-      filterRow(options) {
-        const tables = { [options.sourceTable.table]: options.record };
-        const value = filter.evaluate(tables);
-        if (isJsonValue(value)) {
-          return [normalizeParameterValue(value)];
-        } else {
-          return [];
-        }
-      }
-    });
 
     if (isRowValueClause(this.left)) {
+      // Something like `SELECT * FROM comments WHERE issue_id IN (SELECT id FROM issue WHERE owner_id = request.user())`
+      // This groups rows into buckets identified by comments.issue_id, which happens in filterRow.
+      // When a user connects, we need to resolve all the issue ids they own. This happens with an indirection:
+      //  1. In the subquery evaluator, we create an index from owner_id to issue ids.
+      //  2. When we have users, we use that index to find issue ids dynamically, with which we can build the buckets
+      //     to sync.
+      context.currentVariant.parameters.push({
+        lookup: {
+          type: 'in',
+          subquery: subqueryEvaluator
+        },
+        filterRow(options) {
+          const tables = { [options.sourceTable.table]: options.record };
+          const value = filter.evaluate(tables);
+          if (isJsonValue(value)) {
+            return [normalizeParameterValue(value)];
+          } else {
+            return [];
+          }
+        }
+      });
     } else if (isParameterValueClause(this.left)) {
       // Something like `SELECT * FROM comments WHERE request.user_id() IN (SELECT id FROM users WHERE is_admin)`.
       // This doesn't introduce a bucket parameter, but we still need to create a parameter lookup to determine whether
       // a given user should have access to the bucket or not. Because lookups can only be exact (we can't evaluate
       // `SELECT id FROM users WHERE is_admin` into a single ParameterLookup), we need to push the filter down into
       // the subquery, e.g. `WHERE EXISTS(SELECT id FROM users WHERE is_admin AND id = request.user_id())`.
-      const left = this.left;
       const subqueryEvaluator = this.right.compileEvaluator(context);
 
       context.currentVariant.requestFilters.push({
@@ -222,6 +224,42 @@ export class InOperator extends FilterOperator {
   }
 }
 
+/**
+ * An operator of the form `<left> && <right>`, where `right` is a subqery.
+ */
+export class OverlapOperator extends FilterOperator {
+  private left: RowValueClause;
+  private right: Subquery;
+
+  constructor(left: RowValueClause, right: Subquery) {
+    super();
+    this.left = left;
+    this.right = right;
+  }
+
+  compile(context: BucketCompilationContext): void {
+    const subqueryEvaluator = this.right.compileEvaluator(context);
+    const filter = this.left;
+
+    context.currentVariant.parameters.push({
+      lookup: {
+        type: 'overlap',
+        subquery: subqueryEvaluator
+      },
+      filterRow(options) {
+        const tables = { [options.sourceTable.table]: options.record };
+        const value = filter.evaluate(tables);
+        if (value == null) {
+          return [];
+        }
+
+        const parsed = checkJsonArray(value, 'Left side of && must evaluate to an array');
+        return parsed.map(normalizeParameterValue);
+      }
+    });
+  }
+}
+
 /**
  * A filter that matches when _something_ exists in a subquery.
  *

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

@@ -26,6 +26,7 @@ import {
   InOperator,
   Not,
   Or,
+  OverlapOperator,
   ScalarExpression,
   Subquery
 } from './filter.js';
@@ -297,7 +298,35 @@ class SyncStreamCompiler {
   }
 
   private compileOverlapOperator(tools: SqlTools, clause: ExprBinary): FilterOperator {
-    throw 'todo';
+    const left = tools.compileClause(clause.left);
+    if (isClauseError(left)) {
+      return recoverErrorClause(tools);
+    }
+
+    if (clause.right.type == 'select') {
+      if (!isRowValueClause(left)) {
+        if (!isClauseError(left)) {
+          tools.error('The left-hand side of an && operator must be derived from the row to sync..', clause.left);
+        }
+
+        return recoverErrorClause(tools);
+      }
+
+      const subqueryResult = this.compileSubquery(clause.right);
+      if (!subqueryResult) {
+        return recoverErrorClause(tools);
+      }
+      const [subquery] = subqueryResult;
+      return new OverlapOperator(left, subquery);
+    }
+
+    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
+    // 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.compileOverlapClause(clause.left, left, clause.right, right);
+    return compiledClauseToFilter(tools, combined);
   }
 
   private compileSubquery(stmt: SelectStatement): [Subquery, SqlTools] | undefined {