Start evaluating rows for sync streams

Author: simolus3

packages/service-core/src/routes/endpoints/sync-rules.ts

@@ -1,5 +1,5 @@
 import { ErrorCode, errors, router, schema } from '@powersync/lib-services-framework';
-import { SqlSyncRules, SyncRulesErrors } from '@powersync/service-sync-rules';
+import { SqlBucketDescriptor, SqlSyncRules, SyncRulesErrors } from '@powersync/service-sync-rules';
 import type { FastifyPluginAsync } from 'fastify';
 import * as t from 'ts-codec';
 
@@ -202,33 +202,36 @@ async function debugSyncRules(apiHandler: RouteAPI, sync_rules: string) {
 
     return {
       valid: true,
-      bucket_definitions: rules.bucketDescriptors.map((d) => {
-        let all_parameter_queries = [...d.parameterQueries.values()].flat();
-        let all_data_queries = [...d.dataQueries.values()].flat();
-        return {
-          name: d.name,
-          bucket_parameters: d.bucketParameters,
-          global_parameter_queries: d.globalParameterQueries.map((q) => {
-            return {
-              sql: q.sql
-            };
-          }),
-          parameter_queries: all_parameter_queries.map((q) => {
-            return {
-              sql: q.sql,
-              table: q.sourceTable,
-              input_parameters: q.inputParameters
-            };
-          }),
-
-          data_queries: all_data_queries.map((q) => {
-            return {
-              sql: q.sql,
-              table: q.sourceTable,
-              columns: q.columnOutputNames()
-            };
-          })
-        };
+      bucket_definitions: rules.bucketSources.map((d) => {
+        // TODO: Equivalent for streams
+        if (d instanceof SqlBucketDescriptor) {
+          let all_parameter_queries = [...d.parameterQueries.values()].flat();
+          let all_data_queries = [...d.dataQueries.values()].flat();
+          return {
+            name: d.name,
+            bucket_parameters: d.bucketParameters,
+            global_parameter_queries: d.globalParameterQueries.map((q) => {
+              return {
+                sql: q.sql
+              };
+            }),
+            parameter_queries: all_parameter_queries.map((q) => {
+              return {
+                sql: q.sql,
+                table: q.sourceTable,
+                input_parameters: q.inputParameters
+              };
+            }),
+
+            data_queries: all_data_queries.map((q) => {
+              return {
+                sql: q.sql,
+                table: q.sourceTable,
+                columns: q.columnOutputNames()
+              };
+            })
+          };
+        }
       }),
       source_tables: resolved_tables,
       data_tables: rules.debugGetOutputTables()

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

@@ -1,11 +1,11 @@
 import {
   BucketDescription,
   BucketPriority,
-  isValidPriority,
   RequestedStream,
   RequestParameters,
   ResolvedBucket,
-  SqlSyncRules
+  SqlSyncRules,
+  SyncStream
 } from '@powersync/service-sync-rules';
 
 import * as storage from '../storage/storage-index.js';
@@ -234,11 +234,10 @@ export class BucketChecksumState {
         this.logger.info(message, { checkpoint: base.checkpoint, user_id: user_id, buckets: allBuckets.length });
       };
       bucketsToFetch = allBuckets;
-      this.parameterState.syncRules.bucketDescriptors;
 
       const subscriptions: util.StreamDescription[] = [];
-      for (const desc of this.parameterState.syncRules.bucketDescriptors) {
-        if (desc.type == SqlBucketDescriptorType.STREAM && this.parameterState.isSubscribedToStream(desc)) {
+      for (const desc of this.parameterState.syncRules.bucketSources) {
+        if (desc instanceof SyncStream && this.parameterState.isSubscribedToStream(desc)) {
           subscriptions.push({
             name: desc.name,
             is_default: desc.subscribedToByDefault
@@ -455,7 +454,7 @@ export class BucketParameterState {
     };
   }
 
-  isSubscribedToStream(desc: SqlBucketDescriptor): boolean {
+  isSubscribedToStream(desc: SyncStream): boolean {
     return (desc.subscribedToByDefault && this.includeDefaultStreams) || this.subscribedStreamNames.has(desc.name);
   }
 

packages/sync-rules/package.json

@@ -24,11 +24,13 @@
     "@powersync/service-jsonbig": "workspace:^",
     "@syncpoint/wkx": "^0.5.0",
     "ajv": "^8.12.0",
+    "lodash": "^4.17.21",
     "pgsql-ast-parser": "^11.1.0",
     "uuid": "^11.1.0",
     "yaml": "^2.3.1"
   },
   "devDependencies": {
+    "@types/lodash": "^4.17.5",
     "@types/node": "^22.16.2",
     "vitest": "^3.0.5"
   }

packages/sync-rules/src/BaseSqlDataQuery.ts

@@ -4,14 +4,29 @@ import { ColumnDefinition } from './ExpressionType.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
 import { SqlTools } from './sql_filters.js';
 import { TablePattern } from './TablePattern.js';
-import { QueryParameters, QuerySchema, SourceSchema, SourceSchemaTable, SqliteJsonRow, SqliteRow } from './types.js';
+import {
+  EvaluationResult,
+  QueryParameters,
+  QuerySchema,
+  SourceSchema,
+  SourceSchemaTable,
+  SqliteJsonRow,
+  SqliteRow
+} from './types.js';
 import { filterJsonRow } from './utils.js';
+import { castAsText } from './sql_functions.js';
 
 export interface RowValueExtractor {
   extract(tables: QueryParameters, into: SqliteRow): void;
   getTypes(schema: QuerySchema, into: Record<string, ColumnDefinition>): void;
 }
 
+export interface EvaluateRowOptions {
+  table: SourceTableInterface;
+  row: SqliteRow;
+  bucketIds: (params: QueryParameters) => string[];
+}
+
 export interface BaseSqlDataQueryOptions {
   sourceTable: TablePattern;
   table: string;
@@ -149,6 +164,39 @@ export class BaseSqlDataQuery {
     return result;
   }
 
+  evaluateRowWithOptions(options: EvaluateRowOptions): EvaluationResult[] {
+    try {
+      const { table, row, bucketIds } = options;
+
+      const tables = { [this.table]: this.addSpecialParameters(table, row) };
+      const resolvedBucketIds = bucketIds(tables);
+
+      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 outputTable = this.getOutputName(table.table);
+
+      return resolvedBucketIds.map((bucketId) => {
+        return {
+          bucket: bucketId,
+          table: outputTable,
+          id: id,
+          data
+        } as EvaluationResult;
+      });
+    } catch (e) {
+      return [{ error: e.message ?? `Evaluating data query failed` }];
+    }
+  }
+
   protected transformRow(tables: QueryParameters): SqliteJsonRow {
     let result: SqliteRow = {};
     for (let extractor of this.extractors) {

packages/sync-rules/src/BucketSource.ts

@@ -0,0 +1,69 @@
+import { BucketParameterQuerier, ParameterLookup } from './BucketParameterQuerier.js';
+import { ColumnDefinition } from './ExpressionType.js';
+import { SourceTableInterface } from './SourceTableInterface.js';
+import { GetQuerierOptions } from './SqlSyncRules.js';
+import { TablePattern } from './TablePattern.js';
+import { EvaluatedParametersResult, EvaluateRowOptions, EvaluationResult, SourceSchema, SqliteRow } from './types.js';
+
+/**
+ * An interface declaring
+ *
+ *  - which buckets the sync service should create when processing change streams from the database.
+ *  - how data in source tables maps to data in buckets (e.g. when we're not selecting all columns).
+ *  - which buckets a given connection has access to.
+ *
+ * There are two ways to define bucket sources: Via sync rules made up of parameter and data queries, and via stream
+ * definitions that only consist of a single query.
+ */
+export interface BucketSource {
+  name: string;
+
+  /**
+   * Given a row in a source table that affects sync parameters, returns a structure to index which buckets rows should
+   * be associated with.
+   *
+   * The returned {@link ParameterLookup} can be referenced by {@link pushBucketParameterQueriers} to allow the storage
+   * system to find buckets.
+   */
+  evaluateParameterRow(sourceTable: SourceTableInterface, row: SqliteRow): EvaluatedParametersResult[];
+
+  /**
+   * Given a row as it appears in a table that affects sync data, return buckets, logical table names and transformed
+   * data for rows to add to buckets.
+   */
+  evaluateRow(options: EvaluateRowOptions): EvaluationResult[];
+
+  /**
+   * Reports {@link BucketParameterQuerier}s resolving buckets that a specific stream request should have access to.
+   *
+   * @param result The target array to insert queriers into.
+   * @param options Options, including parameters that may affect the buckets loaded by this source.
+   */
+  pushBucketParameterQueriers(result: BucketParameterQuerier[], options: GetQuerierOptions): void;
+
+  /**
+   * Whether {@link pushBucketParameterQueriers} may include a querier where
+   * {@link BucketParameterQuerier.hasDynamicBuckets} is true.
+   *
+   * This is mostly used for testing.
+   */
+  hasDynamicBucketQueries(): boolean;
+
+  getSourceTables(): Set<TablePattern>;
+
+  /** Whether the table possibly affects the buckets resolved by this source. */
+  tableSyncsParameters(table: SourceTableInterface): boolean;
+
+  /** Whether the table possibly affects the contents of buckets resolved by this source. */
+  tableSyncsData(table: SourceTableInterface): boolean;
+
+  /**
+   * Given a static schema, infer all logical tables and associated columns that appear in buckets defined by this
+   * source.
+   *
+   * This is use to generate the client-side schema.
+   */
+  resolveResultSets(schema: SourceSchema): ResultSetDescription[];
+}
+
+export type ResultSetDescription = { name: string; columns: ColumnDefinition[] };

packages/sync-rules/src/SqlBucketDescriptor.ts

@@ -1,5 +1,6 @@
 import { BucketDescription, BucketInclusionReason, ResolvedBucket } from './BucketDescription.js';
 import { BucketParameterQuerier, mergeBucketParameterQueriers } from './BucketParameterQuerier.js';
+import { BucketSource, ResultSetDescription } from './BucketSource.js';
 import { IdSequence } from './IdSequence.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
 import { SqlDataQuery } from './SqlDataQuery.js';
@@ -16,6 +17,7 @@ import {
   EvaluationResult,
   QueryParseOptions,
   RequestParameters,
+  SourceSchema,
   SqliteRow,
   StreamParseOptions
 } from './types.js';
@@ -34,7 +36,7 @@ export enum SqlBucketDescriptorType {
   STREAM
 }
 
-export class SqlBucketDescriptor {
+export class SqlBucketDescriptor implements BucketSource {
   name: string;
   bucketParameters?: string[];
   type: SqlBucketDescriptorType;
@@ -137,19 +139,16 @@ export class SqlBucketDescriptor {
   /**
    * @deprecated Use `pushBucketParameterQueriers` instead and merge at the top-level.
    */
-  getBucketParameterQuerier(options: GetQuerierOptions, parameters: RequestParameters): BucketParameterQuerier {
+  getBucketParameterQuerier(options: GetQuerierOptions): BucketParameterQuerier {
     const queriers: BucketParameterQuerier[] = [];
-    this.pushBucketParameterQueriers(queriers, options, parameters);
+    this.pushBucketParameterQueriers(queriers, options);
 
     return mergeBucketParameterQueriers(queriers);
   }
 
-  pushBucketParameterQueriers(
-    result: BucketParameterQuerier[],
-    options: GetQuerierOptions,
-    parameters: RequestParameters
-  ) {
+  pushBucketParameterQueriers(result: BucketParameterQuerier[], options: GetQuerierOptions) {
     const reasons = [this.bucketInclusionReason(options)];
+    const parameters = options.globalParameters;
     const staticBuckets = this.getStaticBucketDescriptions(parameters, reasons);
     const staticQuerier = {
       staticBuckets,
@@ -223,4 +222,13 @@ export class SqlBucketDescriptor {
     }
     return false;
   }
+
+  resolveResultSets(schema: SourceSchema): ResultSetDescription[] {
+    const descriptions: ResultSetDescription[] = [];
+    for (const query of this.dataQueries) {
+      descriptions.push(...query.getColumnOutputs(schema));
+    }
+
+    return descriptions;
+  }
 }

packages/sync-rules/src/SqlDataQuery.ts

@@ -186,36 +186,15 @@ export class SqlDataQuery extends BaseSqlDataQuery {
   }
 
   evaluateRow(table: SourceTableInterface, row: SqliteRow): EvaluationResult[] {
-    try {
-      const tables = { [this.table]: this.addSpecialParameters(table, row) };
-      const bucketParameters = this.filter.filterRow(tables);
-      const bucketIds = bucketParameters.map((params) =>
-        getBucketId(this.descriptorName, this.bucketParameters, params)
-      );
-
-      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 query = this;
+
+    return this.evaluateRowWithOptions({
+      table,
+      row,
+      bucketIds(tables) {
+        const bucketParameters = query.filter.filterRow(tables);
+        return bucketParameters.map((params) => getBucketId(query.descriptorName, query.bucketParameters, params));
       }
-      const outputTable = this.getOutputName(table.table);
-
-      return bucketIds.map((bucketId) => {
-        return {
-          bucket: bucketId,
-          table: outputTable,
-          id: id,
-          data
-        } as EvaluationResult;
-      });
-    } catch (e) {
-      return [{ error: e.message ?? `Evaluating data query failed` }];
-    }
+    });
   }
 }