Remove need for hydration on BucketDataSource.

Author: rkistner

packages/sync-rules/src/BaseSqlDataQuery.ts

@@ -6,9 +6,10 @@ import { AvailableTable, SqlTools } from './sql_filters.js';
 import { castAsText } from './sql_functions.js';
 import { TablePattern } from './TablePattern.js';
 import {
-  EvaluationResult,
   QueryParameters,
   QuerySchema,
+  SourceEvaluatedRow,
+  SourceEvaluationResult,
   SourceSchema,
   SourceSchemaTable,
   SqliteJsonRow,
@@ -24,7 +25,7 @@ export interface RowValueExtractor {
 export interface EvaluateRowOptions {
   table: SourceTableInterface;
   row: SqliteRow;
-  bucketIds: (params: QueryParameters) => string[];
+  serializedBucketParameters: (params: QueryParameters) => string[];
 }
 
 export interface BaseSqlDataQueryOptions {
@@ -169,13 +170,14 @@ export class BaseSqlDataQuery {
     }
   }
 
-  evaluateRowWithOptions(options: EvaluateRowOptions): EvaluationResult[] {
+  evaluateRowWithOptions(options: EvaluateRowOptions): SourceEvaluationResult[] {
     try {
-      const { table, row, bucketIds } = options;
+      const { table, row, serializedBucketParameters } = options;
 
       const tables = { [this.table.nameInSchema]: this.addSpecialParameters(table, row) };
-      const resolvedBucketIds = bucketIds(tables);
-      if (resolvedBucketIds.length == 0) {
+      // Array of _serialized_ parameters, one per output result.
+      const resolvedBucketParameters = serializedBucketParameters(tables);
+      if (resolvedBucketParameters.length == 0) {
         // Short-circuit: No need to transform the row if there are no matching buckets.
         return [];
       }
@@ -193,13 +195,13 @@ export class BaseSqlDataQuery {
       }
       const outputTable = this.getOutputName(table.name);
 
-      return resolvedBucketIds.map((bucketId) => {
+      return resolvedBucketParameters.map((serializedBucketParameters) => {
         return {
-          bucket: bucketId,
+          serializedBucketParameters,
           table: outputTable,
           id: id,
           data
-        } as EvaluationResult;
+        } satisfies SourceEvaluatedRow;
       });
     } catch (e) {
       return [{ error: e.message ?? `Evaluating data query failed` }];

packages/sync-rules/src/BucketSource.ts

@@ -4,7 +4,17 @@ import { DEFAULT_HYDRATION_STATE, HydrationState, ParameterLookupScope } from '.
 import { SourceTableInterface } from './SourceTableInterface.js';
 import { GetQuerierOptions } from './SqlSyncRules.js';
 import { TablePattern } from './TablePattern.js';
-import { EvaluatedParametersResult, EvaluateRowOptions, EvaluationResult, SourceSchema, SqliteRow } from './types.js';
+import {
+  EvaluatedParametersResult,
+  EvaluatedRow,
+  EvaluateRowOptions,
+  EvaluationResult,
+  isEvaluationError,
+  SourceEvaluationResult,
+  SourceSchema,
+  SqliteRow
+} from './types.js';
+import { buildBucketName } from './utils.js';
 
 export interface CreateSourceParams {
   hydrationState: HydrationState;
@@ -30,7 +40,7 @@ export interface BucketSource {
    * Specifically, bucket definitions would always have a single data source, while stream definitions may have
    * one per variant.
    */
-  readonly dataSources: BucketDataSourceDefinition[];
+  readonly dataSources: BucketDataSource[];
 
   /**
    * BucketParameterQuerierSource describing the parameter queries / stream subqueries in this bucket/stream definition.
@@ -57,8 +67,11 @@ export interface HydratedBucketSource {
 
 /**
  * Encodes a static definition of a bucket source, as parsed from sync rules or stream definitions.
+ *
+ * This does not require any "hydration" itself: All results are independent of bucket names.
+ * The higher-level HydratedSyncRules will use a HydrationState to generate bucket names.
  */
-export interface BucketDataSourceDefinition {
+export interface BucketDataSource {
   /**
    * Bucket prefix if no transformations are defined.
    *
@@ -70,10 +83,16 @@ export interface BucketDataSourceDefinition {
    * For debug use only.
    */
   readonly bucketParameters: string[];
+
   getSourceTables(): Set<TablePattern>;
-  createDataSource(params: CreateSourceParams): BucketDataSource;
   tableSyncsData(table: SourceTableInterface): 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): SourceEvaluationResult[];
+
   /**
    * Given a static schema, infer all logical tables and associated columns that appear in buckets defined by this
    * source.
@@ -121,29 +140,11 @@ export interface BucketParameterQuerierSourceDefinition {
    *
    * Note that queriers do not persist data themselves; they only resolve which buckets to load based on request parameters.
    */
-  readonly querierDataSource: BucketDataSourceDefinition;
+  readonly querierDataSource: BucketDataSource;
 
   createParameterQuerierSource(params: CreateSourceParams): BucketParameterQuerierSource;
 }
 
-/**
- * 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 BucketDataSource {
-  /**
-   * 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[];
-}
-
 export interface BucketParameterLookupSource {
   /**
    * Given a row in a source table that affects sync parameters, returns a structure to index which buckets rows should
@@ -165,10 +166,9 @@ export interface BucketParameterQuerierSource {
   pushBucketParameterQueriers(result: PendingQueriers, options: GetQuerierOptions): void;
 }
 
-export interface DebugMergedSource
-  extends BucketDataSource,
-    BucketParameterLookupSource,
-    BucketParameterQuerierSource {}
+export interface DebugMergedSource extends BucketParameterLookupSource, BucketParameterQuerierSource {
+  evaluateRow(options: EvaluateRowOptions): EvaluationResult[];
+}
 
 export enum BucketSourceType {
   SYNC_RULE,
@@ -177,14 +177,31 @@ export enum BucketSourceType {
 
 export type ResultSetDescription = { name: string; columns: ColumnDefinition[] };
 
-export function mergeDataSources(sources: BucketDataSource[]): BucketDataSource {
+export function hydrateEvaluateRow(
+  hydrationState: HydrationState,
+  source: BucketDataSource
+): (options: EvaluateRowOptions) => EvaluationResult[] {
+  const scope = hydrationState.getBucketSourceScope(source);
+  return (options: EvaluateRowOptions): EvaluationResult[] => {
+    return source.evaluateRow(options).map((result) => {
+      if (isEvaluationError(result)) {
+        return result;
+      }
+      return {
+        bucket: buildBucketName(scope, result.serializedBucketParameters),
+        id: result.id,
+        table: result.table,
+        data: result.data
+      } satisfies EvaluatedRow;
+    });
+  };
+}
+
+export function mergeDataSources(hydrationState: HydrationState, sources: BucketDataSource[]) {
+  const withScope = sources.map((source) => hydrateEvaluateRow(hydrationState, source));
   return {
     evaluateRow(options: EvaluateRowOptions): EvaluationResult[] {
-      let results: EvaluationResult[] = [];
-      for (let source of sources) {
-        results.push(...source.evaluateRow(options));
-      }
-      return results;
+      return withScope.flatMap((source) => source(options));
     }
   };
 }
@@ -218,9 +235,7 @@ export function mergeParameterQuerierSources(sources: BucketParameterQuerierSour
 export function debugHydratedMergedSource(bucketSource: BucketSource, params?: CreateSourceParams): DebugMergedSource {
   const hydrationState = params?.hydrationState ?? DEFAULT_HYDRATION_STATE;
   const resolvedParams = { hydrationState };
-  const dataSource = mergeDataSources(
-    bucketSource.dataSources.map((source) => source.createDataSource(resolvedParams))
-  );
+  const dataSource = mergeDataSources(hydrationState, bucketSource.dataSources);
   const parameterLookupSource = mergeParameterLookupSources(
     bucketSource.parameterLookupSources.map((source) => source.createParameterLookupSource(resolvedParams))
   );

packages/sync-rules/src/SyncRules.ts …ages/sync-rules/src/HydratedSyncRules.tspackages/sync-rules/src/SyncRules.ts renamed to packages/sync-rules/src/HydratedSyncRules.ts packages/sync-rules/src/SyncRules.ts renamed to packages/sync-rules/src/HydratedSyncRules.ts

@@ -1,11 +1,10 @@
 import {
   BucketDataSource,
   BucketParameterLookupSource,
-  BucketParameterQuerierSource,
-  BucketParameterQuerierSourceDefinition,
   CreateSourceParams,
   HydratedBucketSource
 } from './BucketSource.js';
+import { BucketDataScope } from './HydrationState.js';
 import {
   BucketParameterQuerier,
   CompatibilityContext,
@@ -35,6 +34,7 @@ export class HydratedSyncRules {
   bucketSources: HydratedBucketSource[] = [];
   bucketDataSources: BucketDataSource[];
   bucketParameterLookupSources: BucketParameterLookupSource[];
+  bucketSourceHydration: Map<BucketDataSource, BucketDataScope> = new Map();
 
   eventDescriptors: SqlEventDescriptor[] = [];
   compatibility: CompatibilityContext = CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY;
@@ -53,6 +53,13 @@ export class HydratedSyncRules {
     this.bucketParameterLookupSources = params.bucketParameterLookupSources;
     this.definition = params.definition;
 
+    const hydrationState = params.createParams.hydrationState;
+
+    for (let source of this.bucketDataSources) {
+      const state = hydrationState.getBucketSourceScope(source);
+      this.bucketSourceHydration.set(source, state);
+    }
+
     if (params.eventDescriptors) {
       this.eventDescriptors = params.eventDescriptors;
     }
@@ -107,7 +114,24 @@ export class HydratedSyncRules {
   evaluateRowWithErrors(options: EvaluateRowOptions): { results: EvaluatedRow[]; errors: EvaluationError[] } {
     let rawResults: EvaluationResult[] = [];
     for (let source of this.bucketDataSources) {
-      rawResults.push(...source.evaluateRow(options));
+      const sourceResults = source.evaluateRow(options);
+      if (sourceResults.length == 0) {
+        continue;
+      }
+      const bucketPrefix = this.bucketSourceHydration.get(source)!.bucketPrefix;
+      rawResults.push(
+        ...sourceResults.map((sourceRow) => {
+          if (isEvaluationError(sourceRow)) {
+            return sourceRow;
+          }
+          return {
+            bucket: bucketPrefix + sourceRow.serializedBucketParameters,
+            id: sourceRow.id,
+            table: sourceRow.table,
+            data: sourceRow.data
+          } satisfies EvaluatedRow;
+        })
+      );
     }
 
     const results = rawResults.filter(isEvaluatedRow) as EvaluatedRow[];

packages/sync-rules/src/HydrationState.ts

@@ -1,6 +1,6 @@
-import { BucketDataSourceDefinition, BucketParameterLookupSourceDefinition } from './BucketSource.js';
+import { BucketDataSource, BucketParameterLookupSourceDefinition } from './BucketSource.js';
 
-export interface BucketSourceState {
+export interface BucketDataScope {
   /** The prefix is the bucket name before the parameters. */
   bucketPrefix: string;
 }
@@ -18,13 +18,13 @@ export interface ParameterLookupScope {
  * both to re-use mappings across hydrations of different sync rule versions, or to generate new mappings.
  */
 export interface HydrationState<
-  T extends BucketSourceState = BucketSourceState,
+  T extends BucketDataScope = BucketDataScope,
   U extends ParameterLookupScope = ParameterLookupScope
 > {
   /**
    * Given a bucket data source definition, get the bucket prefix to use for it.
    */
-  getBucketSourceState(source: BucketDataSourceDefinition): T;
+  getBucketSourceScope(source: BucketDataSource): T;
 
   /**
    * Given a bucket parameter lookup definition, get the persistence name to use.
@@ -38,7 +38,7 @@ export interface HydrationState<
  * This is the legacy default behavior with no bucket versioning.
  */
 export const DEFAULT_HYDRATION_STATE: HydrationState = {
-  getBucketSourceState(source: BucketDataSourceDefinition) {
+  getBucketSourceScope(source: BucketDataSource) {
     return {
       bucketPrefix: source.defaultBucketPrefix
     };
@@ -62,7 +62,7 @@ export const DEFAULT_HYDRATION_STATE: HydrationState = {
  */
 export function versionedHydrationState(version: number): HydrationState {
   return {
-    getBucketSourceState(source: BucketDataSourceDefinition): BucketSourceState {
+    getBucketSourceScope(source: BucketDataSource): BucketDataScope {
       return {
         bucketPrefix: `${version}#${source.defaultBucketPrefix}`
       };

packages/sync-rules/src/SqlBucketDescriptor.ts

@@ -1,10 +1,4 @@
-import {
-  BucketDataSource,
-  BucketDataSourceDefinition,
-  BucketSource,
-  BucketSourceType,
-  CreateSourceParams
-} from './BucketSource.js';
+import { BucketDataSource, BucketSource, BucketSourceType } from './BucketSource.js';
 import { ColumnDefinition } from './ExpressionType.js';
 import { IdSequence } from './IdSequence.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
@@ -16,7 +10,7 @@ import { TablePattern } from './TablePattern.js';
 import { TableValuedFunctionSqlParameterQuery } from './TableValuedFunctionSqlParameterQuery.js';
 import { CompatibilityContext } from './compatibility.js';
 import { SqlRuleError } from './errors.js';
-import { EvaluationResult, QueryParseOptions, SourceSchema } from './types.js';
+import { EvaluateRowOptions, QueryParseOptions, SourceEvaluationResult, SourceSchema } from './types.js';
 
 export interface QueryParseResult {
   /**
@@ -142,7 +136,7 @@ export class SqlBucketDescriptor implements BucketSource {
   }
 }
 
-export class BucketDefinitionDataSource implements BucketDataSourceDefinition {
+export class BucketDefinitionDataSource implements BucketDataSource {
   constructor(private descriptor: SqlBucketDescriptor) {}
 
   /**
@@ -156,22 +150,16 @@ export class BucketDefinitionDataSource implements BucketDataSourceDefinition {
     return this.descriptor.name;
   }
 
-  createDataSource(params: CreateSourceParams): BucketDataSource {
-    const hydrationState = params.hydrationState;
-    const bucketPrefix = hydrationState.getBucketSourceState(this).bucketPrefix;
-    return {
-      evaluateRow: (options) => {
-        let results: EvaluationResult[] = [];
-        for (let query of this.descriptor.dataQueries) {
-          if (!query.applies(options.sourceTable)) {
-            continue;
-          }
-
-          results.push(...query.evaluateRow(options.sourceTable, options.record, bucketPrefix));
-        }
-        return results;
+  evaluateRow(options: EvaluateRowOptions) {
+    let results: SourceEvaluationResult[] = [];
+    for (let query of this.descriptor.dataQueries) {
+      if (!query.applies(options.sourceTable)) {
+        continue;
       }
-    };
+
+      results.push(...query.evaluateRow(options.sourceTable, options.record));
+    }
+    return results;
   }
 
   getSourceTables(): Set<TablePattern> {