diff --git a/packages/sync-rules/src/sync_plan/evaluator/parameter_evaluator.ts b/packages/sync-rules/src/sync_plan/evaluator/parameter_evaluator.ts index 0a43dd7d8..22302428d 100644 --- a/packages/sync-rules/src/sync_plan/evaluator/parameter_evaluator.ts +++ b/packages/sync-rules/src/sync_plan/evaluator/parameter_evaluator.ts @@ -1,6 +1,5 @@ import { ParameterLookupSource, ScopedParameterLookup, UnscopedParameterLookup } from '../../BucketParameterQuerier.js'; import { ParameterIndexLookupCreator } from '../../BucketSource.js'; -import { HashMap, listEquality, StableHasher } from '../../compiler/equality.js'; import { HydrationState } from '../../HydrationState.js'; import { RequestParameters, SqliteParameterValue, SqliteValue } from '../../types.js'; import { isValidParameterValue } from '../../utils.js'; @@ -14,6 +13,7 @@ import { MapSourceVisitor, visitExpr } from '../expression_visitor.js'; import * as plan from '../plan.js'; import { StreamInput } from './bucket_source.js'; import { PreparedParameterIndexLookupCreator } from './parameter_index_lookup_creator.js'; +import { AsyncJoinLookup, ResultSet, ResultSetColumn, ResultSetElement } from './result_set.js'; /** * Finds bucket parameters for a given request or subscription. @@ -55,15 +55,21 @@ import { PreparedParameterIndexLookupCreator } from './parameter_index_lookup_cr */ export class RequestParameterEvaluators { private constructor( - readonly stream: plan.StreamOptions, + private readonly stream: plan.StreamOptions, /** * Pending lookup stages, or their cached outputs. */ - readonly lookupStages: PreparedExpandingLookup[][], + private readonly lookupStages: LookupStage[], /** * Pending parameter values, or their cached outputs. */ - readonly parameterValues: PreparedParameterValue[] + private readonly parameterValues: PreparedParameterValue[], + + /** + * The materialized result set containing lookup values. {@link parameterValues} are read from this result set as a + * final step. + */ + private readonly resultSet: ResultSet ) {} /** @@ -77,33 +83,21 @@ export class RequestParameterEvaluators { * instead of re-evaluating them on every parameter lookup change. */ clone(): RequestParameterEvaluators { - function cloneValue(value: PreparedParameterValue): PreparedParameterValue { - switch (value.type) { - case 'intersection': - return { type: 'intersection', values: value.values.map(cloneValue) }; - case 'request': - case 'lookup': - case 'cached': - return value; - } - } + const clonedParameters = new Map(); - function cloneLookup(lookup: PreparedExpandingLookup): PreparedExpandingLookup { - switch (lookup.type) { - case 'parameter': - // We need to clone the instantiation array as well. - return { type: 'parameter', lookup: lookup.lookup, instantiation: lookup.instantiation.map(cloneValue) }; - case 'table_valued': - case 'cached': - return lookup; - } + function cloneParameter(original: PreparedParameterValue) { + const existing = clonedParameters.get(original); + if (existing != null) return existing; + + const clone = original.clone(); + clonedParameters.set(original, clone); + return clone; } - return new RequestParameterEvaluators( - this.stream, - this.lookupStages.map((stage) => stage.map(cloneLookup)), - this.parameterValues.map(cloneValue) - ); + const copiedStages = this.lookupStages.map((s) => s.clone(cloneParameter)); + const outputValues = this.parameterValues.map(cloneParameter); + + return new RequestParameterEvaluators(this.stream, copiedStages, outputValues, this.resultSet.clone()); } /** @@ -115,13 +109,48 @@ export class RequestParameterEvaluators { * If dynamic lookups are required to resolve parameters, returns `undefined`. */ partiallyInstantiate(input: PartialInstantiationInput): SqliteParameterValue[][] | undefined { - const helper = new PartialInstantiator(input, this); + try { + // At this point, we can resolve table-valued lookups and parameter values based only on request data. + for (const stage of this.lookupStages) { + let needsParameterLookups = false; + + for (const element of stage.lookups) { + if (element instanceof TableValuedExpandingLookup) { + const outputs = element.read(input.request); + element.wasResolved = true; + this.resultSet.multiply(element.resultSetIndex, outputs); + + this.#checkInstantiable(); + } else { + needsParameterLookups = true; + } + } - this.lookupStages.forEach((stage, stageIndex) => { - stage.forEach((_, indexInStage) => helper.expandingLookupSync(stageIndex, indexInStage)); - }); + for (const instantiation of stage.inputParameters()) { + if (instantiation instanceof RequestParameterValue) { + instantiation.resolveWith(input); + } else if (instantiation.lookup instanceof ParameterIndexExpandingLookup) { + needsParameterLookups = true; + } + } - return helper.tryResolveInstantiation(this.parameterValues)?.map(withoutProvenance); + if (!needsParameterLookups) { + for (const intersection of stage.intersections) { + this.#applyIntersectionConstraint(intersection); + } + } + } + + for (const parameter of this.parameterValues) { + if (parameter instanceof RequestParameterValue) parameter.resolveWith(input); + } + + return this.#readParameters(); + } catch (e) { + if (e === uninstantiableException) return []; + + throw e; + } } /** @@ -130,16 +159,137 @@ export class RequestParameterEvaluators { * Because this needs to lookup parameter indexes, it is asynchronous. */ async instantiate(input: InstantiationInput): Promise { - const helper = new FullInstantiator(input, this); + try { + for (const { lookups, intersections } of this.lookupStages) { + for (const lookup of lookups) { + if (lookup instanceof ParameterIndexExpandingLookup) { + await this.#instantiateLookup(lookup, input); + } + } + + for (const intersection of intersections) { + this.#applyIntersectionConstraint(intersection); + } + } + + const params = this.#readParameters(); + if (params == null) { + throw new Error('internal error: Should have been able to resolve instantiation after instantiating stages.'); + } + return params; + } catch (e) { + if (e === uninstantiableException) return []; + + throw e; + } + } + + #checkInstantiable() { + if (this.resultSet.length === 0) throw uninstantiableException; + } + + #readParameters(): SqliteParameterValue[][] | undefined { + for (const { intersections, lookups } of this.lookupStages) { + for (const intersection of intersections) { + if (!intersection.wasApplied) return undefined; + } + + for (const element of lookups) { + if (!element.wasResolved) return undefined; + } + } + + return this.#readValues(this.parameterValues); + } + + #readValues(values: PreparedParameterValue[]): SqliteParameterValue[][] { + const allInstantiations: SqliteParameterValue[][] = []; + + for (const row of this.resultSet.projectUnique(values.filter((v) => v instanceof LookupParameterValue))) { + allInstantiations.push(this.#evaluateAgainstRow(row, values)); + } + + return allInstantiations; + } + + #evaluateAgainstRow(row: SqliteParameterValue[], projection: PreparedParameterValue[]) { + let lookupIndex = 0; + + return projection.map((v) => { + if (v instanceof LookupParameterValue) { + return row[lookupIndex++]; + } else { + return v.requireResolved(); + } + }); + } + + #applyIntersectionConstraint(constraint: RequiredIntersection) { + if (constraint.wasApplied) return; + + // If any parameter of the intersection is a scalar value derived from a request, that value. + let knownValue: SqliteParameterValue | undefined; + const intersection: ResultSetColumn[] = []; + + for (const value of constraint.values) { + if (value instanceof RequestParameterValue) { + const evaluated = value.requireResolved(); + if (knownValue !== undefined && evaluated !== knownValue) { + throw uninstantiableException; + } - for (let i = 0; i < this.lookupStages.length; i++) { - // Within a stage, we can resolve lookups concurrently. - await Promise.all(this.lookupStages[i].map((_, j) => helper.expandingLookup(i, j))); + knownValue = evaluated; + } else { + intersection.push(value); + } } - // At this point, all lookups have been resolved and we can synchronously evaluate parameters which might depend on - // those lookups. - return helper.resolveInstantiation(this.parameterValues).map(withoutProvenance); + this.resultSet.formIntersection(intersection, knownValue); + constraint.wasApplied = true; + this.#checkInstantiable(); + } + + async #instantiateLookup(lookup: ParameterIndexExpandingLookup, input: InstantiationInput) { + const scope = input.hydrationState.getParameterIndexLookupScope(lookup.lookup); + const resolvedLookup = lookup.lookup as PreparedParameterIndexLookupCreator; + + await this.resultSet.joinAsync( + lookup.instantiation.filter((v) => v instanceof LookupParameterValue), + lookup.resultSetIndex, + async (inputs) => { + const bucketStorageLookups = new Map(); + for (const input of inputs) { + bucketStorageLookups.set( + ScopedParameterLookup.normalized( + scope, + UnscopedParameterLookup.normalized(this.#evaluateAgainstRow(input.inputs, lookup.instantiation)) + ), + input + ); + } + + const outputs = await input.source.getParameterSets( + [...bucketStorageLookups.keys()], + `Stream ${this.stream.name} evaluating parameter on ${resolvedLookup.sourceTable.tablePattern}` + ); + + for (const { lookup, rows } of outputs) { + const join = bucketStorageLookups.get(lookup)!; + + for (const row of rows) { + const length = Object.entries(row).length; + const asArray: SqliteParameterValue[] = []; + for (let i = 0; i < length; i++) { + asArray.push(row[i.toString()] as SqliteParameterValue); + } + + join.foundRows.push(asArray); + } + } + } + ); + lookup.wasResolved = true; + this.#checkInstantiable(); } /** @@ -159,8 +309,9 @@ export class RequestParameterEvaluators { input: StreamInput, engine: ScalarExpressionEngine ) { - const mappedStages: PreparedExpandingLookup[][] = []; - const lookupToStage = new Map(); + const mappedStages: LookupStage[] = []; + let amountOfLookups = 0; + const lookupToStage = new Map(); function mapParameterValue(value: plan.ParameterValue): PreparedParameterValue { if (value.type == 'request') { @@ -169,17 +320,26 @@ export class RequestParameterEvaluators { const prepared = engine.prepareEvaluator({ filters: [], outputs: [mapper.transform(value.expr)] }); const instantiation = mapper.instantiation; - return { - type: 'request', - read(request) { - return prepared.evaluate(parametersForRequest(request, instantiation))[0][0]; - } - }; + return new RequestParameterValue( + (request) => prepared.evaluate(parametersForRequest(request, instantiation))[0][0] + ); } else if (value.type == 'lookup') { - const stagePosition = lookupToStage.get(value.lookup)!; - return { type: 'lookup', lookup: stagePosition, resultIndex: value.resultIndex }; + const lookup = lookupToStage.get(value.lookup)!; + return new LookupParameterValue(lookup, value.resultIndex); } else { - return { type: 'intersection', values: mapParameterValues(value.values) }; + const intersectionInputs = mapParameterValues(value.values); + + if (mappedStages.length > 0) { + mappedStages[mappedStages.length - 1].intersections.push({ values: intersectionInputs, wasApplied: false }); + } else { + // Intersection in first stage, e.g. for request parameters. Add a stage just for this. + const stage = new LookupStage([], [{ values: intersectionInputs, wasApplied: false }]); + mappedStages.push(stage); + } + + // Non-intersecting rows will be pruned from the result set or, for scalar parameter values, mark the querier + // as uninstantiable. So, we can replace the intersection value with any inner value. + return intersectionInputs[0]; } } @@ -188,20 +348,17 @@ export class RequestParameterEvaluators { } for (const stage of lookupStages) { - const stageIndex = mappedStages.length; - const mappedStage: PreparedExpandingLookup[] = []; - mappedStages.push(mappedStage); + const mappedStage = new LookupStage([], []); for (const lookup of stage) { - const index = mappedStage.length; - lookupToStage.set(lookup, { stage: stageIndex, index }); + let resolved: PreparedExpandingLookup; if (lookup.type == 'parameter') { - mappedStage.push({ - type: 'parameter', - lookup: input.preparedLookups.get(lookup.lookup)!, - instantiation: mapParameterValues(lookup.instantiation) - }); + resolved = new ParameterIndexExpandingLookup( + amountOfLookups++, + input.preparedLookups.get(lookup.lookup)!, + mapParameterValues(lookup.instantiation) + ); } else { // Create an expression like SELECT FROM table_valued() WHERE const mapInputs = mapExternalDataToInstantiation(); @@ -222,430 +379,160 @@ export class RequestParameterEvaluators { filters: lookup.filters.map((e) => visitExpr(mapOutputs, e, null)) }); - mappedStage.push({ - type: 'table_valued', - read(request) { - return [ - ...filterParameterRows(prepared.evaluate(parametersForRequest(request, mapInputs.instantiation))) - ]; - } - }); + resolved = new TableValuedExpandingLookup(amountOfLookups++, (request) => [ + ...filterParameterRows(prepared.evaluate(parametersForRequest(request, mapInputs.instantiation))) + ]); } - } - } - return new RequestParameterEvaluators(stream, mappedStages, mapParameterValues(values)); - } -} - -class PartialInstantiator { - constructor( - protected readonly input: I, - protected readonly evaluators: RequestParameterEvaluators - ) {} - - tryResolveInstantiation(params: PreparedParameterValue[]): ParameterValueWithRow[][] | undefined { - const stages = this.evaluators.lookupStages; - let hasUninstantiatedStage = false; - for (let stageIndex = 0; stageIndex < stages.length; stageIndex++) { - const stage = stages[stageIndex]; - for (let indexInStage = 0; indexInStage < stage.length; indexInStage++) { - const resolvedValues = this.expandingLookupSync(stageIndex, indexInStage); - if (resolvedValues == null) { - // Requires an asynchronous lookup to instantiate. - hasUninstantiatedStage = true; - continue; - } - - if (resolvedValues.length == 0) { - // Empty lookup stages make the entire graph uninstantiable, even if they're not used as a parameter. The - // reason for that is that queries like `WHERE 'static_value' IN (SELECT name FROM users WHERE id = auth.user_id())` - // are implemented as lookup stages, so we can't ignore them. - // Note that there is no construct like `OR` in a querier lookup (those always get compiled into separate - // queries), so any stage being empty guarantees that everything is uninstantiable. - return []; - } + lookupToStage.set(lookup, resolved); + mappedStage.lookups.push(resolved); } - } - if (hasUninstantiatedStage) { - return undefined; + mappedStages.push(mappedStage); } - // If we got to this point, all stages have been resolved. So we can resolve parameters without further async work. - return [...this.resolveInputs(params)]; - } - - protected *resolveInputs(params: PreparedParameterValue[]): Generator { - const parameterValues = params.map((_, index) => { - const cached = this.parameterSync(params, index); - if (cached == null) { - // This method is only called for inputs from an earlier stage, which should have been resolved at this point. - throw new Error('Should have been able to resolve parameter from earlier stage synchronously.'); - } - return cached; - }); - - yield* mergeValueCombinations(parameterValues); + const rs = new ResultSet(amountOfLookups); + return new RequestParameterEvaluators(stream, mappedStages, mapParameterValues(values), rs); } +} - /** - * If possible, evaluates an element in an array of parameter values and replaces the parameter with a marker - * indicating it as cached. - */ - parameterSync(parent: PreparedParameterValue[], index: number): ParameterValueWithRow[] | undefined { - const current = parent[index]; - if (current.type === 'cached') { - return current.values; - } else if (current.type === 'intersection') { - const columns: ParameterValueWithRow[][] = []; - for (let i = 0; i < current.values.length; i++) { - const evaluated = this.parameterSync(current.values, i); - if (evaluated == null) { - return undefined; // Can't evaluate sub-parameter - } - columns.push(evaluated); - } - - // For the most part, this just needs to find an intersection of values present in all columns. It gets more - // complicated for rows with provenance, however. For those. we need to ensure we find an intersection of values - // with compatible source rows. For example, consider an intersection of the same parameter lookup with columns - // `c1` and `c2`, and assume that we had the following rows: - // - // 1. Row {c1: 'a', c2: 'a'} - // 2. Row {c1: 'a', c2: 'b'} - // - // The intersection of this has one value: `a`, with a provenance of Row 1. To achieve this, we re-create rows - // by tracking a canonical value per row. If we see another value in the same row, we know that row can't - // contribute to the intersection because it has different values for `c1` and `c2`. - const poison = Symbol('poison'); - const valuesByResultSet = new Map>(); - const completedRows = new Map>(); - function markRowAsCompleted(resultSet: symbol, rowid: number) { - let rowids = completedRows.get(resultSet); - if (rowids == null) { - rowids = new Set(); - completedRows.set(resultSet, rowids); - } - - rowids.add(rowid); - } - - // Eliminate rows with conflicting values. - for (const column of columns) { - nextValue: for (const value of column) { - const row = value.directOrigin; - if (row) { - let forResultSet = valuesByResultSet.get(row.resultSet); - if (forResultSet == null) { - forResultSet = new Map(); - valuesByResultSet.set(row.resultSet, forResultSet); - } - - const existingValue = forResultSet.get(row.row); - if (existingValue != null && existingValue != value.value) { - forResultSet.set(row.row, poison); - markRowAsCompleted(row.resultSet, row.row); - continue nextValue; - } else { - forResultSet.set(row.row, value.value); - } - } - } - } - - function shouldSkipValue(value: ParameterValueWithRow): boolean { - if (value.directOrigin) { - const { resultSet, row } = value.directOrigin; - - const ignoredRowIds = completedRows.get(resultSet); - if (ignoredRowIds?.has(row)) return true; - - const valuesForResultSet = valuesByResultSet.get(resultSet); - if (valuesForResultSet == null) return false; - - if (valuesForResultSet.get(row) != value.value) return true; - } - - return false; - } - - let intersection: Map | null = null; - for (const column of columns) { - if (intersection == null) { - intersection = new Map(); - - for (const value of column) { - if (shouldSkipValue(value)) continue; - - const existing = intersection.get(value.value); - if (existing != null) { - existing.push(value.provenance); - } else { - intersection.set(value.value, [value.provenance]); - } +/** + * An internal exception thrown when no instantiation exists for a parameter. + * + * This is an exception to allow aborting the evaluator early. + */ +const uninstantiableException = Symbol.for('uninstantiable'); - for (const { resultSet, row } of value.provenance) { - // Any other value derived from this row must have the same value (otherwise we would have eliminated it). - // So we don't have to consider this row again. - markRowAsCompleted(resultSet, row); - } - } - } else { - const unmatchedValues = new Set(intersection.keys()); - - for (const value of column) { - const existing = intersection.get(value.value); - if (existing == null) { - // Value not in intersection, ignore. - } else { - unmatchedValues.delete(value.value); - - if (!shouldSkipValue(value)) { - // An intersection value is derived from all inputs, so we track them all as provenance. - existing.push(value.provenance); - } - } - } +export type PreparedExpandingLookup = TableValuedExpandingLookup | ParameterIndexExpandingLookup; - for (const unmatched of unmatchedValues) { - // Values in intersection before, but not in evaluated - intersection.delete(unmatched); - } - } +type CloneParameter = (original: PreparedParameterValue) => PreparedParameterValue; - if (intersection!.size == 0) { - // Empty intersection, we don't even need to evaluate the rest. - break; - } - } +abstract class BasePreparedExpandingLookup implements ResultSetElement { + wasResolved = false; - let values: ParameterValueWithRow[] = []; - if (intersection) { - intersection.forEach((provenances, value) => { - for (const provenance of provenances) { - values.push({ value, provenance }); - } - }); - } + constructor(readonly resultSetIndex: number) {} - parent[index] = { type: 'cached', values }; - return values; - } else if (current.type === 'lookup') { - const resolvedLookup = this.expandingLookupSync(current.lookup.stage, current.lookup.index); - if (resolvedLookup) { - const values = resolvedLookup.map((row) => row[current.resultIndex]); - parent[index] = { type: 'cached', values }; - return values; - } - } else if (current.type === 'request') { - const value = current.read(this.input.request); - const values: ParameterValueWithRow[] = isValidParameterValue(value) - ? [ - { - value, - provenance: [] - } - ] - : []; - - parent[index] = { type: 'cached', values }; - return values; - } + abstract clone(parameters: CloneParameter): BasePreparedExpandingLookup; +} - return undefined; +class TableValuedExpandingLookup extends BasePreparedExpandingLookup { + constructor( + resultSetIndex: number, + readonly read: (request: RequestParameters) => SqliteParameterValue[][] + ) { + super(resultSetIndex); } - expandingLookupSync(stage: number, index: number): ParameterValueWithRow[][] | undefined { - const lookup = this.evaluators.lookupStages[stage][index]; - if (lookup.type == 'table_valued') { - // We can evaluate this table-valued function already. - const resultSetMarker = Symbol(); - const values = lookup.read(this.input.request).map((values, rowid) => { - const directOrigin: VirtualSourceRow = { resultSet: resultSetMarker, row: rowid }; - const provenance: [VirtualSourceRow] = [directOrigin]; - return values.map((value) => ({ value, provenance, directOrigin }) satisfies ParameterValueWithRow); - }); - - this.evaluators.lookupStages[stage][index] = { type: 'cached', values }; - return values; - } else if (lookup.type == 'cached') { - return lookup.values; - } - - return undefined; + override clone(): TableValuedExpandingLookup { + const lookup = new TableValuedExpandingLookup(this.resultSetIndex, this.read); + lookup.wasResolved = this.wasResolved; + return lookup; } } -class FullInstantiator extends PartialInstantiator { - resolveInstantiation(params: PreparedParameterValue[]): ParameterValueWithRow[][] { - const resolved = this.tryResolveInstantiation(params); - if (resolved == null) { - throw new Error('internal error: Should have been able to resolve instantiation after instantiating stages.'); - } - - return resolved; +class ParameterIndexExpandingLookup extends BasePreparedExpandingLookup { + constructor( + resultSetIndex: number, + readonly lookup: ParameterIndexLookupCreator, + readonly instantiation: PreparedParameterValue[] + ) { + super(resultSetIndex); } - async expandingLookup(stage: number, index: number): Promise { - const lookup = this.evaluators.lookupStages[stage][index]; - if (lookup.type == 'parameter') { - const scope = this.input.hydrationState.getParameterIndexLookupScope(lookup.lookup); - const resolvedLookup = lookup.lookup as PreparedParameterIndexLookupCreator; - - interface PendingLookup { - lookup: ScopedParameterLookup; - provenancePaths: VirtualSourceRow[][]; - resultSet: symbol; - } - - // It's possible that we'll have the same logical lookup with multiple provenance values. For instance, if the - // outputs of another lookup with two columns (where only one column is an input to this lookup) are passed into - // this, we can have two lookups with identical keys but different provenances. This hash map de-duplicates keys. - const pendingLookups = new HashMap( - FullInstantiator.parameterArrayEquality - ); - - for (const values of this.resolveInputs(lookup.instantiation)) { - const provenance: VirtualSourceRow[] = []; - for (const value of values) { - provenance.push(...value.provenance); - } - - const directValues = withoutProvenance(values); - pendingLookups.setOrUpdate(directValues, (old) => { - if (old == null) { - return { - lookup: ScopedParameterLookup.normalized(scope, UnscopedParameterLookup.normalized(directValues)), - provenancePaths: [provenance], - resultSet: Symbol(`lookup ${stage}.${index}`) - }; - } else { - old.provenancePaths.push(provenance); - return old; - } - }); - } - - const lookupsToProvenance = new Map(); - for (const [_, pending] of pendingLookups.entries) { - lookupsToProvenance.set(pending.lookup, pending); - } - - const outputs = await this.input.source.getParameterSets( - [...lookupsToProvenance.keys()], - `Stream ${this.evaluators.stream.name} evaluating parameter on ${resolvedLookup.sourceTable.tablePattern}` - ); - - const values = outputs.flatMap(({ lookup, rows }) => { - const { provenancePaths, resultSet } = lookupsToProvenance.get(lookup)!; - return provenancePaths.flatMap((origin, provenanceIndex) => { - return rows.map((row, rowid) => { - const length = Object.entries(row).length; - const asArray: ParameterValueWithRow[] = []; + override clone(cloneParameter: CloneParameter): ParameterIndexExpandingLookup { + const lookup = new ParameterIndexExpandingLookup( + this.resultSetIndex, + this.lookup, + this.instantiation.map(cloneParameter) + ); + lookup.wasResolved = this.wasResolved; + return lookup; + } +} - for (let i = 0; i < length; i++) { - // Stream parameters generate an output row like {0: , 1: , ...}. - const value = row[i.toString()] as SqliteParameterValue; - - // All paths share one result set because the lookup was deduplicated. Include the path index in the row - // identity because the same output rows are instantiated once for every path. - const directOrigin = length > 1 ? { resultSet, row: provenanceIndex * rows.length + rowid } : undefined; - - asArray.push({ - value, - // Note: Not tracking provenance for parameters with just a single output is purely a performance - // optimization. If there's just a single value, we don't need to correlate it with other columns in the - // row. Not adding provenance saves some work in mergeValueCombinations. - provenance: directOrigin != null ? [...origin, directOrigin] : origin, - directOrigin - }); - } - return asArray; - }); - }); - }); +class LookupStage { + constructor( + /** + * Lookups that only have dependencies on prior stages. + */ + readonly lookups: PreparedExpandingLookup[], + /** + * A list of constraints enforcing that specific columns must have equal values. + * + * These constraints are evaluated after lookups, and may reference lookups in this stage. + */ + readonly intersections: RequiredIntersection[] + ) {} - this.evaluators.lookupStages[stage][index] = { type: 'cached', values }; - return values; + *inputParameters() { + for (const intersection of this.intersections) { + yield* intersection.values; } - const other = this.expandingLookupSync(stage, index); - if (other == null) { - throw new Error('internal error: Unable to resolve non-parameter lookup synchronously?'); + for (const lookup of this.lookups) { + if (lookup instanceof ParameterIndexExpandingLookup) { + yield* lookup.instantiation; + } } - return other; } - private static readonly parameterArrayEquality = listEquality(StableHasher.parameterValueEquality); + clone(cloneParameter: CloneParameter): LookupStage { + return new LookupStage( + this.lookups.map((l) => l.clone(cloneParameter)), + this.intersections.map(({ values, wasApplied }) => ({ values: values.map(cloneParameter), wasApplied })) + ); + } } -export type PreparedExpandingLookup = - | { type: 'parameter'; lookup: ParameterIndexLookupCreator; instantiation: PreparedParameterValue[] } - | { type: 'table_valued'; read(request: RequestParameters): SqliteParameterValue[][] } - | { type: 'cached'; values: ParameterValueWithRow[][] }; +interface RequiredIntersection { + values: PreparedParameterValue[]; + wasApplied: boolean; +} /** * A {@link plan.ParameterValue} that can be evaluated against request parameters. * - * Additionally, this includes the `cached` variant which allows partially instantiating parameters. + * Additionally, this includes the `static` variant which allows partially instantiating parameters. */ -export type PreparedParameterValue = - | { type: 'request'; read(request: RequestParameters): SqliteValue } - | { type: 'lookup'; lookup: { stage: number; index: number }; resultIndex: number } - | { type: 'intersection'; values: PreparedParameterValue[] } - | { type: 'cached'; values: ParameterValueWithRow[] }; +export type PreparedParameterValue = RequestParameterValue | LookupParameterValue; -interface ParameterValueWithRow { - value: SqliteParameterValue; +class RequestParameterValue { + #resolved: SqliteParameterValue | undefined; - /** - * Information on how this value was resolved. - * - * We track how a parameter value was resolved to be able to merge parameters correctly. A Sync Stream with multiple - * independent parameters generates their cartesian product as buckets. When multiple parameters are resolved from the - * same row though, we can't use the full cartesian product. As an example, consider this stream: - * - * ```sql - * SELECT products.* FROM products, stores - * WHERE stores.name = products.store_name - * AND stores.region = products.region - * AND stores.id = subscription.parameter('store'); - * ``` - * - * Here, the bucket shape consists of two parameters (`store_name` and `region`). But since they're derived from the - * same `stores` row, we can't combine them freely. For each parameter, `store_name` and `region` must come from the - * same row. Here, the `provenance` would have a single entry and both parameters would have the same - * {@link VirtualSourceRow.resultSet}. - * - * For static values, such as constants or scalar values derived from request parameters, this array is empty. It's - * also possible for this to contain more than one entry, though: - * - * 1. For intersection values, we track the provenance of all input values. - * 2. It's possible to nest parameters. For instance, in the query `SELECT data.* FROM data, a, b WHERE a.a = data.a - * AND b.b = a.b AND data.c = b.c`, we have to parameters (`a` and `c`). `a` can be resolved from a lookup in - * table `a`, but we need to go through a second lookup to resolve `c`. Here, we can only combine value `c` with - * value `a` if the two were derived from the same row in `a`. So, the row `b` derived through `a` would include - * provenance elements of row `a` here. - */ - provenance: VirtualSourceRow[]; - // If set, must be contained in provenance - directOrigin?: VirtualSourceRow; -} + constructor(private readonly read: (request: RequestParameters) => SqliteValue) {} -interface VirtualSourceRow { - /** - * An opaque identifier for the result set this row was derived from. - */ - resultSet: symbol; - /** - * A number uniquely identifying this row in its result set. - */ - row: number; + requireResolved() { + if (this.#resolved === undefined) throw new Error('Expected request values to be resolved here'); + return this.#resolved; + } + + resolveWith({ request }: PartialInstantiationInput): SqliteParameterValue { + if (this.#resolved !== undefined) return this.#resolved; + + const value = this.read(request); + if (isValidParameterValue(value)) { + return (this.#resolved = value); + } else { + throw uninstantiableException; + } + } + + clone(): RequestParameterValue { + const clone = new RequestParameterValue(this.read); + clone.#resolved = this.#resolved; + return clone; + } } -function withoutProvenance(source: ParameterValueWithRow[]): SqliteParameterValue[] { - return source.map(({ value }) => value); +class LookupParameterValue implements ResultSetColumn { + constructor( + readonly lookup: PreparedExpandingLookup, + readonly outputIndex: number + ) {} + + clone(): LookupParameterValue { + return new LookupParameterValue(this.lookup, this.outputIndex); + } } export interface PartialInstantiationInput { @@ -692,63 +579,11 @@ function* filterParameterRows(rows: SqliteValue[][]): Generator { - // Partial backtracking results, the current instantiation is fixed for 0..nextParameter in generateCombinations. - const partialResults = new Array(valuesByParameter.length); - // A map from result sets to rows used in the partial instantiation. - const usedRows = new Map(); - - function installRowIfNoConflict(value: ParameterValueWithRow): [boolean, symbol[]] { - const addedResultSets: symbol[] = []; - - for (const origin of value.provenance) { - const { resultSet, row } = origin; - const existingRow = usedRows.get(resultSet); - if (existingRow === undefined) { - addedResultSets.push(resultSet); - usedRows.set(resultSet, row); - } else if (existingRow == row) { - continue; - } else { - // The current instantiation already contains a value from the same result set but derived from a different - // row. So we must ignore this parameter value. - return [false, addedResultSets]; - } - } - - return [true, addedResultSets]; - } - - function uninstallResultSets(resultSets: symbol[]) { - for (const rs of resultSets) { - usedRows.delete(rs); - } - } - - function* generateCombinations(nextParameter: number): Generator { - if (nextParameter >= valuesByParameter.length) { - yield [...partialResults]; - return; - } +/* - const availableValues = valuesByParameter[nextParameter]; - for (const available of availableValues) { - const [canUse, addedResultSets] = installRowIfNoConflict(available); - if (canUse) { - partialResults[nextParameter] = available; - yield* generateCombinations(nextParameter + 1); - } +Intersection notes: - uninstallResultSets(addedResultSets); - } - } +Intersection entirely on request parameters? Add to static request filter, pick any. - yield* generateCombinations(0); -} +Intersection with at least one lookup output? Add as pre-condition to stage where the output is used. + */ diff --git a/packages/sync-rules/src/sync_plan/evaluator/result_set.ts b/packages/sync-rules/src/sync_plan/evaluator/result_set.ts new file mode 100644 index 000000000..264bf8b76 --- /dev/null +++ b/packages/sync-rules/src/sync_plan/evaluator/result_set.ts @@ -0,0 +1,216 @@ +import { HashMap, listEquality, StableHasher } from '../../compiler/equality.js'; +import { SqliteParameterValue } from '../../types.js'; + +/** + * A mutable result set of parameter results. + * + * This is used to represent parameter results when resolving buckets: Each expanding lookup is joined onto a pending + * result set until all lookups have been applied. Once all result sets have been added, bucket parameters can be read + * by reading columns in each row. + */ +export class ResultSet { + #totalLookups: number; + #rows: ResultSetRow[]; + + /** + * @param totalLookups - The total amount of lookups that will be joined to this result set. + */ + constructor(totalLookups: number) { + this.#totalLookups = totalLookups; + const initialRow = new Array(totalLookups); + initialRow.fill(undefined); + this.#rows = [initialRow]; + } + + get length(): number { + return this.#rows.length; + } + + clone(): ResultSet { + const rs = new ResultSet(this.#totalLookups); + rs.#rows.splice(0, 1); // Remove the initial unit row + + for (const row of this.#rows) { + // We can shallow-clone rows, inner items are frozen once added into the result set. + rs.#rows.push(row.slice()); + } + return rs; + } + + /** + * Extracts unique values by looking values for each column in this result set. + */ + *projectUnique(columns: ResultSetColumn[]): Iterable { + for (const { group, first } of this.#groupBy(columns, (values) => values)) { + if (first) { + yield group; + } + } + } + + /** + * Adds a new result set by forming the cartesian product with the given values. + */ + multiply(resultSetIndex: number, rows: SqliteParameterValue[][]) { + if (rows.length === 0) { + this.#rows = []; + } + + const originalLength = this.#rows.length; + for (let i = 0; i < originalLength; i++) { + this.#multiplyAtRow(resultSetIndex, i, rows); + } + } + + /** + * @param keys - Join keys that are already present in the result set. + * @param resultSetIndex - The index of the resl set being joined. + * @param performLookup - Adds resolved rows to each unique instantiation of join keys. + */ + async joinAsync( + keys: ResultSetColumn[], + resultSetIndex: number, + performLookup: (lookups: AsyncJoinLookup[]) => Promise + ) { + const lookupsByRow: AsyncJoinLookup[] = []; + const uniqueLookups: AsyncJoinLookup[] = []; + + for (const { group, first } of this.#groupBy(keys, (values) => ({ inputs: values, foundRows: [] }))) { + if (first) uniqueLookups.push(group); + lookupsByRow.push(group); + } + + await performLookup(uniqueLookups); + + const deletedRows: number[] = []; + const originalLength = this.#rows.length; + for (let i = 0; i < originalLength; i++) { + const lookup = lookupsByRow[i]; + if (lookup.foundRows.length > 0) { + this.#multiplyAtRow(resultSetIndex, i, lookup.foundRows); + } else { + // The row has no matching join partner, so remove it. We can't split it immediately because #multiplyAtRow is + // still iterating through rows. + deletedRows.push(i); + } + } + + let offset = 0; + for (const toDelete of deletedRows) { + this.#rows.splice(toDelete - offset, 1); + offset++; + } + } + + /** + * Removes rows where the given columns have different values. + * + * If a fixed value is passed, this also removes rows where any of the given columns has a different value. + */ + formIntersection(columns: ResultSetColumn[], fixedValue?: SqliteParameterValue) { + const keptRows: ResultSetRow[] = []; + + row: for (const row of this.#rows) { + let requiredValue = fixedValue; + + for (const column of columns) { + const evaluated = lookupInRow(row, column); + if (requiredValue !== undefined && evaluated !== requiredValue) { + // Intersection doesn't match, skip this row. + continue row; + } + + requiredValue = evaluated; + } + + keptRows.push(row); + } + + this.#rows = keptRows; + } + + #multiplyAtRow(resultSetIndex: number, rowIndex: number, rows: SqliteParameterValue[][]) { + // Add first element of product to existing row, remaining as new rows. + const row = this.#rows[rowIndex]; + row[resultSetIndex] = Object.freeze(rows[0]); + + for (let j = 1; j < rows.length; j++) { + const copy = row.slice(); + copy[resultSetIndex] = Object.freeze(rows[j]); + this.#rows.push(copy); + } + } + + *#groupBy(columns: ResultSetColumn[], generateGroup: (values: SqliteParameterValue[]) => T) { + const originalLength = this.#rows.length; + + if (columns.length === 1) { + // Fast path, we can use native sets. + const [column] = columns; + const foundValues = new Map(); + + for (let i = 0; i < originalLength; i++) { + const row = this.#rows[i]; + const value = lookupInRow(row, column); + const existingGroup = foundValues.get(value); + + if (existingGroup != null) { + yield { group: existingGroup, first: false }; + } else { + const group = generateGroup([value]); + foundValues.set(value, group); + yield { group, first: true }; + } + } + } else { + const foundValues = new HashMap(parameterArrayEquality); + + for (let i = 0; i < originalLength; i++) { + const row = this.#rows[i]; + const values = columns.map((c) => lookupInRow(row, c)); + + let isFirst = false; + const group = foundValues.putIfAbsent(values, () => { + isFirst = true; + return generateGroup(values); + }); + + yield { group, first: isFirst }; + } + } + } +} + +export interface ResultSetElement { + resultSetIndex: number; +} + +export interface ResultSetColumn { + lookup: ResultSetElement; + outputIndex: number; +} + +export interface AsyncJoinLookup { + inputs: SqliteParameterValue[]; + foundRows: SqliteParameterValue[][]; +} + +/** + * A row in a result set. + * + * While this is semantically a list of columns, that representation would require a lot of copying on each join. + * So, we represent each lookup result as an array of values (that we can re-use when we create new rows for joins). + * Result sets that have not yet been processed are represented as undefined. + */ +type ResultSetRow = (ReadonlyArray | undefined)[]; + +function lookupInRow(row: ResultSetRow, column: ResultSetColumn): SqliteParameterValue { + const valuesForResultSet = row[column.lookup.resultSetIndex]; + if (valuesForResultSet === undefined) { + throw new Error('Tried to lookup values set before it was joined to result set'); + } + + return valuesForResultSet[column.outputIndex]; +} + +const parameterArrayEquality = listEquality(StableHasher.parameterValueEquality); diff --git a/packages/sync-rules/test/src/sync_plan/evaluator/evaluator.test.ts b/packages/sync-rules/test/src/sync_plan/evaluator/evaluator.test.ts index e1ee6837c..1b70011ed 100644 --- a/packages/sync-rules/test/src/sync_plan/evaluator/evaluator.test.ts +++ b/packages/sync-rules/test/src/sync_plan/evaluator/evaluator.test.ts @@ -550,6 +550,32 @@ streams: expect(querier.staticBuckets.map((e) => e.bucket)).toStrictEqual(['stream|0["user"]']); }); + syncTest('intersection of request data', ({ sync }) => { + const desc = sync.prepareSyncStreams(` +config: + edition: 3 + +streams: + stream: + auto_subscribe: true + query: SELECT * FROM issues WHERE a = auth.parameter('x') AND a IN auth.parameter('y') +`); + + function queryWith(x: string, y: string[]) { + const { querier, errors } = desc.getBucketParameterQuerier({ + globalParameters: requestParameters({ sub: 'user', x, y }), + hasDefaultStreams: true, + streams: {} + }); + expect(errors).toStrictEqual([]); + expect(querier.hasDynamicBuckets).toStrictEqual(false); + return querier.staticBuckets.map((e) => e.bucket); + } + + expect(queryWith('p1', ['p2'])).toStrictEqual([]); + expect(queryWith('p1', ['p1', 'p2'])).toStrictEqual(['stream|0["p1"]']); + }); + syncTest('parameter lookups', async ({ sync }) => { const desc = sync.prepareSyncStreams(` config: @@ -846,11 +872,7 @@ streams: // Duplicates do not need to be removed here, but they must not make lookup // columns independent and create impossible pairs like [2, "A"]. - expect(dynamicBuckets.map((bucket) => bucket.bucket)).toStrictEqual([ - 'stream|0[1,"A"]', - 'stream|0[1,"A"]', - 'stream|0[2,"B"]' - ]); + expect(dynamicBuckets.map((bucket) => bucket.bucket)).toStrictEqual(['stream|0[1,"A"]', 'stream|0[2,"B"]']); }); syncTest('preserves correlation across bigint lookup output columns', async ({ sync }) => { @@ -1103,8 +1125,8 @@ streams: expect(querier.staticBuckets.map((e) => e.bucket)).toStrictEqual([ 'stream|0["a1","b1"]', - 'stream|0["a1","b2"]', 'stream|0["a2","b1"]', + 'stream|0["a1","b2"]', 'stream|0["a2","b2"]' ]); }); diff --git a/packages/sync-rules/test/src/sync_plan/evaluator/result_set.test.ts b/packages/sync-rules/test/src/sync_plan/evaluator/result_set.test.ts new file mode 100644 index 000000000..27490cf55 --- /dev/null +++ b/packages/sync-rules/test/src/sync_plan/evaluator/result_set.test.ts @@ -0,0 +1,223 @@ +import { describe, expect, test } from 'vitest'; +import { ResultSet, ResultSetElement } from '../../../../src/sync_plan/evaluator/result_set.js'; +import { SqliteParameterValue } from '../../../../src/types.js'; + +describe('ResultSet', () => { + test('unit result set', () => { + const empty = new ResultSet(0); + + expect(empty.length).toStrictEqual(1); + expect([...empty.projectUnique([])]).toStrictEqual([[]]); + }); + + describe('projectUnique', () => { + const rs = new ResultSet(1); + rs.multiply(0, [ + ['a1', 'b1'], + ['a2', 'b2'], + ['a2', 'b1'], + ['a2', 'b2'] + ]); + + test('single column', () => { + expect([...rs.projectUnique([{ lookup: element(0), outputIndex: 0 }])]).toStrictEqual([['a1'], ['a2']]); + + expect([...rs.projectUnique([{ lookup: element(0), outputIndex: 1 }])]).toStrictEqual([['b1'], ['b2']]); + }); + + test('multiple columns', () => { + expect([ + ...rs.projectUnique([ + { lookup: element(0), outputIndex: 0 }, + { lookup: element(0), outputIndex: 1 } + ]) + ]).toStrictEqual([ + ['a1', 'b1'], + ['a2', 'b2'], + ['a2', 'b1'] + ]); + }); + }); + + describe('multiply', () => { + test('empty', () => { + const rs = new ResultSet(1); + expect(rs.length).toStrictEqual(1); + + rs.multiply(0, []); + expect(rs.length).toStrictEqual(0); + }); + + test('is cartesian product', () => { + const rs = new ResultSet(2); + + rs.multiply(0, [['a'], ['b']]); + rs.multiply(1, [[0], [1]]); + expect([ + ...rs.projectUnique([ + { + lookup: element(0), + outputIndex: 0 + }, + { + lookup: element(1), + outputIndex: 0 + } + ]) + ]).toStrictEqual([ + ['a', 0], + ['b', 0], + ['a', 1], + ['b', 1] + ]); + }); + }); + + describe('formIntersection', () => { + const col0 = { lookup: element(0), outputIndex: 0 }; + const col1 = { lookup: element(1), outputIndex: 0 }; + + test('with a fixed value, removes rows where any column differs from it', () => { + const rs = new ResultSet(2); + rs.multiply(0, [['a'], ['b']]); + rs.multiply(1, [['a'], ['b']]); + + rs.formIntersection([col0, col1], 'a'); + + expect([...rs.projectUnique([col0, col1])]).toStrictEqual([['a', 'a']]); + }); + + test('without a fixed value, removes rows where the columns differ from each other', () => { + const rs = new ResultSet(2); + rs.multiply(0, [['a'], ['b']]); + rs.multiply(1, [['a'], ['b']]); + + rs.formIntersection([col0, col1]); + + expect([...rs.projectUnique([col0, col1])]).toStrictEqual([ + ['a', 'a'], + ['b', 'b'] + ]); + }); + }); + + describe('joinAsync', () => { + const col0 = { lookup: element(0), outputIndex: 0 }; + const col1 = { lookup: element(1), outputIndex: 0 }; + + test('expands each row with matching values, deduplicating lookups', async () => { + const rs = new ResultSet(2); + rs.multiply(0, [['a'], ['b'], ['a']]); + + const matches: Record = { + a: [[1], [2]], + b: [[3]] + }; + + let lookupCount = 0; + await rs.joinAsync([col0], 1, async (lookups) => { + // The lookup for 'a' is only performed once, even though it's shared by two rows. + expect(lookups.length).toStrictEqual(2); + lookupCount++; + + for (const lookup of lookups) { + lookup.foundRows.push(...matches[lookup.inputs[0] as string]); + } + }); + + expect(lookupCount).toStrictEqual(1); + expect([...rs.projectUnique([col0, col1])]).toStrictEqual([ + ['a', 1], + ['b', 3], + ['a', 2] + ]); + }); + + test('removes rows without a matching join partner', async () => { + const rs = new ResultSet(2); + rs.multiply(0, [['a'], ['b'], ['c'], ['a']]); + + const matches: Record = { + a: [[10]], + b: [], + c: [[30], [31]] + }; + + await rs.joinAsync([col0], 1, async (lookups) => { + for (const lookup of lookups) { + lookup.foundRows.push(...matches[lookup.inputs[0] as string]); + } + }); + + expect(rs.length).toStrictEqual(4); + expect([...rs.projectUnique([col0, col1])]).toStrictEqual([ + ['a', 10], + ['c', 30], + ['c', 31] + ]); + }); + + test('removing all rows results in an empty result set', async () => { + const rs = new ResultSet(2); + rs.multiply(0, [['a'], ['b']]); + + await rs.joinAsync([col0], 1, async () => { + // Leave foundRows empty for every lookup. + }); + + expect(rs.length).toStrictEqual(0); + expect([...rs.projectUnique([col0, col1])]).toStrictEqual([]); + }); + + test('supports composite join keys', async () => { + const rs = new ResultSet(2); + rs.multiply(0, [ + [1, 'x'], + [1, 'y'], + [2, 'x'] + ]); + + const colKey0 = { lookup: element(0), outputIndex: 0 }; + const colKey1 = { lookup: element(0), outputIndex: 1 }; + + const matches: Record = { + '1,x': [[100]], + '1,y': [[200]], + '2,x': [[300], [301]] + }; + + await rs.joinAsync([colKey0, colKey1], 1, async (lookups) => { + for (const lookup of lookups) { + lookup.foundRows.push(...matches[lookup.inputs.join(',')]); + } + }); + + expect([...rs.projectUnique([colKey0, colKey1, col1])]).toStrictEqual([ + [1, 'x', 100], + [1, 'y', 200], + [2, 'x', 300], + [2, 'x', 301] + ]); + }); + + test('empty join key', async () => { + const rs = new ResultSet(2); + rs.multiply(0, [['a'], ['b']]); + + await rs.joinAsync([], 1, async (lookups) => { + expect(lookups).toMatchObject([{ inputs: [] }]); + + lookups[0].foundRows.push(['x']); + }); + + expect([...rs.projectUnique([col0, col1])]).toStrictEqual([ + ['a', 'x'], + ['b', 'x'] + ]); + }); + }); +}); + +function element(index: number): ResultSetElement { + return { resultSetIndex: index }; +}