Support multiple data sources.

Author: rkistner

packages/sync-rules/src/BucketSource.ts

@@ -25,8 +25,12 @@ export interface BucketSource {
    * BucketDataSource describing the data in this bucket/stream definition.
    *
    * The same data source could in theory be present in multiple stream definitions.
+   *
+   * Sources must _only_ be split into multiple ones if they will result in different buckets being created.
+   * Specifically, bucket definitions would always have a single data source, while stream definitions may have
+   * one per variant.
    */
-  readonly dataSource: BucketDataSourceDefinition;
+  readonly dataSources: BucketDataSourceDefinition[];
 
   /**
    * BucketParameterQuerierSource describing the parameter queries / stream subqueries in this bucket/stream definition.

packages/sync-rules/src/SqlBucketDescriptor.ts

@@ -54,8 +54,8 @@ export class SqlBucketDescriptor implements BucketDataSourceDefinition, BucketSo
     return this.bucketParametersInternal ?? [];
   }
 
-  get dataSource() {
-    return this;
+  get dataSources() {
+    return [this];
   }
 
   get parameterLookupSources() {

packages/sync-rules/src/SqlSyncRules.ts

@@ -240,7 +240,7 @@ export class SqlSyncRules {
       }
 
       rules.bucketSources.push(descriptor);
-      rules.bucketDataSources.push(descriptor.dataSource);
+      rules.bucketDataSources.push(...descriptor.dataSources);
       rules.bucketParameterLookupSources.push(...descriptor.parameterLookupSources);
       rules.bucketParameterQuerierSources.push(...descriptor.parameterQuerierSources);
     }
@@ -268,7 +268,7 @@ export class SqlSyncRules {
         rules.withScalar(data, (q) => {
           const [parsed, errors] = syncStreamFromSql(key, q, queryOptions);
           rules.bucketSources.push(parsed);
-          rules.bucketDataSources.push(parsed.dataSource);
+          rules.bucketDataSources.push(...parsed.dataSources);
           rules.bucketParameterLookupSources.push(...parsed.parameterLookupSources);
           rules.bucketParameterQuerierSources.push(...parsed.parameterQuerierSources);
           return {

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

@@ -34,7 +34,7 @@ export class SyncStream implements BucketSource {
   variants: StreamVariant[];
   data: BaseSqlDataQuery;
 
-  public readonly dataSource: BucketDataSourceDefinition;
+  public readonly dataSources: BucketDataSourceDefinition[];
   public readonly parameterLookupSources: BucketParameterLookupSourceDefinition[];
   public readonly parameterQuerierSources: BucketParameterQuerierSourceDefinition[];
 
@@ -45,7 +45,7 @@ export class SyncStream implements BucketSource {
     this.variants = [];
     this.data = data;
 
-    this.dataSource = new SyncStreamDataSource(this, data);
+    this.dataSources = [new SyncStreamDataSource(this, data)];
     this.parameterQuerierSources = [new SyncStreamParameterQuerierSource(this)];
     this.parameterLookupSources = [new SyncStreamParameterLookupSource(this)];
   }

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

@@ -38,7 +38,7 @@ describe('streams', () => {
     expect(desc.variants).toHaveLength(1);
     expect(evaluateBucketIds(desc, COMMENTS, { id: 'foo' })).toStrictEqual(['1#stream|0[]']);
     expect(
-      desc.dataSource
+      desc.dataSources[0]
         .createDataSource({ bucketIdTransformer })
         .evaluateRow({ sourceTable: USERS, record: { id: 'foo' } })
     ).toHaveLength(0);
@@ -747,7 +747,7 @@ describe('streams', () => {
       );
       const row = { id: 'id', account_id: 'account_id' };
 
-      expect(stream.dataSource.tableSyncsData(accountMember)).toBeTruthy();
+      expect(stream.dataSources[0].tableSyncsData(accountMember)).toBeTruthy();
       expect(stream.parameterLookupSources[0].tableSyncsParameters(accountMember)).toBeTruthy();
 
       // Ensure lookup steps work.
@@ -779,7 +779,7 @@ describe('streams', () => {
 
       // And that the data alias is respected for generated schemas.
       const outputSchema = {};
-      stream.dataSource.resolveResultSets(schema, outputSchema);
+      stream.dataSources[0].resolveResultSets(schema, outputSchema);
       expect(Object.keys(outputSchema)).toStrictEqual(['outer']);
     });
 
@@ -937,16 +937,20 @@ const options: StreamParseOptions = {
 const bucketIdTransformer = SqlSyncRules.versionedBucketIdTransformer('1');
 
 function evaluateBucketIds(stream: SyncStream, sourceTable: SourceTableInterface, record: SqliteRow) {
-  return stream.dataSource
-    .createDataSource({ bucketIdTransformer })
-    .evaluateRow({ sourceTable, record })
-    .map((r) => {
-      if ('error' in r) {
-        throw new Error(`Unexpected error evaluating row: ${r.error}`);
-      }
+  return stream.dataSources
+    .map((s) =>
+      s
+        .createDataSource({ bucketIdTransformer })
+        .evaluateRow({ sourceTable, record })
+        .map((r) => {
+          if ('error' in r) {
+            throw new Error(`Unexpected error evaluating row: ${r.error}`);
+          }
 
-      return r.bucket;
-    });
+          return r.bucket;
+        })
+    )
+    .flat();
 }
 
 async function createQueriers(

packages/sync-rules/test/src/sync_rules.test.ts

@@ -108,7 +108,6 @@ bucket_definitions:
     );
     const hydrated = rules.hydrate({ bucketIdTransformer });
     const parameterLookupSource = rules.bucketParameterLookupSources[0];
-    expect(parameterLookupSource.bucketParameters).toEqual([]);
     expect(hydrated.evaluateParameterRow(USERS, { id: 'user1', is_admin: 1 })).toEqual([
       {
         bucketParameters: [{}],