Sync streams evaluator

packages/sync-rules/src/BaseSqlDataQuery.ts

@@ -15,7 +15,7 @@ import {
   SqliteJsonRow,
   SqliteRow
 } from './types.js';
-import { filterJsonRow } from './utils.js';
+import { filterJsonRow, idFromData } from './utils.js';
 
 export interface RowValueExtractor {
   extract(tables: QueryParameters, into: SqliteRow): void;
@@ -183,16 +183,7 @@ export class BaseSqlDataQuery {
       }
 
       const data = this.transformRow(tables);
-      let id = data.id;
-      if (typeof id != 'string') {
-        // While an explicit cast would be better, this covers against very common
-        // issues when initially testing out sync, for example when the id column is an
-        // auto-incrementing integer.
-        // If there is no id column, we use a blank id. This will result in the user syncing
-        // a single arbitrary row for this table - better than just not being able to sync
-        // anything.
-        id = castAsText(id) ?? '';
-      }
+      const id = idFromData(data);
       const outputTable = this.getOutputName(table.name);
 
       return resolvedBucketParameters.map((serializedBucketParameters) => {

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

@@ -178,11 +178,12 @@ export class PostgresToSqlite {
           return this.invalidExpression(expr.function, 'DISTINCT, ORDER BY, FILTER and OVER clauses are not supported');
         }
 
-        const forbiddenReason = forbiddenFunctions[expr.function.name];
+        const functionName = expr.function.name.toLowerCase();
+        const forbiddenReason = forbiddenFunctions[functionName];
         if (forbiddenReason) {
           return this.invalidExpression(expr.function, `Forbidden call: ${forbiddenReason}`);
         }
-        let allowedArgs = supportedFunctions[expr.function.name];
+        let allowedArgs = supportedFunctions[functionName];
         if (allowedArgs == null) {
           return this.invalidExpression(expr.function, 'Unknown function');
         } else {
@@ -204,7 +205,7 @@ export class PostgresToSqlite {
 
         return {
           type: 'function',
-          function: expr.function.name,
+          function: functionName,
           parameters: expr.args.map((a) => this.translateNodeWithLocation(a))
         };
       }
@@ -216,6 +217,9 @@ export class PostgresToSqlite {
         const left = this.translateNodeWithLocation(expr.left);
         const right = this.translateNodeWithLocation(expr.right);
         if (expr.op === 'LIKE') {
+          // We don't support LIKE in the old bucket definition system, and want to make sure we're clear about ICU,
+          // case sensitivity and changing the escape character first. TODO: Support later.
+          this.options.errors.report('LIKE expressions are not currently supported.', expr);
           return { type: 'function', function: 'like', parameters: [left, right] };
         } else if (expr.op === 'NOT LIKE') {
           return {
@@ -243,8 +247,9 @@ export class PostgresToSqlite {
         let rightHandSideOfIs: SqlExpression<ExpressionInput>;
 
         switch (expr.op) {
-          case '+':
           case '-':
+            return this.invalidExpression(expr, 'Unary minus is not currently supported');
+          case '+':
             return { type: 'unary', operator: expr.op, operand: this.translateNodeWithLocation(expr.operand) };
           case 'NOT':
             return { type: 'unary', operator: 'not', operand: this.translateNodeWithLocation(expr.operand) };

packages/sync-rules/src/index.ts

@@ -33,3 +33,6 @@ export * from './HydratedSyncRules.js';
 export * from './compiler/compiler.js';
 export * from './sync_plan/plan.js';
 export { serializeSyncPlan } from './sync_plan/serialize.js';
+export { addPrecompiledSyncPlanToRules } from './sync_plan/evaluator/index.js';
+export { javaScriptExpressionEngine } from './sync_plan/engine/javascript.js';
+export { nodeSqliteExpressionEngine } from './sync_plan/engine/sqlite.js';

packages/sync-rules/src/sql_functions.ts

@@ -1102,7 +1102,7 @@ const TYPE_ORDERING = {
   blob: 3
 };
 
-function compare(a: SqliteValue, b: SqliteValue): number {
+export function compare(a: SqliteValue, b: SqliteValue): number {
   // https://www.sqlite.org/datatype3.html#comparisons
   if (a == null && b == null) {
     // Only for IS / IS NOT

packages/sync-rules/src/sync_plan/engine/javascript.ts

@@ -0,0 +1,283 @@
+import {
+  compare,
+  CompatibilityContext,
+  generateSqlFunctions,
+  SQLITE_FALSE,
+  SQLITE_TRUE,
+  sqliteBool,
+  sqliteNot
+} from '../../index.js';
+import { cast, evaluateOperator, SqlFunction } from '../../sql_functions.js';
+import { cartesianProduct } from '../../streams/utils.js';
+import { generateTableValuedFunctions } from '../../TableValuedFunctions.js';
+import { SqliteRow, SqliteValue } from '../../types.js';
+import {
+  ExternalData,
+  UnaryExpression,
+  BinaryExpression,
+  BetweenExpression,
+  ScalarInExpression,
+  CaseWhenExpression,
+  CastExpression,
+  ScalarFunctionCallExpression,
+  LiteralExpression,
+  SqlExpression
+} from '../expression.js';
+import { ExpressionVisitor, visitExpr } from '../expression_visitor.js';
+import {
+  ScalarExpressionEngine,
+  ScalarExpressionEvaluator,
+  TableValuedFunction,
+  TableValuedFunctionOutput
+} from './scalar_expression_engine.js';
+
+/**
+ * Creates a {@link ScalarExpressionEngine} implemented by evaluating scalar expressions in JavaScript.
+ */
+export function javaScriptExpressionEngine(compatibility: CompatibilityContext): ScalarExpressionEngine {
+  const tableValued = generateTableValuedFunctions(compatibility);
+  const regularFunctions = generateSqlFunctions(compatibility);
+  const compiler = new ExpressionToJavaScriptFunction({
+    named: regularFunctions.named,
+    jsonExtractJson: regularFunctions.operatorJsonExtractJson,
+    jsonExtractSql: regularFunctions.operatorJsonExtractSql
+  });
+
+  return {
+    close() {},
+    prepareEvaluator({ outputs = [], filters = [], tableValuedFunctions = [] }): ScalarExpressionEvaluator {
+      function compileScalar(expr: SqlExpression<number | TableValuedFunctionOutput>) {
+        return compiler.compile(expr);
+      }
+
+      const resolvedTableValuedFunctions = tableValuedFunctions.map((fn) => {
+        const found = tableValued[fn.name];
+        if (found == null) {
+          throw new Error(`Unknown table-valued function: ${fn.name}`);
+        }
+
+        const inputs = fn.inputs.map(compileScalar);
+        return {
+          original: fn,
+          evaluate: (input: PendingStatementEvaluation) => {
+            return found.call(inputs.map((f) => f(input)));
+          }
+        };
+      });
+
+      const columns = outputs.map(compileScalar);
+      const compiledFilters = filters.map(compileScalar);
+
+      return {
+        evaluate(inputs) {
+          // First, evaluate table-valued functions (if any).
+          const perFunctionResults: [TableValuedFunction, SqliteRow][][] = [];
+
+          for (const { original, evaluate } of resolvedTableValuedFunctions) {
+            perFunctionResults.push(
+              evaluate({ inputs }).map((row) => [original, row] satisfies [TableValuedFunction, SqliteRow])
+            );
+          }
+
+          const rows: SqliteValue[][] = [];
+          // We're doing an inner join on all table-valued functions, which we implement as a cross join on which each
+          // filter is evaluated. Having more than one table-valued function per statement would be very rare in
+          // practice.
+          row: for (const sourceRow of cartesianProduct(...perFunctionResults)) {
+            const byFunction = new Map<TableValuedFunction, SqliteRow>();
+            for (const [fn, output] of sourceRow) {
+              byFunction.set(fn, output);
+            }
+
+            const input: PendingStatementEvaluation = { inputs, row: byFunction };
+
+            for (const filter of compiledFilters) {
+              if (!sqliteBool(filter(input))) {
+                continue row;
+              }
+            }
+            rows.push(columns.map((c) => c(input)));
+          }
+
+          return rows;
+        }
+      };
+    }
+  };
+}
+
+interface PendingStatementEvaluation {
+  inputs: SqliteValue[];
+  row?: Map<TableValuedFunction, SqliteRow>;
+}
+
+type ExpressionImplementation = (input: PendingStatementEvaluation) => SqliteValue;
+
+interface KnownFunctions {
+  named: Record<string, SqlFunction>;
+  jsonExtractJson: SqlFunction; // -> operator
+  jsonExtractSql: SqlFunction; // ->> operator
+}
+
+class ExpressionToJavaScriptFunction
+  implements ExpressionVisitor<number | TableValuedFunctionOutput, ExpressionImplementation, null>
+{
+  constructor(readonly functions: KnownFunctions) {}
+
+  compile(expr: SqlExpression<number | TableValuedFunctionOutput>): ExpressionImplementation {
+    return visitExpr(this, expr, null);
+  }
+
+  visitExternalData(expr: ExternalData<number | TableValuedFunctionOutput>, arg: null): ExpressionImplementation {
+    if (typeof expr.source === 'number') {
+      const index = expr.source;
+      //  -1 because variables in SQLite are 1-indexed.
+      return ({ inputs }) => inputs[index - 1];
+    } else {
+      const { column, function: fn } = expr.source;
+
+      return ({ row }) => {
+        const result = row!.get(fn)!;
+        return result[column];
+      };
+    }
+  }
+
+  visitUnaryExpression(expr: UnaryExpression<number | TableValuedFunctionOutput>): ExpressionImplementation {
+    const operand = this.compile(expr.operand);
+
+    switch (expr.operator) {
+      case '+':
+        return operand;
+      case 'not':
+        return (input) => sqliteNot(operand(input));
+      // case '~':
+      // case '-':
+      //   throw new Error(`unary operator not supported: ${expr.operator}`);
+    }
+  }
+
+  visitBinaryExpression(expr: BinaryExpression<number | TableValuedFunctionOutput>): ExpressionImplementation {
+    const left = this.compile(expr.left);
+    const right = this.compile(expr.right);
+    const operator = expr.operator.toUpperCase();
+
+    return (input) => evaluateOperator(operator, left(input), right(input));
+  }
+
+  visitBetweenExpression(expr: BetweenExpression<number | TableValuedFunctionOutput>): ExpressionImplementation {
+    const low = this.compile(expr.low);
+    const high = this.compile(expr.high);
+    const value = this.compile(expr.value);
+
+    return (input) => {
+      const evaluatedValue = value(input);
+
+      const geqLow = evaluateOperator('>=', evaluatedValue, low(input));
+      const leqHigh = evaluateOperator('<=', evaluatedValue, high(input));
+      if (geqLow == null || leqHigh == null) {
+        return null;
+      }
+
+      return sqliteBool(geqLow) && sqliteBool(leqHigh);
+    };
+  }
+
+  visitScalarInExpression(expr: ScalarInExpression<number | TableValuedFunctionOutput>): ExpressionImplementation {
+    const target = this.compile(expr.target);
+    const inQuery = expr.in.map((q) => this.compile(q));
+
+    return (input) => {
+      const evaluatedTarget = target(input);
+      if (evaluatedTarget == null) {
+        return null;
+      }
+
+      let hasNullQuery = false;
+      for (const q of inQuery) {
+        const evaluated = q(input);
+        if (evaluated == null) {
+          hasNullQuery = true;
+          continue;
+        }
+
+        if (compare(evaluatedTarget, evaluated) == 0) {
+          return SQLITE_TRUE;
+        }
+      }
+
+      return hasNullQuery ? null : SQLITE_FALSE;
+    };
+  }
+
+  visitCaseWhenExpression(expr: CaseWhenExpression<number | TableValuedFunctionOutput>): ExpressionImplementation {
+    const compiledWhens = expr.whens.map((w) => ({ when: this.compile(w.when), then: this.compile(w.then) }));
+    const compiledElse = expr.else && this.compile(expr.else);
+
+    if (expr.operand) {
+      const operand = this.compile(expr.operand);
+      return (input) => {
+        const evaluatedOperand = operand(input);
+
+        for (const { when, then } of compiledWhens) {
+          if (evaluateOperator('=', evaluatedOperand, when(input))) {
+            return then(input);
+          }
+        }
+
+        return compiledElse ? compiledElse(input) : null;
+      };
+    } else {
+      return (input) => {
+        for (const { when, then } of compiledWhens) {
+          if (sqliteBool(when(input))) {
+            return then(input);
+          }
+        }
+
+        return compiledElse ? compiledElse(input) : null;
+      };
+    }
+  }
+
+  visitCastExpression(expr: CastExpression<number | TableValuedFunctionOutput>): ExpressionImplementation {
+    const operand = this.compile(expr.operand);
+    return (input) => {
+      return cast(operand(input), expr.cast_as);
+    };
+  }
+
+  visitScalarFunctionCallExpression(
+    expr: ScalarFunctionCallExpression<number | TableValuedFunctionOutput>
+  ): ExpressionImplementation {
+    let fnImpl: SqlFunction;
+    if (expr.function === '->') {
+      fnImpl = this.functions.jsonExtractJson;
+    } else if (expr.function === '->>') {
+      fnImpl = this.functions.jsonExtractSql;
+    } else {
+      fnImpl = this.functions.named[expr.function];
+      if (!fnImpl) {
+        throw new Error(`Function not implemented: ${expr.function}`);
+      }
+    }
+
+    const args = expr.parameters.map((p) => this.compile(p));
+    return (input) => {
+      return fnImpl.call(...args.map((f) => f(input)));
+    };
+  }
+
+  visitLiteralExpression(expr: LiteralExpression): ExpressionImplementation {
+    switch (expr.type) {
+      case 'lit_null':
+        return () => null;
+      case 'lit_double':
+        return () => expr.value;
+      case 'lit_int':
+        return () => BigInt(expr.base10);
+      case 'lit_string':
+        return () => expr.value;
+    }
+  }
+}