Use separate sources per sync stream variant.

Author: rkistner

packages/sync-rules/src/BaseSqlDataQuery.ts

@@ -182,6 +182,10 @@ export class BaseSqlDataQuery {
 
       const tables = { [this.table.nameInSchema]: this.addSpecialParameters(table, row) };
       const resolvedBucketIds = bucketIds(tables);
+      if (resolvedBucketIds.length == 0) {
+        // Short-circuit: No need to transform the row if there are no matching buckets.
+        return [];
+      }
 
       const data = this.transformRow(tables);
       let id = data.id;

packages/sync-rules/src/BucketSource.ts

@@ -16,6 +16,12 @@ export interface CreateSourceParams {
   bucketIdTransformer: BucketIdTransformer;
 }
 
+/**
+ * A BucketSource is a _logical_ bucket or sync stream definition. It is primarily used to group together
+ * related BucketDataSource, BucketParameterLookupSource and BucketParameterQuerierSource definitions,
+ * for the purpose of subscribing to specific streams. It does not directly define the implementation
+ * or replication process.
+ */
 export interface BucketSource {
   readonly name: string;
   readonly type: BucketSourceType;
@@ -116,8 +122,6 @@ export interface BucketParameterQuerierSourceDefinition {
  * definitions that only consist of a single query.
  */
 export interface BucketDataSource {
-  readonly definition: BucketDataSourceDefinition;
-
   /**
    * 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.
@@ -126,7 +130,6 @@ export interface BucketDataSource {
 }
 
 export interface BucketParameterLookupSource {
-  readonly definition: BucketParameterLookupSourceDefinition;
   /**
    * Given a row in a source table that affects sync parameters, returns a structure to index which buckets rows should
    * be associated with.
@@ -138,8 +141,6 @@ export interface BucketParameterLookupSource {
 }
 
 export interface BucketParameterQuerierSource {
-  readonly definition: BucketParameterQuerierSourceDefinition;
-
   /**
    * Reports {@link BucketParameterQuerier}s resolving buckets that a specific stream request should have access to.
    *
@@ -149,9 +150,70 @@ export interface BucketParameterQuerierSource {
   pushBucketParameterQueriers(result: PendingQueriers, options: GetQuerierOptions): void;
 }
 
+export interface DebugMergedSource
+  extends BucketDataSource,
+    BucketParameterLookupSource,
+    BucketParameterQuerierSource {}
+
 export enum BucketSourceType {
   SYNC_RULE,
   SYNC_STREAM
 }
 
 export type ResultSetDescription = { name: string; columns: ColumnDefinition[] };
+
+export function mergeDataSources(sources: BucketDataSource[]): BucketDataSource {
+  return {
+    evaluateRow(options: EvaluateRowOptions): EvaluationResult[] {
+      let results: EvaluationResult[] = [];
+      for (let source of sources) {
+        results.push(...source.evaluateRow(options));
+      }
+      return results;
+    }
+  };
+}
+
+export function mergeParameterLookupSources(sources: BucketParameterLookupSource[]): BucketParameterLookupSource {
+  return {
+    evaluateParameterRow(sourceTable: SourceTableInterface, row: SqliteRow): EvaluatedParametersResult[] {
+      let results: EvaluatedParametersResult[] = [];
+      for (let source of sources) {
+        results.push(...source.evaluateParameterRow(sourceTable, row));
+      }
+      return results;
+    }
+  };
+}
+
+export function mergeParameterQuerierSources(sources: BucketParameterQuerierSource[]): BucketParameterQuerierSource {
+  return {
+    pushBucketParameterQueriers(result: PendingQueriers, options: GetQuerierOptions): void {
+      for (let source of sources) {
+        source.pushBucketParameterQueriers(result, options);
+      }
+    }
+  };
+}
+
+/**
+ * For production purposes, we typically need to operate on the different sources separately. However, for debugging,
+ * it is useful to have a single merged source that can evaluate everything.
+ */
+export function debugHydratedMergedSource(bucketSource: BucketSource, params?: CreateSourceParams): DebugMergedSource {
+  const resolvedParams = params ?? { bucketIdTransformer: (id: string) => id };
+  const dataSource = mergeDataSources(
+    bucketSource.dataSources.map((source) => source.createDataSource(resolvedParams))
+  );
+  const parameterLookupSource = mergeParameterLookupSources(
+    bucketSource.parameterLookupSources.map((source) => source.createParameterLookupSource(resolvedParams))
+  );
+  const parameterQuerierSource = mergeParameterQuerierSources(
+    bucketSource.parameterQuerierSources.map((source) => source.createParameterQuerierSource(resolvedParams))
+  );
+  return {
+    evaluateParameterRow: parameterLookupSource.evaluateParameterRow.bind(parameterLookupSource),
+    evaluateRow: dataSource.evaluateRow.bind(dataSource),
+    pushBucketParameterQueriers: parameterQuerierSource.pushBucketParameterQueriers.bind(parameterQuerierSource)
+  };
+}

packages/sync-rules/src/SqlBucketDescriptor.ts

@@ -3,7 +3,10 @@ import {
   BucketDataSourceDefinition,
   BucketSource,
   BucketSourceType,
-  CreateSourceParams
+  CreateSourceParams,
+  DebugMergedSource,
+  mergeParameterLookupSources,
+  mergeParameterQuerierSources
 } from './BucketSource.js';
 import { ColumnDefinition } from './ExpressionType.js';
 import { IdSequence } from './IdSequence.js';
@@ -106,7 +109,6 @@ export class SqlBucketDescriptor implements BucketDataSourceDefinition, BucketSo
 
   createDataSource(params: CreateSourceParams): BucketDataSource {
     return {
-      definition: this,
       evaluateRow: (options) => {
         let results: EvaluationResult[] = [];
         for (let query of this.dataQueries) {

packages/sync-rules/src/SqlParameterQuery.ts

@@ -326,8 +326,6 @@ export class SqlParameterQuery
 
   createParameterQuerierSource(params: CreateSourceParams): BucketParameterQuerierSource {
     return {
-      definition: this,
-
       pushBucketParameterQueriers: (result: PendingQueriers, options: GetQuerierOptions) => {
         const q = this.getBucketParameterQuerier(options.globalParameters, ['default'], params.bucketIdTransformer);
         result.queriers.push(q);
@@ -337,8 +335,6 @@ export class SqlParameterQuery
 
   createParameterLookupSource(params: CreateSourceParams): BucketParameterLookupSource {
     return {
-      definition: this,
-
       evaluateParameterRow: (sourceTable: SourceTableInterface, row: SqliteRow): EvaluatedParametersResult[] => {
         if (this.tableSyncsParameters(sourceTable)) {
           return this.evaluateParameterRow(row);

packages/sync-rules/src/SqlSyncRules.ts

@@ -399,10 +399,8 @@ export class SqlSyncRules {
       : (id: string) => id;
     return new HydratedSyncRules({
       definition: this,
+      createParams: { bucketIdTransformer },
       bucketDataSources: this.bucketDataSources.map((d) => d.createDataSource({ bucketIdTransformer })),
-      bucketParameterQuerierSources: this.bucketParameterQuerierSources.map((d) =>
-        d.createParameterQuerierSource({ bucketIdTransformer })
-      ),
       bucketParameterLookupSources: this.bucketParameterLookupSources.map((d) =>
         d.createParameterLookupSource({ bucketIdTransformer })
       ),

packages/sync-rules/src/StaticSqlParameterQuery.ts

@@ -180,8 +180,6 @@ export class StaticSqlParameterQuery implements BucketParameterQuerierSourceDefi
 
   createParameterQuerierSource(params: CreateSourceParams): BucketParameterQuerierSource {
     return {
-      definition: this,
-
       pushBucketParameterQueriers: (result: PendingQueriers, options: GetQuerierOptions) => {
         const staticBuckets = this.getStaticBucketDescriptions(
           options.globalParameters,

packages/sync-rules/src/SyncRules.ts

@@ -3,6 +3,7 @@ import {
   BucketParameterLookupSource,
   BucketParameterQuerierSource,
   BucketParameterQuerierSourceDefinition,
+  CreateSourceParams,
   HydratedBucketSource
 } from './BucketSource.js';
 import {
@@ -33,7 +34,6 @@ import { EvaluatedParametersResult, EvaluateRowOptions, EvaluationResult, Sqlite
 export class HydratedSyncRules {
   bucketSources: HydratedBucketSource[] = [];
   bucketDataSources: BucketDataSource[];
-  bucketParameterQuerierSources: BucketParameterQuerierSource[];
   bucketParameterLookupSources: BucketParameterLookupSource[];
 
   eventDescriptors: SqlEventDescriptor[] = [];
@@ -43,14 +43,13 @@ export class HydratedSyncRules {
 
   constructor(params: {
     definition: SqlSyncRules;
+    createParams: CreateSourceParams;
     bucketDataSources: BucketDataSource[];
-    bucketParameterQuerierSources: BucketParameterQuerierSource[];
     bucketParameterLookupSources: BucketParameterLookupSource[];
     eventDescriptors?: SqlEventDescriptor[];
     compatibility?: CompatibilityContext;
   }) {
     this.bucketDataSources = params.bucketDataSources;
-    this.bucketParameterQuerierSources = params.bucketParameterQuerierSources;
     this.bucketParameterLookupSources = params.bucketParameterLookupSources;
     this.definition = params.definition;
 
@@ -61,21 +60,13 @@ export class HydratedSyncRules {
       this.compatibility = params.compatibility;
     }
 
-    let querierMap = new Map<BucketParameterQuerierSourceDefinition, HydratedBucketSource>();
     for (let definition of this.definition.bucketSources) {
       const hydratedBucketSource: HydratedBucketSource = { definition: definition, parameterQuerierSources: [] };
       this.bucketSources.push(hydratedBucketSource);
       for (let querier of definition.parameterQuerierSources) {
-        querierMap.set(querier, hydratedBucketSource);
+        hydratedBucketSource.parameterQuerierSources.push(querier.createParameterQuerierSource(params.createParams));
       }
     }
-    for (let querier of params.bucketParameterQuerierSources) {
-      const bucketSource = querierMap.get(querier.definition);
-      if (bucketSource == null) {
-        throw new Error('Cannot find BucketSource for BucketParameterQuerierSource');
-      }
-      bucketSource.parameterQuerierSources.push(querier);
-    }
   }
 
   // These methods do not depend on hydration, so we can just forward them to the definition.

packages/sync-rules/src/TableValuedFunctionSqlParameterQuery.ts

@@ -225,8 +225,6 @@ export class TableValuedFunctionSqlParameterQuery implements BucketParameterQuer
 
   createParameterQuerierSource(params: CreateSourceParams): BucketParameterQuerierSource {
     return {
-      definition: this,
-
       pushBucketParameterQueriers: (result: PendingQueriers, options: GetQuerierOptions) => {
         const staticBuckets = this.getStaticBucketDescriptions(
           options.globalParameters,

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

@@ -100,14 +100,13 @@ class SyncStreamCompiler {
     let filter = this.whereClauseToFilters(tools, query.where);
     filter = filter.toDisjunctiveNormalForm(tools);
 
+    const variants = filter.isValid(tools) ? filter.compileVariants(this.descriptorName) : [];
     const stream = new SyncStream(
       this.descriptorName,
-      new BaseSqlDataQuery(this.compileDataQuery(tools, query, alias, sourceTable))
+      new BaseSqlDataQuery(this.compileDataQuery(tools, query, alias, sourceTable)),
+      variants
     );
     stream.subscribedToByDefault = this.options.auto_subscribe ?? false;
-    if (filter.isValid(tools)) {
-      stream.variants = filter.compileVariants(this.descriptorName);
-    }
 
     this.errors.push(...tools.errors);
     if (this.parameterDetector.usesStreamParameters && stream.subscribedToByDefault) {