Introduce BucketSource interface

Author: simolus3

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

@@ -202,34 +202,7 @@ 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((source) => source.debugRepresentation()),
       source_tables: resolved_tables,
       data_tables: rules.debugGetOutputTables()
     };

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

@@ -1,7 +1,8 @@
 import {
   BucketDescription,
   BucketPriority,
-  isValidPriority,
+  BucketSource,
+  BucketSourceType,
   RequestedStream,
   RequestParameters,
   ResolvedBucket,
@@ -22,7 +23,6 @@ import { JSONBig } from '@powersync/service-jsonbig';
 import { BucketParameterQuerier } from '@powersync/service-sync-rules/src/BucketParameterQuerier.js';
 import { SyncContext } from './SyncContext.js';
 import { getIntersection, hasIntersection } from './util.js';
-import { SqlBucketDescriptor, SqlBucketDescriptorType } from '@powersync/service-sync-rules/src/SqlBucketDescriptor.js';
 
 export interface BucketChecksumStateOptions {
   syncContext: SyncContext;
@@ -234,14 +234,13 @@ 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 source of this.parameterState.syncRules.bucketSources) {
+        if (source.type == BucketSourceType.SYNC_STREAM && this.parameterState.isSubscribedToStream(source)) {
           subscriptions.push({
-            name: desc.name,
-            is_default: desc.subscribedToByDefault
+            name: source.name,
+            is_default: source.subscribedToByDefault
           });
         }
       }
@@ -455,7 +454,7 @@ export class BucketParameterState {
     };
   }
 
-  isSubscribedToStream(desc: SqlBucketDescriptor): boolean {
+  isSubscribedToStream(desc: BucketSource): boolean {
     return (desc.subscribedToByDefault && this.includeDefaultStreams) || this.subscribedStreamNames.has(desc.name);
   }
 

packages/sync-rules/src/BucketSource.ts

@@ -0,0 +1,81 @@
+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;
+  type: BucketSourceType;
+
+  subscribedToByDefault: boolean;
+
+  /**
+   * 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, tables: Record<string, Record<string, ColumnDefinition>>): void;
+
+  debugWriteOutputTables(result: Record<string, { query: string }[]>): void;
+
+  debugRepresentation(): any;
+}
+
+export enum BucketSourceType {
+  SYNC_RULE,
+  SYNC_STREAM
+}
+
+export type ResultSetDescription = { name: string; columns: ColumnDefinition[] };

packages/sync-rules/src/SqlBucketDescriptor.ts

@@ -1,12 +1,13 @@
-import { BucketDescription, BucketInclusionReason, ResolvedBucket } from './BucketDescription.js';
+import { BucketInclusionReason, ResolvedBucket } from './BucketDescription.js';
 import { BucketParameterQuerier, mergeBucketParameterQueriers } from './BucketParameterQuerier.js';
+import { BucketSource, BucketSourceType, ResultSetDescription } from './BucketSource.js';
+import { ColumnDefinition } from './ExpressionType.js';
 import { IdSequence } from './IdSequence.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
 import { SqlDataQuery } from './SqlDataQuery.js';
 import { SqlParameterQuery } from './SqlParameterQuery.js';
 import { GetQuerierOptions, SyncRulesOptions } from './SqlSyncRules.js';
 import { StaticSqlParameterQuery } from './StaticSqlParameterQuery.js';
-import { StreamQuery } from './StreamQuery.js';
 import { TablePattern } from './TablePattern.js';
 import { TableValuedFunctionSqlParameterQuery } from './TableValuedFunctionSqlParameterQuery.js';
 import { SqlRuleError } from './errors.js';
@@ -16,8 +17,8 @@ import {
   EvaluationResult,
   QueryParseOptions,
   RequestParameters,
-  SqliteRow,
-  StreamParseOptions
+  SourceSchema,
+  SqliteRow
 } from './types.js';
 
 export interface QueryParseResult {
@@ -29,23 +30,20 @@ export interface QueryParseResult {
   errors: SqlRuleError[];
 }
 
-export enum SqlBucketDescriptorType {
-  SYNC_RULE,
-  STREAM
-}
-
-export class SqlBucketDescriptor {
+export class SqlBucketDescriptor implements BucketSource {
   name: string;
   bucketParameters?: string[];
-  type: SqlBucketDescriptorType;
-  subscribedToByDefault: boolean;
 
-  constructor(name: string, type: SqlBucketDescriptorType) {
+  constructor(name: string) {
     this.name = name;
-    this.type = type;
+  }
 
-    // Sync-rule style buckets are subscribed to by default, streams are opt-in unless their definition says otherwise.
-    this.subscribedToByDefault = type == SqlBucketDescriptorType.SYNC_RULE;
+  get type(): BucketSourceType {
+    return BucketSourceType.SYNC_RULE;
+  }
+
+  public get subscribedToByDefault(): boolean {
+    return true;
   }
 
   /**
@@ -94,24 +92,6 @@ export class SqlBucketDescriptor {
     };
   }
 
-  addUnifiedStreamQuery(sql: string, options: StreamParseOptions): QueryParseResult {
-    const [query, errors] = StreamQuery.fromSql(this.name, sql, options);
-    for (const parameterQuery of query.inferredParameters) {
-      if (parameterQuery instanceof StaticSqlParameterQuery) {
-        this.globalParameterQueries.push(parameterQuery);
-      } else {
-        this.parameterQueries.push(parameterQuery);
-      }
-    }
-    this.dataQueries.push(query.data);
-    this.subscribedToByDefault = options.default ?? false;
-
-    return {
-      parsed: true,
-      errors
-    };
-  }
-
   evaluateRow(options: EvaluateRowOptions): EvaluationResult[] {
     let results: EvaluationResult[] = [];
     for (let query of this.dataQueries) {
@@ -137,20 +117,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
-  ) {
-    const reasons = [this.bucketInclusionReason(options)];
-    const staticBuckets = this.getStaticBucketDescriptions(parameters, reasons);
+  pushBucketParameterQueriers(result: BucketParameterQuerier[], options: GetQuerierOptions) {
+    const reasons = [this.bucketInclusionReason()];
+    const staticBuckets = this.getStaticBucketDescriptions(options.globalParameters, reasons);
     const staticQuerier = {
       staticBuckets,
       hasDynamicBuckets: false,
@@ -163,7 +139,9 @@ export class SqlBucketDescriptor {
       return;
     }
 
-    const dynamicQueriers = this.parameterQueries.map((query) => query.getBucketParameterQuerier(parameters, reasons));
+    const dynamicQueriers = this.parameterQueries.map((query) =>
+      query.getBucketParameterQuerier(options.globalParameters, reasons)
+    );
     result.push(...dynamicQueriers);
   }
 
@@ -198,12 +176,8 @@ export class SqlBucketDescriptor {
     return result;
   }
 
-  private bucketInclusionReason(parameters: GetQuerierOptions): BucketInclusionReason {
-    if (this.type == SqlBucketDescriptorType.STREAM && !this.subscribedToByDefault) {
-      return { subscription: this.name };
-    } else {
-      return 'default';
-    }
+  private bucketInclusionReason(): BucketInclusionReason {
+    return 'default';
   }
 
   tableSyncsData(table: SourceTableInterface): boolean {
@@ -223,4 +197,58 @@ export class SqlBucketDescriptor {
     }
     return false;
   }
+
+  resolveResultSets(schema: SourceSchema, tables: Record<string, Record<string, ColumnDefinition>>) {
+    for (let query of this.dataQueries) {
+      const outTables = query.getColumnOutputs(schema);
+      for (let table of outTables) {
+        tables[table.name] ??= {};
+        for (let column of table.columns) {
+          if (column.name != 'id') {
+            tables[table.name][column.name] ??= column;
+          }
+        }
+      }
+    }
+  }
+
+  debugWriteOutputTables(result: Record<string, { query: string }[]>): void {
+    for (let q of this.dataQueries) {
+      result[q.table!] ??= [];
+      const r = {
+        query: q.sql
+      };
+
+      result[q.table!].push(r);
+    }
+  }
+
+  debugRepresentation() {
+    let all_parameter_queries = [...this.parameterQueries.values()].flat();
+    let all_data_queries = [...this.dataQueries.values()].flat();
+    return {
+      name: this.name,
+      type: this.type.toString(),
+      bucket_parameters: this.bucketParameters,
+      global_parameter_queries: this.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()
+        };
+      })
+    };
+  }
 }