Move SubqueryParameterLookupSource deeper to be specific to variants.

Author: rkistner

packages/sync-rules/src/BucketSource.ts

@@ -65,6 +65,13 @@ export interface HydratedBucketSource {
  * Encodes a static definition of a bucket source, as parsed from sync rules or stream definitions.
  */
 export interface BucketDataSourceDefinition {
+  /**
+   * Bucket prefix if no transformations are defined.
+   *
+   * Transformations may use this as a base, or may generate an entirely different prefix.
+   */
+  readonly defaultBucketPrefix: string;
+
   /**
    * For debug use only.
    */
@@ -90,6 +97,14 @@ export interface BucketDataSourceDefinition {
  * This is only relevant for parameter queries that query tables.
  */
 export interface BucketParameterLookupSourceDefinition {
+  /**
+   * lookupName + queryId is used to uniquely identify parameter queries for parameter storage.
+   *
+   * This defines the default values if no transformations are applied.
+   */
+  defaultLookupName: string;
+  defaultQueryId: string;
+
   getSourceTables(): Set<TablePattern>;
   createParameterLookupSource(params: CreateSourceParams): BucketParameterLookupSource;
 

packages/sync-rules/src/SqlBucketDescriptor.ts

@@ -146,6 +146,10 @@ export class BucketDefinitionDataSource implements BucketDataSourceDefinition {
     return this.descriptor.bucketParameters;
   }
 
+  public get defaultBucketPrefix(): string {
+    return this.descriptor.name;
+  }
+
   createDataSource(params: CreateSourceParams): BucketDataSource {
     return {
       evaluateRow: (options) => {

packages/sync-rules/src/SqlParameterQuery.ts

@@ -316,6 +316,14 @@ export class SqlParameterQuery
     this.errors = options.errors ?? [];
   }
 
+  public get defaultLookupName(): string {
+    return this.descriptorName;
+  }
+
+  public get defaultQueryId(): string {
+    return this.queryId;
+  }
+
   tableSyncsParameters(table: SourceTableInterface): boolean {
     return this.sourceTable.matches(table);
   }

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

@@ -1,6 +1,13 @@
 import { isParameterValueClause, isRowValueClause, SQLITE_TRUE, sqliteBool } from '../sql_support.js';
 import { TablePattern } from '../TablePattern.js';
-import { ParameterMatchClause, ParameterValueClause, RowValueClause, SqliteJsonValue } from '../types.js';
+import {
+  EvaluatedParametersResult,
+  ParameterMatchClause,
+  ParameterValueClause,
+  RowValueClause,
+  SqliteJsonValue,
+  SqliteRow
+} from '../types.js';
 import { isJsonValue, normalizeParameterValue } from '../utils.js';
 import { SqlTools } from '../sql_filters.js';
 import { checkJsonArray, OPERATOR_NOT } from '../sql_functions.js';
@@ -10,6 +17,12 @@ import { StreamVariant } from './variant.js';
 import { SubqueryEvaluator } from './parameter.js';
 import { cartesianProduct } from './utils.js';
 import { NodeLocation } from 'pgsql-ast-parser';
+import {
+  BucketParameterLookupSource,
+  BucketParameterLookupSourceDefinition,
+  CreateSourceParams
+} from '../BucketSource.js';
+import { SourceTableInterface } from '../SourceTableInterface.js';
 
 /**
  * An intermediate representation of a `WHERE` clause for stream queries.
@@ -253,19 +266,10 @@ export class Subquery {
 
     const evaluator: SubqueryEvaluator = {
       parameterTable: this.table,
-      lookupsForParameterRow(sourceTable, row) {
-        const value = column.evaluate({ [sourceTable.name]: row });
-        if (!isJsonValue(value)) {
-          return null;
-        }
-
-        const lookups: ParameterLookup[] = [];
-        for (const [variant, id] of innerVariants) {
-          for (const instantiation of variant.instantiationsForRow({ sourceTable, record: row })) {
-            lookups.push(ParameterLookup.normalized(context.streamName, id, instantiation));
-          }
-        }
-        return { value, lookups };
+      lookupSources(streamName) {
+        return innerVariants.map(([variant, id]) => {
+          return new SubqueryParameterLookupSource(evaluator, column, variant, id, streamName);
+        });
       },
       lookupsForRequest(parameters) {
         const lookups: ParameterLookup[] = [];
@@ -510,3 +514,75 @@ export class EvaluateSimpleCondition extends FilterOperator {
     );
   }
 }
+
+export class SubqueryParameterLookupSource implements BucketParameterLookupSourceDefinition {
+  constructor(
+    private subquery: SubqueryEvaluator,
+    private column: RowValueClause,
+    private innerVariant: StreamVariant,
+    public readonly defaultQueryId: string,
+    private streamName: string
+  ) {}
+
+  get defaultLookupName() {
+    return this.streamName;
+  }
+
+  getSourceTables(): Set<TablePattern> {
+    let result = new Set<TablePattern>();
+    result.add(this.subquery.parameterTable);
+    return result;
+  }
+
+  /**
+   * Creates lookup indices for dynamically-resolved parameters.
+   *
+   * Resolving dynamic parameters is a two-step process: First, for tables referenced in subqueries, we create an index
+   * to resolve which request parameters would match rows in subqueries. Then, when resolving bucket ids for a request,
+   * we compute subquery results by looking up results in that index.
+   *
+   * This implements the first step of that process.
+   *
+   * @param result The array into which evaluation results should be written to.
+   * @param sourceTable A table we depend on in a subquery.
+   * @param row Row data to index.
+   */
+  evaluateParameterRow(sourceTable: SourceTableInterface, row: SqliteRow): EvaluatedParametersResult[] {
+    if (this.subquery.parameterTable.matches(sourceTable)) {
+      // Theoretically we're doing duplicate work by doing this for each innerVariant in a subquery.
+      // In practice, we don't have more than one innerVariant per subquery right now, so this is fine.
+      const value = this.column.evaluate({ [sourceTable.name]: row });
+      if (!isJsonValue(value)) {
+        return [];
+      }
+
+      const lookups: ParameterLookup[] = [];
+      for (const instantiation of this.innerVariant.instantiationsForRow({ sourceTable, record: row })) {
+        // TODO: dynamic lookup name and query id
+        lookups.push(ParameterLookup.normalized(this.defaultLookupName, this.defaultQueryId, instantiation));
+      }
+
+      // The row of the subquery. Since we only support subqueries with a single column, we unconditionally name the
+      // column `result` for simplicity.
+      const resultRow = { result: value };
+
+      return lookups.map((l) => ({
+        lookup: l,
+        bucketParameters: [resultRow]
+      }));
+    }
+    return [];
+  }
+
+  createParameterLookupSource(params: CreateSourceParams): BucketParameterLookupSource {
+    return {
+      evaluateParameterRow: (sourceTable, row) => {
+        return this.evaluateParameterRow(sourceTable, row);
+      }
+    };
+  }
+
+  tableSyncsParameters(table: SourceTableInterface): boolean {
+    return this.subquery.parameterTable.matches(table);
+  }
+}

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

@@ -1,4 +1,5 @@
 import { ParameterLookup } from '../BucketParameterQuerier.js';
+import { BucketParameterLookupSource, BucketParameterLookupSourceDefinition } from '../BucketSource.js';
 import { SourceTableInterface } from '../SourceTableInterface.js';
 import { TablePattern } from '../TablePattern.js';
 import {
@@ -38,8 +39,9 @@ export interface BucketParameter {
 export interface SubqueryEvaluator {
   parameterTable: TablePattern;
 
-  lookupsForParameterRow(sourceTable: SourceTableInterface, row: SqliteRow): SubqueryLookups | null;
   lookupsForRequest(params: RequestParameters): ParameterLookup[];
+
+  lookupSources(streamName: string): BucketParameterLookupSourceDefinition[];
 }
 
 export interface SubqueryLookups {

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

@@ -47,7 +47,7 @@ export class SyncStream implements BucketSource {
 
     this.dataSources = variants.map((variant) => new SyncStreamDataSource(this, data, variant));
     this.parameterQuerierSources = variants.map((variant) => new SyncStreamParameterQuerierSource(this, variant));
-    this.parameterLookupSources = variants.map((variant) => new SyncStreamParameterLookupSource(this, variant));
+    this.parameterLookupSources = variants.flatMap((variant) => variant.lookupSources(name));
   }
 
   public get type(): BucketSourceType {
@@ -81,6 +81,10 @@ export class SyncStreamDataSource implements BucketDataSourceDefinition {
     return [];
   }
 
+  public get defaultBucketPrefix(): string {
+    return this.variant.defaultBucketPrefix(this.stream.name);
+  }
+
   getSourceTables(): Set<TablePattern> {
     return new Set<TablePattern>([this.data.sourceTable]);
   }
@@ -199,39 +203,3 @@ export class SyncStreamParameterQuerierSource implements BucketParameterQuerierS
     }
   }
 }
-
-export class SyncStreamParameterLookupSource implements BucketParameterLookupSourceDefinition {
-  constructor(
-    private stream: SyncStream,
-    private variant: StreamVariant
-  ) {}
-
-  getSourceTables(): Set<TablePattern> {
-    let result = new Set<TablePattern>();
-    for (const subquery of this.variant.subqueries) {
-      result.add(subquery.parameterTable);
-    }
-
-    return result;
-  }
-
-  createParameterLookupSource(params: CreateSourceParams): BucketParameterLookupSource {
-    return {
-      evaluateParameterRow: (sourceTable, row) => {
-        const result: EvaluatedParametersResult[] = [];
-        this.variant.pushParameterRowEvaluation(result, sourceTable, row);
-        return result;
-      }
-    };
-  }
-
-  tableSyncsParameters(table: SourceTableInterface): boolean {
-    for (const subquery of this.variant.subqueries) {
-      if (subquery.parameterTable.matches(table)) {
-        return true;
-      }
-    }
-
-    return false;
-  }
-}

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

@@ -1,6 +1,12 @@
 import { BucketInclusionReason, ResolvedBucket } from '../BucketDescription.js';
 import { BucketParameterQuerier, ParameterLookup } from '../BucketParameterQuerier.js';
+import {
+  BucketParameterLookupSource,
+  BucketParameterLookupSourceDefinition,
+  CreateSourceParams
+} from '../BucketSource.js';
 import { SourceTableInterface } from '../SourceTableInterface.js';
+import { TablePattern } from '../TablePattern.js';
 import {
   BucketIdTransformer,
   EvaluatedParametersResult,
@@ -65,6 +71,14 @@ export class StreamVariant {
     this.requestFilters = [];
   }
 
+  defaultBucketPrefix(streamName: string): string {
+    return `${streamName}|${this.id}`;
+  }
+
+  lookupSources(streamName: string): BucketParameterLookupSourceDefinition[] {
+    return this.subqueries.flatMap((subquery) => subquery.lookupSources(streamName));
+  }
+
   /**
    * Given a row in the table this stream selects from, returns all ids of buckets to which that row belongs to.
    */
@@ -216,41 +230,6 @@ export class StreamVariant {
     );
   }
 
-  /**
-   * Creates lookup indices for dynamically-resolved parameters.
-   *
-   * Resolving dynamic parameters is a two-step process: First, for tables referenced in subqueries, we create an index
-   * to resolve which request parameters would match rows in subqueries. Then, when resolving bucket ids for a request,
-   * we compute subquery results by looking up results in that index.
-   *
-   * This implements the first step of that process.
-   *
-   * @param result The array into which evaluation results should be written to.
-   * @param sourceTable A table we depend on in a subquery.
-   * @param row Row data to index.
-   */
-  pushParameterRowEvaluation(result: EvaluatedParametersResult[], sourceTable: SourceTableInterface, row: SqliteRow) {
-    for (const subquery of this.subqueries) {
-      if (subquery.parameterTable.matches(sourceTable)) {
-        const lookups = subquery.lookupsForParameterRow(sourceTable, row);
-        if (lookups == null) {
-          continue;
-        }
-
-        // The row of the subquery. Since we only support subqueries with a single column, we unconditionally name the
-        // column `result` for simplicity.
-        const resultRow = { result: lookups.value };
-
-        result.push(
-          ...lookups.lookups.map((l) => ({
-            lookup: l,
-            bucketParameters: [resultRow]
-          }))
-        );
-      }
-    }
-  }
-
   debugRepresentation(): any {
     return {
       id: this.id,