Initial split of data vs parameter sources.

Author: rkistner

packages/sync-rules/src/BucketSource.ts

@@ -15,7 +15,55 @@ import { EvaluatedParametersResult, EvaluateRowOptions, EvaluationResult, Source
  * 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 {
+export interface BucketDataSource {
+  readonly name: string;
+  readonly type: BucketSourceType;
+
+  readonly subscribedToByDefault: boolean;
+
+  /**
+   * 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 and errors into.
+   * @param options Options, including parameters that may affect the buckets loaded by this source.
+   */
+  pushBucketParameterQueriers(result: PendingQueriers, options: GetQuerierOptions): void;
+
+  getSourceTables(): Set<TablePattern>;
+
+  /** 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;
+}
+
+/**
+ * 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 BucketParameterSource {
   readonly name: string;
   readonly type: BucketSourceType;
 
@@ -30,12 +78,6 @@ export interface BucketSource {
    */
   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.
    *
@@ -57,17 +99,6 @@ export interface BucketSource {
   /** 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;

packages/sync-rules/src/SqlBucketDescriptor.ts

@@ -1,6 +1,6 @@
 import { BucketInclusionReason, ResolvedBucket } from './BucketDescription.js';
 import { BucketParameterQuerier, mergeBucketParameterQueriers, PendingQueriers } from './BucketParameterQuerier.js';
-import { BucketSource, BucketSourceType, ResultSetDescription } from './BucketSource.js';
+import { BucketDataSource, BucketParameterSource, BucketSourceType, ResultSetDescription } from './BucketSource.js';
 import { ColumnDefinition } from './ExpressionType.js';
 import { IdSequence } from './IdSequence.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
@@ -32,7 +32,7 @@ export interface QueryParseResult {
   errors: SqlRuleError[];
 }
 
-export class SqlBucketDescriptor implements BucketSource {
+export class SqlBucketDescriptor implements BucketDataSource, BucketParameterSource {
   name: string;
   bucketParameters?: string[];
 

packages/sync-rules/src/SqlSyncRules.ts

@@ -1,11 +1,14 @@
 import { isScalar, LineCounter, parseDocument, Scalar, YAMLMap, YAMLSeq } from 'yaml';
 import { isValidPriority } from './BucketDescription.js';
 import { BucketParameterQuerier, mergeBucketParameterQueriers, QuerierError } from './BucketParameterQuerier.js';
+import { BucketDataSource, BucketParameterSource } from './BucketSource.js';
+import { CompatibilityContext, CompatibilityEdition, CompatibilityOption } from './compatibility.js';
 import { SqlRuleError, SyncRulesErrors, YamlError } from './errors.js';
 import { SqlEventDescriptor } from './events/SqlEventDescriptor.js';
 import { validateSyncRulesSchema } from './json_schema.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
 import { QueryParseResult, SqlBucketDescriptor } from './SqlBucketDescriptor.js';
+import { syncStreamFromSql } from './streams/from_sql.js';
 import { TablePattern } from './TablePattern.js';
 import {
   BucketIdTransformer,
@@ -28,9 +31,6 @@ import {
   StreamParseOptions,
   SyncRules
 } from './types.js';
-import { BucketSource } from './BucketSource.js';
-import { syncStreamFromSql } from './streams/from_sql.js';
-import { CompatibilityContext, CompatibilityEdition, CompatibilityOption } from './compatibility.js';
 import { applyRowContext } from './utils.js';
 
 const ACCEPT_POTENTIALLY_DANGEROUS_QUERIES = Symbol('ACCEPT_POTENTIALLY_DANGEROUS_QUERIES');
@@ -94,7 +94,9 @@ export interface GetBucketParameterQuerierResult {
 }
 
 export class SqlSyncRules implements SyncRules {
-  bucketSources: BucketSource[] = [];
+  bucketDataSources: BucketDataSource[] = [];
+  bucketParameterSources: BucketParameterSource[] = [];
+
   eventDescriptors: SqlEventDescriptor[] = [];
   compatibility: CompatibilityContext = CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY;
 
@@ -245,7 +247,8 @@ export class SqlSyncRules implements SyncRules {
           return descriptor.addDataQuery(q, queryOptions, compatibility);
         });
       }
-      rules.bucketSources.push(descriptor);
+      rules.bucketDataSources.push(descriptor);
+      rules.bucketParameterSources.push(descriptor);
     }
 
     for (const entry of streamMap?.items ?? []) {
@@ -270,7 +273,8 @@ export class SqlSyncRules implements SyncRules {
       if (data instanceof Scalar) {
         rules.withScalar(data, (q) => {
           const [parsed, errors] = syncStreamFromSql(key, q, queryOptions);
-          rules.bucketSources.push(parsed);
+          rules.bucketDataSources.push(parsed);
+          rules.bucketParameterSources.push(parsed);
           return {
             parsed: true,
             errors
@@ -412,7 +416,7 @@ export class SqlSyncRules implements SyncRules {
         };
 
     let rawResults: EvaluationResult[] = [];
-    for (let source of this.bucketSources) {
+    for (let source of this.bucketDataSources) {
       rawResults.push(...source.evaluateRow(resolvedOptions));
     }
 
@@ -438,7 +442,7 @@ export class SqlSyncRules implements SyncRules {
     row: SqliteRow
   ): { results: EvaluatedParameters[]; errors: EvaluationError[] } {
     let rawResults: EvaluatedParametersResult[] = [];
-    for (let source of this.bucketSources) {
+    for (let source of this.bucketParameterSources) {
       rawResults.push(...source.evaluateParameterRow(table, row));
     }
 
@@ -460,7 +464,7 @@ export class SqlSyncRules implements SyncRules {
     const errors: QuerierError[] = [];
     const pending = { queriers, errors };
 
-    for (const source of this.bucketSources) {
+    for (const source of this.bucketParameterSources) {
       if (
         (source.subscribedToByDefault && resolvedOptions.hasDefaultStreams) ||
         source.name in resolvedOptions.streams
@@ -474,12 +478,18 @@ export class SqlSyncRules implements SyncRules {
   }
 
   hasDynamicBucketQueries() {
-    return this.bucketSources.some((s) => s.hasDynamicBucketQueries());
+    return this.bucketParameterSources.some((s) => s.hasDynamicBucketQueries());
   }
 
   getSourceTables(): TablePattern[] {
     const sourceTables = new Map<String, TablePattern>();
-    for (const bucket of this.bucketSources) {
+    for (const bucket of this.bucketDataSources) {
+      for (const r of bucket.getSourceTables()) {
+        const key = `${r.connectionTag}.${r.schema}.${r.tablePattern}`;
+        sourceTables.set(key, r);
+      }
+    }
+    for (const bucket of this.bucketParameterSources) {
       for (const r of bucket.getSourceTables()) {
         const key = `${r.connectionTag}.${r.schema}.${r.tablePattern}`;
         sourceTables.set(key, r);
@@ -513,16 +523,19 @@ export class SqlSyncRules implements SyncRules {
   }
 
   tableSyncsData(table: SourceTableInterface): boolean {
-    return this.bucketSources.some((b) => b.tableSyncsData(table));
+    return this.bucketDataSources.some((b) => b.tableSyncsData(table));
   }
 
   tableSyncsParameters(table: SourceTableInterface): boolean {
-    return this.bucketSources.some((b) => b.tableSyncsParameters(table));
+    return this.bucketParameterSources.some((b) => b.tableSyncsParameters(table));
   }
 
   debugGetOutputTables() {
     let result: Record<string, any[]> = {};
-    for (let bucket of this.bucketSources) {
+    for (let bucket of this.bucketDataSources) {
+      bucket.debugWriteOutputTables(result);
+    }
+    for (let bucket of this.bucketParameterSources) {
       bucket.debugWriteOutputTables(result);
     }
     return result;

packages/sync-rules/src/schema-generators/SchemaGenerator.ts

@@ -10,7 +10,7 @@ export abstract class SchemaGenerator {
   protected getAllTables(source: SqlSyncRules, schema: SourceSchema) {
     let tables: Record<string, Record<string, ColumnDefinition>> = {};
 
-    for (let descriptor of source.bucketSources) {
+    for (let descriptor of source.bucketDataSources) {
       descriptor.resolveResultSets(schema, tables);
     }
 

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

@@ -1,7 +1,7 @@
 import { BaseSqlDataQuery } from '../BaseSqlDataQuery.js';
 import { BucketInclusionReason, BucketPriority, DEFAULT_BUCKET_PRIORITY } from '../BucketDescription.js';
 import { BucketParameterQuerier, PendingQueriers } from '../BucketParameterQuerier.js';
-import { BucketSource, BucketSourceType, ResultSetDescription } from '../BucketSource.js';
+import { BucketDataSource, BucketParameterSource, BucketSourceType, ResultSetDescription } from '../BucketSource.js';
 import { ColumnDefinition } from '../ExpressionType.js';
 import { SourceTableInterface } from '../SourceTableInterface.js';
 import { GetQuerierOptions, RequestedStream } from '../SqlSyncRules.js';
@@ -18,7 +18,7 @@ import {
 } from '../types.js';
 import { StreamVariant } from './variant.js';
 
-export class SyncStream implements BucketSource {
+export class SyncStream implements BucketDataSource, BucketParameterSource {
   name: string;
   subscribedToByDefault: boolean;
   priority: BucketPriority;

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

@@ -17,7 +17,8 @@ describe('sync rules', () => {
 
   test('parse empty sync rules', () => {
     const rules = SqlSyncRules.fromYaml('bucket_definitions: {}', PARSE_OPTIONS);
-    expect(rules.bucketSources).toEqual([]);
+    expect(rules.bucketParameterSources).toEqual([]);
+    expect(rules.bucketDataSources).toEqual([]);
   });
 
   test('parse global sync rules', () => {
@@ -30,7 +31,7 @@ bucket_definitions:
     `,
       PARSE_OPTIONS
     );
-    const bucket = rules.bucketSources[0] as SqlBucketDescriptor;
+    const bucket = rules.bucketDataSources[0] as SqlBucketDescriptor;
     expect(bucket.name).toEqual('mybucket');
     expect(bucket.bucketParameters).toEqual([]);
     const dataQuery = bucket.dataQueries[0];
@@ -70,7 +71,7 @@ bucket_definitions:
     `,
       PARSE_OPTIONS
     );
-    const bucket = rules.bucketSources[0] as SqlBucketDescriptor;
+    const bucket = rules.bucketParameterSources[0] as SqlBucketDescriptor;
     expect(bucket.bucketParameters).toEqual([]);
     const param_query = bucket.globalParameterQueries[0];
 
@@ -102,7 +103,7 @@ bucket_definitions:
     `,
       PARSE_OPTIONS
     );
-    const bucket = rules.bucketSources[0] as SqlBucketDescriptor;
+    const bucket = rules.bucketParameterSources[0] as SqlBucketDescriptor;
     expect(bucket.bucketParameters).toEqual([]);
     const param_query = bucket.parameterQueries[0];
     expect(param_query.bucketParameters).toEqual([]);
@@ -126,9 +127,10 @@ bucket_definitions:
     `,
       PARSE_OPTIONS
     );
-    const bucket = rules.bucketSources[0] as SqlBucketDescriptor;
-    expect(bucket.bucketParameters).toEqual(['user_id', 'device_id']);
-    const param_query = bucket.globalParameterQueries[0];
+    const bucketParameters = rules.bucketParameterSources[0] as SqlBucketDescriptor;
+    const bucketData = rules.bucketDataSources[0] as SqlBucketDescriptor;
+    expect(bucketParameters.bucketParameters).toEqual(['user_id', 'device_id']);
+    const param_query = bucketParameters.globalParameterQueries[0];
     expect(param_query.bucketParameters).toEqual(['user_id', 'device_id']);
     expect(
       rules.getBucketParameterQuerier(normalizeQuerierOptions({ user_id: 'user1' }, { device_id: 'device1' })).querier
@@ -137,7 +139,7 @@ bucket_definitions:
       { bucket: 'mybucket["user1","device1"]', definition: 'mybucket', inclusion_reasons: ['default'], priority: 3 }
     ]);
 
-    const data_query = bucket.dataQueries[0];
+    const data_query = bucketData.dataQueries[0];
     expect(data_query.bucketParameters).toEqual(['user_id', 'device_id']);
     expect(
       rules.evaluateRow({
@@ -176,15 +178,16 @@ bucket_definitions:
     `,
       PARSE_OPTIONS
     );
-    const bucket = rules.bucketSources[0] as SqlBucketDescriptor;
-    expect(bucket.bucketParameters).toEqual(['user_id']);
-    const param_query = bucket.globalParameterQueries[0];
+    const bucketParameters = rules.bucketParameterSources[0] as SqlBucketDescriptor;
+    const bucketData = rules.bucketDataSources[0] as SqlBucketDescriptor;
+    expect(bucketParameters.bucketParameters).toEqual(['user_id']);
+    const param_query = bucketParameters.globalParameterQueries[0];
     expect(param_query.bucketParameters).toEqual(['user_id']);
     expect(
       rules.getBucketParameterQuerier(normalizeQuerierOptions({ user_id: 'user1' })).querier.staticBuckets
     ).toEqual([{ bucket: 'mybucket["user1"]', definition: 'mybucket', inclusion_reasons: ['default'], priority: 3 }]);
 
-    const data_query = bucket.dataQueries[0];
+    const data_query = bucketData.dataQueries[0];
     expect(data_query.bucketParameters).toEqual(['user_id']);
     expect(
       rules.evaluateRow({
@@ -322,8 +325,8 @@ bucket_definitions:
     `,
       PARSE_OPTIONS
     );
-    const bucket = rules.bucketSources[0] as SqlBucketDescriptor;
-    expect(bucket.bucketParameters).toEqual(['user_id']);
+    const bucketParameters = rules.bucketParameterSources[0] as SqlBucketDescriptor;
+    expect(bucketParameters.bucketParameters).toEqual(['user_id']);
     expect(rules.getBucketParameterQuerier(normalizeQuerierOptions({ user_id: 'user1' })).querier).toMatchObject({
       staticBuckets: [{ bucket: 'mybucket["USER1"]', priority: 3 }],
       hasDynamicBuckets: false
@@ -360,8 +363,8 @@ bucket_definitions:
     `,
       PARSE_OPTIONS
     );
-    const bucket = rules.bucketSources[0] as SqlBucketDescriptor;
-    expect(bucket.bucketParameters).toEqual(['user_id']);
+    const bucketParameters = rules.bucketParameterSources[0] as SqlBucketDescriptor;
+    expect(bucketParameters.bucketParameters).toEqual(['user_id']);
     expect(rules.getBucketParameterQuerier(normalizeQuerierOptions({ user_id: 'user1' })).querier).toMatchObject({
       staticBuckets: [{ bucket: 'mybucket["USER1"]', priority: 3 }],
       hasDynamicBuckets: false
@@ -956,8 +959,8 @@ bucket_definitions:
     `,
       PARSE_OPTIONS
     );
-    const bucket = rules.bucketSources[0] as SqlBucketDescriptor;
-    expect(bucket.bucketParameters).toEqual(['user_id']);
+    const bucketParameters = rules.bucketParameterSources[0] as SqlBucketDescriptor;
+    expect(bucketParameters.bucketParameters).toEqual(['user_id']);
     expect(rules.hasDynamicBucketQueries()).toBe(true);
 
     expect(rules.getBucketParameterQuerier(normalizeQuerierOptions({ user_id: 'user1' })).querier).toMatchObject({