Use HydrationState for bucket names.

Author: rkistner

packages/sync-rules/src/BucketSource.ts

@@ -1,5 +1,6 @@
 import { BucketParameterQuerier, ParameterLookup, PendingQueriers } from './BucketParameterQuerier.js';
 import { ColumnDefinition } from './ExpressionType.js';
+import { HydrationState } from './HydrationState.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
 import { GetQuerierOptions } from './SqlSyncRules.js';
 import { TablePattern } from './TablePattern.js';
@@ -13,7 +14,11 @@ import {
 } from './types.js';
 
 export interface CreateSourceParams {
-  bucketIdTransformer: BucketIdTransformer;
+  hydrationState?: HydrationState;
+  /**
+   * @deprecated Use hydrationState instead.
+   */
+  bucketIdTransformer?: BucketIdTransformer;
 }
 
 /**
@@ -123,6 +128,13 @@ export interface BucketParameterQuerierSourceDefinition {
    */
   readonly bucketParameters: string[];
 
+  /**
+   * The data source linked to this querier. This determines the bucket names that the querier generates.
+   *
+   * Note that queriers do not persist data themselves; they only resolve which buckets to load based on request parameters.
+   */
+  readonly querierDataSource: BucketDataSourceDefinition;
+
   createParameterQuerierSource(params: CreateSourceParams): BucketParameterQuerierSource;
 }
 

packages/sync-rules/src/HydrationState.ts

@@ -0,0 +1,105 @@
+import { BucketDataSource, BucketDataSourceDefinition, BucketParameterLookupSourceDefinition } from './BucketSource.js';
+import { BucketIdTransformer, CompatibilityContext, CompatibilityOption, CreateSourceParams } from './index.js';
+
+export interface BucketSourceState {
+  /** The prefix is the bucket name before the parameters. */
+  bucketPrefix: string;
+}
+
+export interface BucketParameterLookupSourceState {
+  /** The lookup name + queryid is used to reference the parameter lookup record. */
+  lookupName: string;
+  queryId: string;
+}
+
+/**
+ * Hydration state information for a source.
+ *
+ * This is what keeps track of bucket name and parameter lookup mappings for hydration. This can be used
+ * both to re-use mappings across hydrations of different sync rule versions, or to generate new mappings.
+ */
+export interface HydrationState<
+  T extends BucketSourceState = BucketSourceState,
+  U extends BucketParameterLookupSourceState = BucketParameterLookupSourceState
+> {
+  /**
+   * Given a bucket data source definition, get the bucket prefix to use for it.
+   */
+  getBucketSourceState(source: BucketDataSourceDefinition): T;
+
+  /**
+   * Given a bucket parameter lookup definition, get the persistence name to use.
+   */
+  getParameterLookupState(source: BucketParameterLookupSourceDefinition): U;
+}
+
+/**
+ * This represents hydration state that performs no transformations.
+ *
+ * This is the legacy default behavior with no bucket versioning.
+ */
+export const DEFAULT_HYDRATION_STATE: HydrationState = {
+  getBucketSourceState(source: BucketDataSourceDefinition) {
+    return {
+      bucketPrefix: source.defaultBucketPrefix
+    };
+  },
+  getParameterLookupState(source) {
+    return {
+      lookupName: source.defaultLookupName,
+      queryId: source.defaultQueryId
+    };
+  }
+};
+
+export function versionedHydrationState(version: number) {
+  return new VersionedHydrationState((bucketId: string) => {
+    return `${version}#${bucketId}`;
+  });
+}
+
+export class VersionedHydrationState implements HydrationState {
+  constructor(private transformer: BucketIdTransformer) {}
+
+  getBucketSourceState(source: BucketDataSourceDefinition): BucketSourceState {
+    return {
+      bucketPrefix: this.transformer(source.defaultBucketPrefix)
+    };
+  }
+
+  getParameterLookupState(source: BucketParameterLookupSourceDefinition): BucketParameterLookupSourceState {
+    // No transformations applied here
+    return {
+      lookupName: source.defaultLookupName,
+      queryId: source.defaultQueryId
+    };
+  }
+}
+
+export class BucketIdTransformerHydrationState implements HydrationState {
+  constructor(private transformer: BucketIdTransformer) {}
+
+  getBucketSourceState(source: BucketDataSourceDefinition): BucketSourceState {
+    return {
+      bucketPrefix: this.transformer(source.defaultBucketPrefix)
+    };
+  }
+
+  getParameterLookupState(source: BucketParameterLookupSourceDefinition): BucketParameterLookupSourceState {
+    // No transformations applied here
+    return {
+      lookupName: source.defaultLookupName,
+      queryId: source.defaultQueryId
+    };
+  }
+}
+
+export function resolveHydrationState(params: CreateSourceParams): HydrationState {
+  if (params.hydrationState) {
+    return params.hydrationState;
+  } else if (params.bucketIdTransformer) {
+    return new BucketIdTransformerHydrationState(params.bucketIdTransformer);
+  } else {
+    return DEFAULT_HYDRATION_STATE;
+  }
+}

packages/sync-rules/src/SqlBucketDescriptor.ts

@@ -6,6 +6,7 @@ import {
   CreateSourceParams
 } from './BucketSource.js';
 import { ColumnDefinition } from './ExpressionType.js';
+import { resolveHydrationState } from './HydrationState.js';
 import { IdSequence } from './IdSequence.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
 import { SqlDataQuery } from './SqlDataQuery.js';
@@ -83,7 +84,13 @@ export class SqlBucketDescriptor implements BucketSource {
   }
 
   addParameterQuery(sql: string, options: QueryParseOptions): QueryParseResult {
-    const parameterQuery = SqlParameterQuery.fromSql(this.name, sql, options, this.parameterIdSequence.nextId());
+    const parameterQuery = SqlParameterQuery.fromSql(
+      this.name,
+      sql,
+      options,
+      this.parameterIdSequence.nextId(),
+      this.dataSource
+    );
     if (this.bucketParametersInternal == null) {
       this.bucketParametersInternal = parameterQuery.bucketParameters;
     } else {
@@ -151,6 +158,8 @@ export class BucketDefinitionDataSource implements BucketDataSourceDefinition {
   }
 
   createDataSource(params: CreateSourceParams): BucketDataSource {
+    const hydrationState = resolveHydrationState(params);
+    const bucketPrefix = hydrationState.getBucketSourceState(this).bucketPrefix;
     return {
       evaluateRow: (options) => {
         let results: EvaluationResult[] = [];
@@ -159,7 +168,7 @@ export class BucketDefinitionDataSource implements BucketDataSourceDefinition {
             continue;
           }
 
-          results.push(...query.evaluateRow(options.sourceTable, options.record, params.bucketIdTransformer));
+          results.push(...query.evaluateRow(options.sourceTable, options.record, bucketPrefix));
         }
         return results;
       }

packages/sync-rules/src/SqlDataQuery.ts

@@ -192,19 +192,13 @@ export class SqlDataQuery extends BaseSqlDataQuery {
     this.filter = options.filter;
   }
 
-  evaluateRow(
-    table: SourceTableInterface,
-    row: SqliteRow,
-    bucketIdTransformer: BucketIdTransformer
-  ): EvaluationResult[] {
+  evaluateRow(table: SourceTableInterface, row: SqliteRow, bucketPrefix: string): EvaluationResult[] {
     return this.evaluateRowWithOptions({
       table,
       row,
       bucketIds: (tables) => {
         const bucketParameters = this.filter.filterRow(tables);
-        return bucketParameters.map((params) =>
-          getBucketId(this.descriptorName, this.bucketParameters, params, bucketIdTransformer)
-        );
+        return bucketParameters.map((params) => getBucketId(bucketPrefix, this.bucketParameters, params));
       }
     });
   }

packages/sync-rules/src/SqlParameterQuery.ts

@@ -19,7 +19,7 @@ import {
   CreateSourceParams
 } from './BucketSource.js';
 import { SqlRuleError } from './errors.js';
-import { GetQuerierOptions } from './index.js';
+import { BucketDataSourceDefinition, GetQuerierOptions } from './index.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
 import { AvailableTable, SqlTools } from './sql_filters.js';
 import { checkUnsupportedFeatures, isClauseError } from './sql_support.js';
@@ -44,6 +44,7 @@ import {
 } from './types.js';
 import { filterJsonRow, getBucketId, isJsonValue, isSelectStatement, normalizeParameterValue } from './utils.js';
 import { DetectRequestParameters } from './validators.js';
+import { HydrationState, resolveHydrationState } from './HydrationState.js';
 
 export interface SqlParameterQueryOptions {
   sourceTable: TablePattern;
@@ -59,6 +60,7 @@ export interface SqlParameterQueryOptions {
   bucketParameters: string[];
   queryId: string;
   tools: SqlTools;
+  querierDataSource: BucketDataSourceDefinition;
   errors?: SqlRuleError[];
 }
 
@@ -75,7 +77,8 @@ export class SqlParameterQuery
     descriptorName: string,
     sql: string,
     options: QueryParseOptions,
-    queryId: string
+    queryId: string,
+    querierDataSource: BucketDataSourceDefinition
   ): SqlParameterQuery | StaticSqlParameterQuery | TableValuedFunctionSqlParameterQuery {
     const parsed = parse(sql, { locationTracking: true });
     const schema = options?.schema;
@@ -91,7 +94,7 @@ export class SqlParameterQuery
 
     if (q.from == null) {
       // E.g. SELECT token_parameters.user_id as user_id WHERE token_parameters.is_admin
-      return StaticSqlParameterQuery.fromSql(descriptorName, sql, q, options, queryId);
+      return StaticSqlParameterQuery.fromSql(descriptorName, sql, q, options, queryId, querierDataSource);
     }
 
     let errors: SqlRuleError[] = [];
@@ -102,7 +105,15 @@ export class SqlParameterQuery
       throw new SqlRuleError('Must SELECT from a single table', sql, q.from?.[0]._location);
     } else if (q.from[0].type == 'call') {
       const from = q.from[0];
-      return TableValuedFunctionSqlParameterQuery.fromSql(descriptorName, sql, from, q, options, queryId);
+      return TableValuedFunctionSqlParameterQuery.fromSql(
+        descriptorName,
+        sql,
+        from,
+        q,
+        options,
+        queryId,
+        querierDataSource
+      );
     } else if (q.from[0].type == 'statement') {
       throw new SqlRuleError('Subqueries are not supported yet', sql, q.from?.[0]._location);
     }
@@ -203,6 +214,7 @@ export class SqlParameterQuery
       bucketParameters,
       queryId,
       tools,
+      querierDataSource,
       errors
     });
 
@@ -297,6 +309,8 @@ export class SqlParameterQuery
   readonly queryId: string;
   readonly tools: SqlTools;
 
+  readonly querierDataSource: BucketDataSourceDefinition;
+
   readonly errors: SqlRuleError[];
 
   constructor(options: SqlParameterQueryOptions) {
@@ -314,6 +328,7 @@ export class SqlParameterQuery
     this.queryId = options.queryId;
     this.tools = options.tools;
     this.errors = options.errors ?? [];
+    this.querierDataSource = options.querierDataSource;
   }
 
   public get defaultLookupName(): string {
@@ -333,15 +348,18 @@ export class SqlParameterQuery
   }
 
   createParameterQuerierSource(params: CreateSourceParams): BucketParameterQuerierSource {
+    const hydrationState = resolveHydrationState(params);
+    const bucketPrefix = hydrationState.getBucketSourceState(this.querierDataSource).bucketPrefix;
     return {
       pushBucketParameterQueriers: (result: PendingQueriers, options: GetQuerierOptions) => {
-        const q = this.getBucketParameterQuerier(options.globalParameters, ['default'], params.bucketIdTransformer);
+        const q = this.getBucketParameterQuerier(options.globalParameters, ['default'], bucketPrefix);
         result.queriers.push(q);
       }
     };
   }
 
   createParameterLookupSource(params: CreateSourceParams): BucketParameterLookupSource {
+    // FIXME: Use HydrationState for lookups.
     return {
       evaluateParameterRow: (sourceTable: SourceTableInterface, row: SqliteRow): EvaluatedParametersResult[] => {
         if (this.tableSyncsParameters(sourceTable)) {
@@ -403,7 +421,7 @@ export class SqlParameterQuery
   resolveBucketDescriptions(
     bucketParameters: SqliteJsonRow[],
     parameters: RequestParameters,
-    transformer: BucketIdTransformer
+    bucketPrefix: string
   ): BucketDescription[] {
     // Filters have already been applied and gotten us the set of bucketParameters - don't attempt to filter again.
     // We _do_ need to evaluate the output columns here, using a combination of precomputed bucketParameters,
@@ -428,7 +446,7 @@ export class SqlParameterQuery
         }
 
         return {
-          bucket: getBucketId(this.descriptorName, this.bucketParameters, result, transformer),
+          bucket: getBucketId(bucketPrefix, this.bucketParameters, result),
           priority: this.priority
         };
       })
@@ -514,7 +532,7 @@ export class SqlParameterQuery
   getBucketParameterQuerier(
     requestParameters: RequestParameters,
     reasons: BucketInclusionReason[],
-    transformer: BucketIdTransformer
+    bucketPrefix: string
   ): BucketParameterQuerier {
     const lookups = this.getLookups(requestParameters);
     if (lookups.length == 0) {
@@ -534,7 +552,7 @@ export class SqlParameterQuery
       parameterQueryLookups: lookups,
       queryDynamicBucketDescriptions: async (source: ParameterLookupSource) => {
         const bucketParameters = await source.getParameterSets(lookups);
-        return this.resolveBucketDescriptions(bucketParameters, requestParameters, transformer).map((bucket) => ({
+        return this.resolveBucketDescriptions(bucketParameters, requestParameters, bucketPrefix).map((bucket) => ({
           ...bucket,
           definition: this.descriptorName,
           inclusion_reasons: reasons