diff --git a/.changeset/compiled-event-plans.md b/.changeset/compiled-event-plans.md new file mode 100644 index 000000000..44abe1149 --- /dev/null +++ b/.changeset/compiled-event-plans.md @@ -0,0 +1,6 @@ +--- +'@powersync/service-sync-rules': minor +'@powersync/service-core': minor +--- + +Compile replication events for every sync-config edition into additive serialized sync-plan data, expose behavioral equality for compiled event models, and normalize legacy sidecar events when loading older plans while preserving raw SQL for older services. diff --git a/modules/module-mongodb-storage/src/storage/implementation/MongoBucketBatch.ts b/modules/module-mongodb-storage/src/storage/implementation/MongoBucketBatch.ts index 5ec3d8eaf..d6bbcd0fc 100644 --- a/modules/module-mongodb-storage/src/storage/implementation/MongoBucketBatch.ts +++ b/modules/module-mongodb-storage/src/storage/implementation/MongoBucketBatch.ts @@ -1,5 +1,5 @@ import { mongo } from '@powersync/lib-service-mongodb'; -import { HydratedSyncConfig, SqlEventDescriptor, SqliteRow, SqliteValue } from '@powersync/service-sync-rules'; +import { HydratedEventDescriptor, HydratedSyncConfig, SqliteRow, SqliteValue } from '@powersync/service-sync-rules'; import * as bson from 'bson'; import { @@ -934,9 +934,9 @@ export abstract class MongoBucketBatch } /** - * Gets relevant {@link SqlEventDescriptor}s for the given {@link SourceTable} + * Gets relevant {@link HydratedEventDescriptor}s for the given {@link SourceTable} */ - protected getTableEvents(table: storage.SourceTable): SqlEventDescriptor[] { + protected getTableEvents(table: storage.SourceTable): HydratedEventDescriptor[] { return this.sync_rules.eventDescriptors.filter((evt) => [...evt.getSourceTables()].some((sourceTable) => sourceTable.matches(table.ref)) ); diff --git a/modules/module-postgres-storage/src/storage/batch/PostgresBucketBatch.ts b/modules/module-postgres-storage/src/storage/batch/PostgresBucketBatch.ts index d3e883d8f..8e4809610 100644 --- a/modules/module-postgres-storage/src/storage/batch/PostgresBucketBatch.ts +++ b/modules/module-postgres-storage/src/storage/batch/PostgresBucketBatch.ts @@ -1334,10 +1334,10 @@ export class PostgresBucketBatch } /** - * Gets relevant {@link SqlEventDescriptor}s for the given {@link SourceTable} + * Gets relevant {@link HydratedEventDescriptor}s for the given {@link SourceTable} * TODO maybe share this with an abstract class */ - protected getTableEvents(table: storage.SourceTable): sync_rules.SqlEventDescriptor[] { + protected getTableEvents(table: storage.SourceTable): sync_rules.HydratedEventDescriptor[] { return this.sync_rules.eventDescriptors.filter((evt) => [...evt.getSourceTables()].some((sourceTable) => sourceTable.matches(table.ref)) ); diff --git a/packages/service-core/src/storage/BucketStorageFactory.ts b/packages/service-core/src/storage/BucketStorageFactory.ts index 97adff689..398bbd9c8 100644 --- a/packages/service-core/src/storage/BucketStorageFactory.ts +++ b/packages/service-core/src/storage/BucketStorageFactory.ts @@ -180,11 +180,10 @@ export interface SerializedSyncPlan { plan: RawSerializedSyncPlan; compatibility: SerializedCompatibilityContext; /** - * Event descriptors are not currently represented in the sync plan because they don't use the sync streams compiler - * yet. + * Raw event SQL persisted as a compatibility mirror for compiled {@link plan} events. * - * We might revisit that in the future, but for now we store SQL text of their definitions here to be able to restore - * them. + * Compiled events are an additive plan field. Older services ignore that field and restore these descriptors through + * the legacy evaluator. Keep dual-writing this field until a future plan version explicitly removes that support. */ eventDescriptors: Record; errors?: ReplicationError[]; @@ -212,13 +211,14 @@ export function updateSyncRulesFromConfig( const { config, errors } = parsed; if (config instanceof PrecompiledSyncConfig) { const eventDescriptors: Record = {}; - for (const event of config.eventDescriptors) { - eventDescriptors[event.name] = event.sourceQueries.map((q) => q.sql); + for (const event of config.plan.events) { + eventDescriptors[event.name] = event.sourceQueries.map((query) => query.sql); } plan = { compatibility: config.compatibility.serialize(), plan: serializeSyncPlan(config.plan), + // Dual-write raw SQL so older services can ignore additive compiled plan events without losing event behavior. eventDescriptors, errors: errors.map((e) => syncConfigYamlErrorToReplicationError(e)) }; diff --git a/packages/service-core/src/storage/PersistedSyncConfigContent.ts b/packages/service-core/src/storage/PersistedSyncConfigContent.ts index 6ee3fc5a7..f98c43941 100644 --- a/packages/service-core/src/storage/PersistedSyncConfigContent.ts +++ b/packages/service-core/src/storage/PersistedSyncConfigContent.ts @@ -2,6 +2,7 @@ import { logger as defaultLogger, ErrorCode, ServiceError } from '@powersync/lib import { CompatibilityContext, CompatibilityOption, + compileEventDefinitions, DEFAULT_HYDRATION_STATE, deserializeSyncPlan, ErrorLocation, @@ -9,7 +10,6 @@ import { HydrationState, nodeSqlite, PrecompiledSyncConfig, - SqlEventDescriptor, SqlSyncRules, SyncConfigWithErrors, versionedHydrationState, @@ -40,17 +40,23 @@ export function parsePersistedSyncConfigContent(options: ParsePersistedSyncConfi const plan = deserializeSyncPlan(compiledPlan.plan); const compatibility = CompatibilityContext.deserialize(compiledPlan.compatibility); - const eventDefinitions: SqlEventDescriptor[] = []; - for (const [name, queries] of Object.entries(compiledPlan.eventDescriptors)) { - const descriptor = new SqlEventDescriptor(name, compatibility); - for (const query of queries) { - descriptor.addSourceQuery(query, parseOptions); + const errors: YamlError[] = []; + // Compiled events are additive to plan versions 1 and 2. New readers prefer them when present; when an older plan + // does not contain them, normalize the dual-written raw SQL at this loading boundary. This keeps legacy event + // evaluators out of PrecompiledSyncConfig while older binaries can continue reading the same persisted config. + if (compiledPlan.plan.events == null) { + const normalized = compileEventDefinitions(compiledPlan.eventDescriptors, parseOptions); + const fatalErrors = normalized.errors.filter((error) => error.type == 'fatal'); + if (fatalErrors.length != 0) { + throw new Error( + `Failed to compile persisted replication events: ${fatalErrors.map((error) => error.message).join(', ')}` + ); } - - eventDefinitions.push(descriptor); + plan.events = normalized.events; + errors.push(...normalized.errors.map((error) => new YamlError(error))); } - const precompiled = new PrecompiledSyncConfig(plan, compatibility, eventDefinitions, { + const precompiled = new PrecompiledSyncConfig(plan, compatibility, { defaultSchema: parseOptions.defaultSchema, sourceText: content }); @@ -59,7 +65,6 @@ export function parsePersistedSyncConfigContent(options: ParsePersistedSyncConfi // This means asUpdateOptions will not change the storage version, even if the default changes. precompiled.storageVersion = storageVersion; - const errors: YamlError[] = []; if (compiledPlan.errors) { for (const error of compiledPlan.errors) { const location: ErrorLocation | undefined = error.location && { diff --git a/packages/service-core/src/storage/ReplicationEventPayload.ts b/packages/service-core/src/storage/ReplicationEventPayload.ts index 38cddbc24..31229aab1 100644 --- a/packages/service-core/src/storage/ReplicationEventPayload.ts +++ b/packages/service-core/src/storage/ReplicationEventPayload.ts @@ -11,6 +11,6 @@ export type EventData = { export type ReplicationEventPayload = { batch: BucketStorageBatch; data: EventData; - event: sync_rules.SqlEventDescriptor; + event: sync_rules.HydratedEventDescriptor; table: SourceTable; }; diff --git a/packages/service-core/test/src/PersistedSyncConfigContent.test.ts b/packages/service-core/test/src/PersistedSyncConfigContent.test.ts new file mode 100644 index 000000000..693b623e0 --- /dev/null +++ b/packages/service-core/test/src/PersistedSyncConfigContent.test.ts @@ -0,0 +1,99 @@ +import { + DEFAULT_HYDRATION_STATE, + DEFAULT_TAG, + nodeSqlite, + PrecompiledSyncConfig, + serializeSyncPlan, + SqlSyncRules +} from '@powersync/service-sync-rules'; +import * as sqlite from 'node:sqlite'; +import { describe, expect, test } from 'vitest'; +import { + parsePersistedSyncConfigContent, + SerializedSyncPlan, + updateSyncRulesFromConfig +} from '../../src/storage/storage-index.js'; + +const EVENT_QUERY = 'SELECT user_id, checkpoint FROM checkpoints WHERE active = true'; +const yamlWithoutEvents = ` +config: + edition: 3 + +streams: + checkpoints: + query: SELECT * FROM checkpoints +`; +const yamlWithEvents = `${yamlWithoutEvents} +event_definitions: + write_checkpoints: + payloads: + - ${EVENT_QUERY} +`; + +describe('persisted compiled replication events', () => { + test('dual-writes additive compiled events and restores the compiled evaluator', () => { + const parsed = SqlSyncRules.fromYaml(yamlWithEvents, { defaultSchema: 'test_schema' }); + const update = updateSyncRulesFromConfig(parsed); + const compiled = update.config.plan!; + const originalEvent = compiled.plan.events![0]; + + expect(compiled.plan.version).toBeLessThanOrEqual(2); + expect(compiled.plan.events).toHaveLength(1); + expect(originalEvent.name).toBe('write_checkpoints'); + expect(compiled.eventDescriptors).toEqual({ write_checkpoints: [EVENT_QUERY] }); + + const restored = restore(compiled); + expect(restored.config).toBeInstanceOf(PrecompiledSyncConfig); + expect(restored.config).not.toHaveProperty('eventDescriptors'); + expect(restored.config.eventDefinitions).toHaveLength(1); + expect(restored.config.eventDefinitions[0].name).toBe('write_checkpoints'); + + const hydrated = restored.config.hydrate({ + hydrationState: DEFAULT_HYDRATION_STATE, + sqlite: nodeSqlite(sqlite) + }); + expect( + hydrated.eventDescriptors[0].evaluateRowWithErrors({ + sourceTable: { connectionTag: DEFAULT_TAG, schema: 'test_schema', name: 'checkpoints' }, + record: { user_id: 'user-1', checkpoint: 4n, active: 1 } + }) + ).toEqual({ result: { data: { user_id: 'user-1', checkpoint: 4n } }, errors: [] }); + + // This represents what an older compiler sees after ignoring the additive plan.events field. The raw descriptor + // mirror is normalized into the compiled representation by the new loading boundary. + const { events: _ignored, ...planWithoutCompiledEvents } = compiled.plan; + const legacyView = restore({ ...compiled, plan: planWithoutCompiledEvents }); + expect(legacyView.config).not.toHaveProperty('eventDescriptors'); + expect(legacyView.config.eventDefinitions).toHaveLength(1); + expect(legacyView.config.eventDefinitions[0].name).toBe('write_checkpoints'); + // Recompiling the raw SQL mirror produces the same serialized event plan. + const legacyEvent = serializeSyncPlan((legacyView.config as PrecompiledSyncConfig).plan).events![0]; + expect(legacyEvent).toEqual(originalEvent); + }); + + test('restores raw event descriptors attached to version 1 and 2 plans', () => { + const parsed = SqlSyncRules.fromYaml(yamlWithoutEvents, { defaultSchema: 'test_schema' }); + const update = updateSyncRulesFromConfig(parsed); + const legacy: SerializedSyncPlan = { + ...update.config.plan!, + eventDescriptors: { write_checkpoints: [EVENT_QUERY] } + }; + + expect(legacy.plan.version).toBeLessThanOrEqual(2); + const restored = restore(legacy); + expect(restored.config).not.toHaveProperty('eventDescriptors'); + expect(restored.config.eventDefinitions).toHaveLength(1); + expect((restored.config as PrecompiledSyncConfig).plan.events).toMatchObject([ + { name: 'write_checkpoints', sourceQueries: [{ sql: EVENT_QUERY }] } + ]); + }); +}); + +function restore(compiledPlan: SerializedSyncPlan) { + return parsePersistedSyncConfigContent({ + content: yamlWithEvents, + compiledPlan, + storageVersion: 1, + parseOptions: { defaultSchema: 'test_schema' } + }); +} diff --git a/packages/sync-rules/src/HydratedSyncConfig.ts b/packages/sync-rules/src/HydratedSyncConfig.ts index bbcbf0dd9..211bf90cd 100644 --- a/packages/sync-rules/src/HydratedSyncConfig.ts +++ b/packages/sync-rules/src/HydratedSyncConfig.ts @@ -8,6 +8,7 @@ import { EvaluationError, GetBucketParameterQuerierResult, GetQuerierOptions, + HydratedEventDescriptor, HydrateSyncConfigParams, HydrationInput, isEvaluatedParameters, @@ -21,7 +22,6 @@ import { QuerierError, ScopedEvaluateParameterRow, ScopedEvaluateRow, - SqlEventDescriptor, SqliteInputValue, SqliteValue, SyncConfig, @@ -52,7 +52,7 @@ export class HydratedSyncConfig { */ private bucketSources: HydratedBucketSource[] = []; - eventDescriptors: SqlEventDescriptor[] = []; + eventDescriptors: HydratedEventDescriptor[] = []; /** * Only a single compatibility context is supported across all merged SyncConfigs. @@ -118,7 +118,9 @@ export class HydratedSyncConfig { this.bucketParameterLookupSources ).evaluateParameterRow; - this.eventDescriptors = definitions.flatMap((definition) => definition.eventDescriptors); + this.eventDescriptors = definitions.flatMap((definition) => + definition.eventDefinitions.map((event) => event.createEvaluator(this.hydrationInput)) + ); if (definitions.length == 1) { this.#bucketSourceDefinitions = definitions[0].bucketSources; diff --git a/packages/sync-rules/src/SyncConfig.ts b/packages/sync-rules/src/SyncConfig.ts index fc3d50da0..f777c83f3 100644 --- a/packages/sync-rules/src/SyncConfig.ts +++ b/packages/sync-rules/src/SyncConfig.ts @@ -6,7 +6,7 @@ import { } from './BucketSource.js'; import { CompatibilityContext } from './compatibility.js'; import { YamlError } from './errors.js'; -import { SqlEventDescriptor } from './events/SqlEventDescriptor.js'; +import { EventDefinition } from './events/EventDescriptor.js'; import { HydratedSyncConfig } from './HydratedSyncConfig.js'; import { SourceTableRef } from './SourceTableRef.js'; import { TablePattern } from './TablePattern.js'; @@ -21,6 +21,8 @@ export abstract class SyncConfig { bucketDataSources: BucketDataSource[] = []; bucketParameterLookupSources: ParameterIndexLookupCreator[] = []; bucketSources: BucketSource[] = []; + /** Prepared event definitions. Executable event descriptors only exist on {@link HydratedSyncConfig}. */ + eventDefinitions: EventDefinition[] = []; compatibility: CompatibilityContext = CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY; /** * If not defined, the storage module picks the latest stable version. @@ -28,7 +30,6 @@ export abstract class SyncConfig { * Only supported storage versions can be set here when parsing from yaml. */ storageVersion: number | undefined; - eventDescriptors: SqlEventDescriptor[] = []; /** * The (YAML-based) source contents from which this sync config has been derived. @@ -71,9 +72,9 @@ export abstract class SyncConfig { sourceTables.set(r.key(), r); } } - for (const event of this.eventDescriptors) { - for (const r of event.getSourceTables()) { - sourceTables.set(r.key(), r); + for (const event of this.eventDefinitions) { + for (const table of event.getSourceTables()) { + sourceTables.set(table.key(), table); } } } @@ -87,11 +88,9 @@ export abstract class SyncConfig { getEventTables(): TablePattern[] { const eventTables = new Map(); - if (this.eventDescriptors) { - for (const event of this.eventDescriptors) { - for (const r of event.getSourceTables()) { - eventTables.set(r.key(), r); - } + for (const event of this.eventDefinitions) { + for (const table of event.getSourceTables()) { + eventTables.set(table.key(), table); } } @@ -99,7 +98,7 @@ export abstract class SyncConfig { } tableTriggersEvent(table: SourceTableRef): boolean { - return this.eventDescriptors.some((bucket) => bucket.tableTriggersEvent(table)); + return this.eventDefinitions.some((event) => event.tableTriggersEvent(table)); } tableSyncsData(table: SourceTableRef): boolean { diff --git a/packages/sync-rules/src/compiler/compiler.ts b/packages/sync-rules/src/compiler/compiler.ts index 13735744c..c11b6a6ba 100644 --- a/packages/sync-rules/src/compiler/compiler.ts +++ b/packages/sync-rules/src/compiler/compiler.ts @@ -1,16 +1,19 @@ import { NodeLocation, parse, PGNode, Statement } from 'pgsql-ast-parser'; -import { StreamOptions, SyncPlan } from '../sync_plan/plan.js'; +import { SqlRuleError } from '../errors.js'; +import { CompiledEventDescriptor, StreamOptions, SyncPlan } from '../sync_plan/plan.js'; import { SourceSchema } from '../types.js'; import { StreamResolver } from './bucket_resolver.js'; import { DangerousParameterDetector } from './detect_dangerous_parameters.js'; -import { HashSet } from './equality.js'; -import { NodeLocations } from './expression.js'; +import { Equality, HashSet } from './equality.js'; +import { expressionBehaviorIdentity, NodeLocations } from './expression.js'; +import { RowExpression, SingleDependencyExpression } from './filter.js'; import { CompilerModelToSyncPlan } from './ir_to_sync_plan.js'; import { StreamQueryParser } from './parser.js'; import { QuerierGraphBuilder } from './querier_graph.js'; -import { PointLookup, RowEvaluator } from './rows.js'; +import { EventRowEvaluator, ExpressionColumnSource, PointLookup, RowEvaluator, StarColumnSource } from './rows.js'; import { SqlScope } from './scope.js'; import { CommonTableExpression, PreparedSubquery } from './sqlite.js'; +import { PhysicalSourceResultSet } from './table.js'; export interface SyncStreamsCompilerOptions { /** @@ -34,21 +37,84 @@ export interface ParseStreamOptions extends StreamOptions { warnOnDangerousParameter: boolean; } +export interface CompiledEvent { + name: string; + sourceQueries: CompiledEventSourceQueryModel[]; +} + +export interface CompiledEventSourceQueryModel { + sql: string; + sourceTable: PhysicalSourceResultSet; + variants: EventRowEvaluator[]; +} + +/** + * Behavioral equality for replication events in the compiler's JavaScript plan model. + * + * Raw SQL and payload-query order are deliberately ignored. The row evaluators compare the tables, filters and + * projected payload through the same compatibility model used while compiling stream plans. + */ +export const compiledEventDefinitionEquality: Equality = { + hash(hasher, value) { + hasher.addString(compiledEventDefinitionIdentity(value)); + }, + equals(a, b) { + return a === b || compiledEventDefinitionIdentity(a) == compiledEventDefinitionIdentity(b); + } +}; + +function compiledEventDefinitionIdentity(event: CompiledEvent): string { + return JSON.stringify([ + event.name, + event.sourceQueries + .map((query) => + JSON.stringify([ + tablePatternIdentity(query.sourceTable), + query.variants.map(compiledEventVariantIdentity).sort() + ]) + ) + .sort() + ]); +} + +function compiledEventVariantIdentity(variant: EventRowEvaluator): string { + return JSON.stringify([ + tablePatternIdentity(variant.syntacticSource), + variant.filters.map((filter) => expressionBehaviorIdentity(filter.expression.node)).sort(), + variant.columns.map((column) => { + if (column instanceof StarColumnSource) { + return 'star'; + } + const expressionColumn = column as ExpressionColumnSource; + return [expressionColumn.alias, expressionBehaviorIdentity(expressionColumn.expression.expression.node)]; + }), + variant.partitionBy.length, + variant.addedFunctions.length + ]); +} + +function tablePatternIdentity(source: PhysicalSourceResultSet): readonly (string | null)[] { + const pattern = source.tablePattern; + return [pattern.connectionTag, pattern.schema, pattern.tablePattern]; +} + /** - * State for compiling sync streams. + * State for compiling sync streams and replication events into a sync plan. * - * The output of compiling all sync streams is a {@link SyncPlan}, a declarative description of the sync process that - * can be serialized to bucket storage. The compiler stores a mutable intermediate representation that is essentially a - * copy of the sync plan, except that we're using JavaScript classes with methods to compute hash codes and equality - * relations. This allows the compiler to efficiently de-duplicate parameters and buckets. + * The compiler stores a mutable intermediate representation that is essentially a copy of the resulting + * {@link SyncPlan}, except that we're using JavaScript classes with methods to compute hash codes and equality + * relations. Stream queries and event definitions remain separate within that model. * - * Overall, the compilation process is as follows: Each data query for a stream is first parsed by + * The stream compilation process is as follows: Each data query for a stream is first parsed by * {@link StreamQueryParser} into a canonicalized intermediate representation (see that class for details). * Then, {@link QuerierGraphBuilder} analyzes a chain of `AND` expressions to identify parameters (as partition keys) * and their instantiation, as well as static filters that need to be added to reach row. */ export class SyncStreamsCompiler { - readonly output = new CompiledStreamQueries(); + readonly output: SyncPlanCompilerModel = { + streams: new CompiledStreamQueries(), + events: [] + }; private readonly locations = new NodeLocations(); constructor(readonly options: SyncStreamsCompilerOptions) {} @@ -124,11 +190,173 @@ export class SyncStreamsCompiler { } }; } + + /** + * Compiles the payload queries for a named replication event. + * + * Event queries intentionally support a smaller surface than stream queries: They must project and filter a single + * physical source table and cannot depend on request parameters, joins, subqueries or table-valued functions. + */ + event(name: string): IndividualEventCompiler { + const event: CompiledEvent = { name, sourceQueries: [] }; + this.output.events.push(event); + + return { + addSourceQuery: (sql: string, errors: ParsingErrorListener) => { + const stmt = tryParse(sql, errors); + if (stmt == null) { + return; + } + + const parser = new StreamQueryParser({ + compiler: this, + originalText: sql, + locations: this.locations, + parentScope: new SqlScope({}), + errors + }); + const query = parser.parse(stmt); + if (query == null) { + return; + } + + if (query.joined.length != 0) { + errors.report('Event payload queries must SELECT from a single physical source table.', query.span.location); + return; + } + + const defaultSchema = this.options.defaultSchema ?? ''; + const sourceTable = query.sourceTable.tablePattern.toTablePattern(defaultSchema); + if ( + event.sourceQueries.some((source) => + source.sourceTable.tablePattern.toTablePattern(defaultSchema).equals(sourceTable) + ) + ) { + errors.report('Each payload query should query a unique table', query.span.location); + return; + } + + const variants: EventRowEvaluator[] = []; + let valid = true; + for (const variant of query.where.terms) { + const filters: RowExpression[] = []; + for (const term of variant.terms) { + if ( + !(term instanceof SingleDependencyExpression) || + term.dependsOnConnection || + (term.resultSet != null && term.resultSet !== query.sourceTable) + ) { + errors.report( + 'Event payload queries cannot depend on request parameters or other tables.', + term instanceof SingleDependencyExpression ? term.expression.location.location : term.location! + ); + valid = false; + continue; + } + + filters.push(new RowExpression(term)); + } + + variants.push( + new EventRowEvaluator({ + columns: query.resultColumns, + syntacticSource: query.sourceTable, + filters, + partitionBy: [], + addedFunctions: [] + }) + ); + } + + if (valid) { + event.sourceQueries.push({ sql, sourceTable: query.sourceTable, variants }); + } + } + }; + } + + /** + * @returns A sync plan representing an immutable snapshot of the compiler output. + */ + toSyncPlan(): SyncPlan { + const translator = new CompilerModelToSyncPlan(); + return translator.translate(this.output); + } +} + +/** + * Compiles raw event SQL stored alongside older sync plans into the current plan representation. + * + * This is the compatibility boundary for plans written before compiled events were added. Callers should reject fatal + * errors rather than carrying legacy event evaluators into a {@link SyncPlan}. + */ +export function compileEventDefinitions( + definitions: Readonly>, + options: SyncStreamsCompilerOptions +): { events: CompiledEventDescriptor[]; errors: SqlRuleError[] } { + const { compiler, errors } = compileEventDefinitionsIntoCompiler(definitions, options); + return { events: compiler.toSyncPlan().events, errors }; +} + +/** + * Compiles raw event SQL and retains the compiler's JavaScript model for behavioral compatibility checks. + */ +export function compileEventDefinitionsToCompilerModel( + definitions: Readonly>, + options: SyncStreamsCompilerOptions +): { events: CompiledEvent[]; errors: SqlRuleError[] } { + const { compiler, errors } = compileEventDefinitionsIntoCompiler(definitions, options); + return { events: compiler.output.events, errors }; +} + +/** + * Shared compilation step for the two event representations above. + * + * Returning the compiler, rather than one of its outputs, is intentional: persisted-plan normalization translates the + * result into serializable sync-plan data, while compatibility matching needs the JavaScript compiler model and its + * behavioral structure. Keeping both paths on this helper ensures they compile the same SQL with the same options. + */ +function compileEventDefinitionsIntoCompiler( + definitions: Readonly>, + options: SyncStreamsCompilerOptions +): { compiler: SyncStreamsCompiler; errors: SqlRuleError[] } { + // Use one compiler for the complete event collection, matching normal sync-config compilation. In particular, this + // keeps all compiler-owned expression locations and intermediate event models in one coherent output snapshot. + const compiler = new SyncStreamsCompiler(options); + const errors: SqlRuleError[] = []; + + for (const [name, queries] of Object.entries(definitions)) { + // event() registers the named model immediately; each payload query then contributes one validated physical-table + // source to that model. Invalid queries report errors and are not added to the compiled event. + const event = compiler.event(name); + for (const sql of queries) { + event.addSourceQuery(sql, { + report(message, location, reportOptions) { + // Locations are offsets into this specific payload SQL string, so bind every reported compiler diagnostic to + // that source before accumulating it with diagnostics from the other event queries. + const error = new SqlRuleError(message, sql, location); + error.type = reportOptions?.isWarning ? 'warning' : 'fatal'; + errors.push(error); + } + }); + } + } + + return { compiler, errors }; } function tryParse(sql: string, errors: ParsingErrorListener): Statement | null { try { - const [stmt] = parse(sql, { locationTracking: true }); + const statements = parse(sql, { locationTracking: true }); + if (statements.length != 1) { + errors.report( + 'Only a single SELECT statement is supported', + statements[1]?._location ?? { start: 0, end: sql.length } + ); + return null; + } + + const [stmt] = statements; return stmt; } catch (e: any) { const location: NodeLocation | undefined = e.token?._location; @@ -161,6 +389,13 @@ export interface IndividualSyncStreamCompiler { finish(): void; } +export interface IndividualEventCompiler { + /** + * Validates and adds one payload query to this event. + */ + addSourceQuery(sql: string, errors: ParsingErrorListener): void; +} + /** * Something reporting errors. * @@ -205,12 +440,16 @@ export class CompiledStreamQueries { canonicalizePointLookup(lookup: PointLookup): PointLookup { return this._pointLookups.getOrInsert(lookup)[0]; } +} - /** - * @returns A sync plan representing an immutable snapshot of this intermediate representation. - */ - toSyncPlan(): SyncPlan { - const translator = new CompilerModelToSyncPlan(); - return translator.translate(this); - } +/** + * Top-level compiler output used to assemble a complete sync plan. + * + * Streams and events are sibling sync-config concerns. Keeping their intermediate state separate prevents stream + * compilation abstractions from acquiring event-specific responsibilities just because both are persisted in one + * {@link SyncPlan}. + */ +export interface SyncPlanCompilerModel { + readonly streams: CompiledStreamQueries; + readonly events: CompiledEvent[]; } diff --git a/packages/sync-rules/src/compiler/expression.ts b/packages/sync-rules/src/compiler/expression.ts index fee1eb130..78cf926e7 100644 --- a/packages/sync-rules/src/compiler/expression.ts +++ b/packages/sync-rules/src/compiler/expression.ts @@ -77,6 +77,100 @@ export class SyncExpression implements EqualsIgnoringResultSet { } } +/** + * A stable identity for expression behavior across syntactic operand ordering changes. + * + * Boolean conjunction/disjunction and equality are commutative. Comparisons are normalized to one direction, so + * `a > b` and `b < a` also match. Operators and constructs where order affects behavior retain their original order. + */ +export function expressionBehaviorIdentity(expression: SqlExpression): string { + switch (expression.type) { + case 'data': + return JSON.stringify(['data', expressionInputIdentity(expression.source)]); + case 'unary': + return JSON.stringify(['unary', expression.operator, expressionBehaviorIdentity(expression.operand)]); + case 'binary': { + if (expression.operator == 'and' || expression.operator == 'or') { + const operands: SqlExpression[] = []; + collectAssociativeOperands(expression, expression.operator, operands); + return JSON.stringify(['binary', expression.operator, operands.map(expressionBehaviorIdentity).sort()]); + } + + let operator = expression.operator; + let left = expression.left; + let right = expression.right; + if (operator == '>' || operator == '>=') { + operator = operator == '>' ? '<' : '<='; + [left, right] = [right, left]; + } + + const operands = [expressionBehaviorIdentity(left), expressionBehaviorIdentity(right)]; + if (operator == '=' || operator == 'is') { + operands.sort(); + } + return JSON.stringify(['binary', operator, operands]); + } + case 'between': + return JSON.stringify([ + 'between', + expressionBehaviorIdentity(expression.value), + expressionBehaviorIdentity(expression.low), + expressionBehaviorIdentity(expression.high) + ]); + case 'scalar_in': + return JSON.stringify([ + 'scalar_in', + expressionBehaviorIdentity(expression.target), + expression.in.map(expressionBehaviorIdentity) + ]); + case 'case_when': + return JSON.stringify([ + 'case_when', + expression.operand == null ? null : expressionBehaviorIdentity(expression.operand), + expression.whens.map((branch) => [ + expressionBehaviorIdentity(branch.when), + expressionBehaviorIdentity(branch.then) + ]), + expression.else == null ? null : expressionBehaviorIdentity(expression.else) + ]); + case 'cast': + return JSON.stringify(['cast', expression.cast_as, expressionBehaviorIdentity(expression.operand)]); + case 'function': + return JSON.stringify(['function', expression.function, expression.parameters.map(expressionBehaviorIdentity)]); + case 'lit_null': + return JSON.stringify(['lit_null']); + case 'lit_double': + return JSON.stringify(['lit_double', expression.value]); + case 'lit_int': + return JSON.stringify(['lit_int', expression.base10]); + case 'lit_string': + return JSON.stringify(['lit_string', expression.value]); + } +} + +function collectAssociativeOperands( + expression: SqlExpression, + operator: 'and' | 'or', + output: SqlExpression[] +): void { + if (expression.type == 'binary' && expression.operator == operator) { + collectAssociativeOperands(expression.left, operator, output); + collectAssociativeOperands(expression.right, operator, output); + } else { + output.push(expression); + } +} + +function expressionInputIdentity(input: ExpressionInput): readonly string[] { + if (input instanceof ColumnInRow) { + return ['column', input.column]; + } else if (input instanceof RowMetadata) { + return ['row_metadata', input.kind]; + } else { + return ['connection_parameter', input.source]; + } +} + class FindExternalData extends RecursiveExpressionVisitor { defaultExpression(expr: SqlExpression, arg: ExpressionInput[]): void { this.visitChildren(expr, arg); diff --git a/packages/sync-rules/src/compiler/ir_to_sync_plan.ts b/packages/sync-rules/src/compiler/ir_to_sync_plan.ts index 5cc3a3992..3d5bcf56a 100644 --- a/packages/sync-rules/src/compiler/ir_to_sync_plan.ts +++ b/packages/sync-rules/src/compiler/ir_to_sync_plan.ts @@ -2,7 +2,7 @@ import { SqlExpression } from '../sync_plan/expression.js'; import { MapSourceVisitor, visitExpr } from '../sync_plan/expression_visitor.js'; import * as plan from '../sync_plan/plan.js'; import * as resolver from './bucket_resolver.js'; -import { CompiledStreamQueries } from './compiler.js'; +import type { SyncPlanCompilerModel } from './compiler.js'; import { Equality, HashMap, StableHasher, unorderedEquality } from './equality.js'; import { ColumnInRow, ExpressionInput, RowMetadata, SyncExpression } from './expression.js'; import * as rows from './rows.js'; @@ -46,12 +46,12 @@ export class CompilerModelToSyncPlan { return mapped; } - translate(source: CompiledStreamQueries): plan.SyncPlan { - const queriersByStream = Object.groupBy(source.resolvers, (r) => r.options.name); + translate(source: SyncPlanCompilerModel): plan.SyncPlan { + const queriersByStream = Object.groupBy(source.streams.resolvers, (r) => r.options.name); return { - dataSources: source.evaluators.map((e) => this.translateRowEvaluator(e)), - parameterIndexes: source.pointLookups.map((p, i) => this.translatePointLookup(p, i)), + dataSources: source.streams.evaluators.map((e) => this.translateRowEvaluator(e)), + parameterIndexes: source.streams.pointLookups.map((p, i) => this.translatePointLookup(p, i)), // Note: data sources and parameter indexes must be translated first because we reference them in stream // resolvers. streams: Object.values(queriersByStream).map((resolvers) => { @@ -60,7 +60,17 @@ export class CompilerModelToSyncPlan { queriers: resolvers!.map((e) => this.translateStreamResolver(e)) }; }), - buckets: this.buckets + buckets: this.buckets, + events: source.events.map((event): plan.CompiledEventDescriptor => { + return { + name: event.name, + sourceQueries: event.sourceQueries.map((query) => ({ + sql: query.sql, + sourceTable: query.sourceTable.tablePattern, + variants: query.variants.map((variant) => this.translateRowProjection(variant)) + })) + }; + }) }; } @@ -112,32 +122,39 @@ export class CompilerModelToSyncPlan { private translateRowEvaluator(value: rows.RowEvaluator): plan.StreamDataSource { return this.translateStatefulObject(value, () => { - const hasher = new StableHasher(); - value.buildBehaviorHashCode(hasher); const mapped = { - sourceTable: value.tablePattern, - hashCode: hasher.buildHashCode(), - tableValuedFunctions: this.translateAddedTableValuedFunctions(value.addedFunctions, value), - columns: value.columns.map((e) => { - if (e instanceof rows.StarColumnSource) { - return 'star'; - } else { - return { - expr: this.translateExpression(e.expression.expression, value.syntacticSource, value.addedFunctions), - alias: e.alias ?? null - }; - } - }), - outputTableName: value.outputName, - filters: value.filters.map((e) => - this.translateExpression(e.expression, value.syntacticSource, value.addedFunctions) - ), - parameters: value.partitionBy.map((e) => this.translatePartitionKey(e, value)) + ...this.translateRowProjection(value), + outputTableName: value.outputName } satisfies plan.StreamDataSource; return mapped; }); } + private translateRowProjection(value: rows.RowEvaluator | rows.EventRowEvaluator): plan.RowProjection { + const hasher = new StableHasher(); + value.buildBehaviorHashCode(hasher); + + return { + sourceTable: value.tablePattern, + hashCode: hasher.buildHashCode(), + tableValuedFunctions: this.translateAddedTableValuedFunctions(value.addedFunctions, value), + columns: value.columns.map((e) => { + if (e instanceof rows.StarColumnSource) { + return 'star'; + } else { + return { + expr: this.translateExpression(e.expression.expression, value.syntacticSource, value.addedFunctions), + alias: e.alias ?? null + }; + } + }), + filters: value.filters.map((e) => + this.translateExpression(e.expression, value.syntacticSource, value.addedFunctions) + ), + parameters: value.partitionBy.map((e) => this.translatePartitionKey(e, value)) + }; + } + private translatePointLookup(value: rows.PointLookup, index: number): plan.StreamParameterIndexLookupCreator { return this.translateStatefulObject(value, () => { const hasher = new StableHasher(); diff --git a/packages/sync-rules/src/compiler/querier_graph.ts b/packages/sync-rules/src/compiler/querier_graph.ts index 05bd84d57..60595ee4d 100644 --- a/packages/sync-rules/src/compiler/querier_graph.ts +++ b/packages/sync-rules/src/compiler/querier_graph.ts @@ -69,7 +69,7 @@ export class QuerierGraphBuilder { return []; } - this.compiler.output.resolvers.push(...buckets); + this.compiler.output.streams.resolvers.push(...buckets); return buckets; } @@ -164,7 +164,7 @@ class PendingQuerierPath { const state = this.resolveResultSet(this.query.sourceTable); const [partitions, partitionValues] = state.resolvePartitions(); - const evaluator = this.builder.compiler.output.canonicalizeEvaluator( + const evaluator = this.builder.compiler.output.streams.canonicalizeEvaluator( new RowEvaluator({ columns: this.query.resultColumns, syntacticSource: this.query.sourceTable, @@ -435,7 +435,7 @@ class PendingQuerierPath { if (data.type == 'point') { const resultSet = data.resultSet; const [partitionKeys, partitionInputs] = resultSet.resolvePartitions(); - const canonicalized = this.builder.compiler.output.canonicalizePointLookup( + const canonicalized = this.builder.compiler.output.streams.canonicalizePointLookup( new PointLookup({ syntacticSource: data.source, filters: resultSet.filters, diff --git a/packages/sync-rules/src/compiler/rows.ts b/packages/sync-rules/src/compiler/rows.ts index 73f544e6d..e100b2b6f 100644 --- a/packages/sync-rules/src/compiler/rows.ts +++ b/packages/sync-rules/src/compiler/rows.ts @@ -71,7 +71,7 @@ export class TableValuedPartitionKey extends PartitionKey { * This includes {@link RowEvaluator}s, which assigns rows into buckets, and {@link PointLookup}, which creates * parameter lookups used to resolve bucket ids when a user connects. */ -export type SourceRowProcessor = RowEvaluator | PointLookup; +export type SourceRowProcessor = RowEvaluator | EventRowEvaluator | PointLookup; interface SourceProcessorOptions { readonly syntacticSource: PhysicalSourceResultSet; @@ -205,6 +205,33 @@ export class RowEvaluator extends BaseSourceRowProcessor { } } +/** + * A row evaluator producing an event payload. + * + * Unlike {@link RowEvaluator}, the alias of the source table does not affect behavior because event payloads don't + * have a logical output table name. + */ +export class EventRowEvaluator extends BaseSourceRowProcessor { + readonly columns: ColumnSource[]; + + constructor(options: SourceProcessorOptions & { columns: ColumnSource[] }) { + super(options); + this.columns = options.columns; + } + + buildBehaviorHashCode(hasher: StableHasher): void { + this.addBaseHashCode(hasher); + // An event's projected columns, expressions and aliases define the payload delivered to its handler. Changing + // them therefore changes event behavior, so include them here to keep this hash consistent with + // behavesIdenticalTo() and with the compiled definition that will be reprocessed. + equalsIgnoringResultSetList.hash(hasher, this.columns); + } + + behavesIdenticalTo(other: EventRowEvaluator): boolean { + return this.baseMatchesOther(other) && equalsIgnoringResultSetList.equals(other.columns, this.columns); + } +} + /** * A point lookup, creating a materialized index. * diff --git a/packages/sync-rules/src/events/CompiledEventSourceQuery.ts b/packages/sync-rules/src/events/CompiledEventSourceQuery.ts new file mode 100644 index 000000000..e274cdd0f --- /dev/null +++ b/packages/sync-rules/src/events/CompiledEventSourceQuery.ts @@ -0,0 +1,112 @@ +import { HydrationInput } from '../BucketSource.js'; +import { SourceTableRef } from '../SourceTableRef.js'; +import { EvaluatedRowProjection, PendingRowProjection } from '../sync_plan/evaluator/row_projection.js'; +import { + CompiledEventDescriptor as CompiledEventDescriptorPlan, + CompiledEventSourceQuery as CompiledEventSourceQueryPlan +} from '../sync_plan/plan.js'; +import { TablePattern } from '../TablePattern.js'; +import { EvaluateRowOptions, SqliteRow } from '../types.js'; +import { EvaluatedEventRowWithErrors, EventDefinition, HydratedEventDescriptor } from './EventDescriptor.js'; + +/** A named event prepared from a compiled sync plan, before scalar expressions are prepared for evaluation. */ +export class PreparedEventDefinition implements EventDefinition { + readonly name: string; + readonly sourceQueries: PreparedEventSourceQuery[]; + + constructor(source: CompiledEventDescriptorPlan, defaultSchema: string) { + this.name = source.name; + this.sourceQueries = source.sourceQueries.map((query) => new PreparedEventSourceQuery(query, defaultSchema)); + } + + createEvaluator(input: HydrationInput): HydratedEventDescriptor { + return new HydratedCompiledEventDescriptor( + this.name, + this.sourceQueries.map((query) => query.createEvaluator(input)) + ); + } + + getSourceTables(): Set { + return sourceTables(this.sourceQueries); + } + + tableTriggersEvent(table: SourceTableRef): boolean { + return this.sourceQueries.some((query) => query.applies(table)); + } +} + +/** An event source query prepared from a compiled sync plan, before scalar expressions are evaluated. */ +export class PreparedEventSourceQuery { + readonly sourceTable: TablePattern; + private readonly variants: PendingRowProjection[]; + + constructor(source: CompiledEventSourceQueryPlan, defaultSchema: string) { + this.sourceTable = source.sourceTable.toTablePattern(defaultSchema); + this.variants = source.variants.map((variant) => new PendingRowProjection(variant, defaultSchema)); + } + + applies(table: SourceTableRef): boolean { + return this.sourceTable.matches(table); + } + + createEvaluator(input: HydrationInput): HydratedCompiledEventSourceQuery { + return new HydratedCompiledEventSourceQuery( + this.sourceTable, + this.variants.map((variant) => variant.instantiate(input.scalarExpressions)) + ); + } +} + +class HydratedCompiledEventDescriptor implements HydratedEventDescriptor { + constructor( + readonly name: string, + readonly sourceQueries: HydratedCompiledEventSourceQuery[] + ) {} + + evaluateRowWithErrors(options: EvaluateRowOptions): EvaluatedEventRowWithErrors { + const matchingQuery = this.sourceQueries.find((query) => query.applies(options.sourceTable)); + if (matchingQuery == null) { + return { errors: [{ error: `No matching source query found for table ${options.sourceTable.name}` }] }; + } + + return matchingQuery.evaluateRowWithErrors(options.sourceTable, options.record); + } + + getSourceTables(): Set { + return sourceTables(this.sourceQueries); + } + + tableTriggersEvent(table: SourceTableRef): boolean { + return this.sourceQueries.some((query) => query.applies(table)); + } +} + +class HydratedCompiledEventSourceQuery { + constructor( + readonly sourceTable: TablePattern, + private readonly variants: ((options: EvaluateRowOptions) => EvaluatedRowProjection[])[] + ) {} + + applies(table: SourceTableRef): boolean { + return this.sourceTable.matches(table); + } + + evaluateRowWithErrors(table: SourceTableRef, row: SqliteRow): EvaluatedEventRowWithErrors { + try { + for (const evaluate of this.variants) { + const [result] = evaluate({ sourceTable: table, record: row }); + if (result != null) { + return { result: { data: result.data }, errors: [] }; + } + } + + return { errors: [] }; + } catch (error) { + return { errors: [{ error: error instanceof Error ? error.message : 'Evaluating event query failed' }] }; + } + } +} + +function sourceTables(queries: readonly { sourceTable: TablePattern }[]): Set { + return new Set(queries.map((query) => query.sourceTable)); +} diff --git a/packages/sync-rules/src/events/EventDescriptor.ts b/packages/sync-rules/src/events/EventDescriptor.ts new file mode 100644 index 000000000..cc9da2474 --- /dev/null +++ b/packages/sync-rules/src/events/EventDescriptor.ts @@ -0,0 +1,31 @@ +import { HydrationInput } from '../BucketSource.js'; +import { SourceTableRef } from '../SourceTableRef.js'; +import { TablePattern } from '../TablePattern.js'; +import { EvaluateRowOptions, EvaluationError, SqliteJsonRow } from '../types.js'; + +export type EvaluatedEventSourceRow = { + data: SqliteJsonRow; +}; + +export type EvaluatedEventRowWithErrors = { + result?: EvaluatedEventSourceRow; + errors: EvaluationError[]; +}; + +/** A parsed event definition whose compiled expressions have not yet been prepared for evaluation. */ +export interface EventDefinition { + readonly name: string; + + createEvaluator(input: HydrationInput): HydratedEventDescriptor; + getSourceTables(): Set; + tableTriggersEvent(table: SourceTableRef): boolean; +} + +/** An event definition whose payload queries can evaluate replicated rows. */ +export interface HydratedEventDescriptor { + readonly name: string; + + evaluateRowWithErrors(options: EvaluateRowOptions): EvaluatedEventRowWithErrors; + getSourceTables(): Set; + tableTriggersEvent(table: SourceTableRef): boolean; +} diff --git a/packages/sync-rules/src/events/SqlEventDescriptor.ts b/packages/sync-rules/src/events/SqlEventDescriptor.ts deleted file mode 100644 index 9de432d6c..000000000 --- a/packages/sync-rules/src/events/SqlEventDescriptor.ts +++ /dev/null @@ -1,67 +0,0 @@ -import { CompatibilityContext } from '../compatibility.js'; -import { SqlRuleError } from '../errors.js'; -import { QueryParseResult } from '../legacy/SqlBucketDescriptor.js'; -import { SourceTableRef } from '../SourceTableRef.js'; -import { SyncRulesOptions } from '../SqlSyncRules.js'; -import { TablePattern } from '../TablePattern.js'; -import { EvaluateRowOptions } from '../types.js'; -import { EvaluatedEventRowWithErrors, SqlEventSourceQuery } from './SqlEventSourceQuery.js'; - -/** - * A sync config event which is triggered from a SQL table change. - */ -export class SqlEventDescriptor { - name: string; - sourceQueries: SqlEventSourceQuery[] = []; - - constructor( - name: string, - private readonly compatibility: CompatibilityContext - ) { - this.name = name; - } - - addSourceQuery(sql: string, options: SyncRulesOptions): QueryParseResult { - const source = SqlEventSourceQuery.fromSql(sql, options, this.compatibility); - - // Each source query should be for a unique table - const existingSourceQuery = this.sourceQueries.find((q) => q.table == source.table); - if (existingSourceQuery) { - return { - parsed: false, - errors: [new SqlRuleError('Each payload query should query a unique table', sql)] - }; - } - - this.sourceQueries.push(source); - - return { - parsed: true, - errors: source.errors - }; - } - - evaluateRowWithErrors(options: EvaluateRowOptions): EvaluatedEventRowWithErrors { - // There should only be 1 payload result per source query - const matchingQuery = this.sourceQueries.find((q) => q.applies(options.sourceTable)); - if (!matchingQuery) { - return { - errors: [{ error: `No matching source query found for table ${options.sourceTable.name}` }] - }; - } - - return matchingQuery.evaluateRowWithErrors(options.sourceTable, options.record); - } - - getSourceTables(): Set { - let result = new Set(); - for (let query of this.sourceQueries) { - result.add(query.sourceTable!); - } - return result; - } - - tableTriggersEvent(table: SourceTableRef): boolean { - return this.sourceQueries.some((query) => query.applies(table)); - } -} diff --git a/packages/sync-rules/src/events/SqlEventSourceQuery.ts b/packages/sync-rules/src/events/SqlEventSourceQuery.ts deleted file mode 100644 index f19fc5b25..000000000 --- a/packages/sync-rules/src/events/SqlEventSourceQuery.ts +++ /dev/null @@ -1,152 +0,0 @@ -import { parse } from 'pgsql-ast-parser'; -import { CompatibilityContext } from '../compatibility.js'; -import { SqlRuleError } from '../errors.js'; -import { ExpressionType } from '../ExpressionType.js'; -import { BaseSqlDataQuery, BaseSqlDataQueryOptions, RowValueExtractor } from '../legacy/BaseSqlDataQuery.js'; -import { AvailableTable, SqlTools } from '../legacy/sql_filters.js'; -import { checkUnsupportedFeatures, isClauseError } from '../legacy/sql_support.js'; -import { TableQuerySchema } from '../legacy/TableQuerySchema.js'; -import { SourceTableRef } from '../SourceTableRef.js'; -import { SyncRulesOptions } from '../SqlSyncRules.js'; -import { TablePattern } from '../TablePattern.js'; -import { EvaluationError, QuerySchema, SqliteJsonRow, SqliteRow } from '../types.js'; -import { isSelectStatement } from '../utils.js'; - -export type EvaluatedEventSourceRow = { - data: SqliteJsonRow; -}; - -export type EvaluatedEventRowWithErrors = { - result?: EvaluatedEventSourceRow; - errors: EvaluationError[]; -}; - -/** - * Defines how a Replicated Row is mapped to source parameters for events. - */ -// TODO: Use Sync Streams compiler infrastruture instead of legacy data queries -export class SqlEventSourceQuery extends BaseSqlDataQuery { - static fromSql(sql: string, options: SyncRulesOptions, compatibility: CompatibilityContext) { - const parsed = parse(sql, { locationTracking: true }); - const schema = options.schema; - - if (parsed.length > 1) { - throw new SqlRuleError('Only a single SELECT statement is supported', sql, parsed[1]?._location); - } - const q = parsed[0]; - if (!isSelectStatement(q)) { - throw new SqlRuleError('Only SELECT statements are supported', sql, q._location); - } - - let errors: SqlRuleError[] = []; - - errors.push(...checkUnsupportedFeatures(sql, q)); - - if (q.from == null || q.from.length != 1 || q.from[0].type != 'table') { - throw new SqlRuleError('Must SELECT from a single table', sql, q.from?.[0]._location); - } - - const tableRef = q.from?.[0].name; - if (tableRef?.name == null) { - throw new SqlRuleError('Must SELECT from a single table', sql, q.from?.[0]._location); - } - const alias = AvailableTable.fromAst(tableRef); - - const sourceTable = new TablePattern(tableRef.schema ?? options.defaultSchema, tableRef.name); - let querySchema: QuerySchema | undefined = undefined; - if (schema) { - const tables = schema.getTables(sourceTable); - if (tables.length == 0) { - const e = new SqlRuleError( - `Table ${sourceTable.schema}.${sourceTable.tablePattern} not found`, - sql, - q.from?.[0]?._location - ); - e.type = 'warning'; - - errors.push(e); - } else { - querySchema = new TableQuerySchema(tables, alias); - } - } - - const tools = new SqlTools({ - table: alias, - parameterTables: [], - valueTables: [alias], - sql, - schema: querySchema, - compatibilityContext: compatibility - }); - - let extractors: RowValueExtractor[] = []; - for (let column of q.columns ?? []) { - const name = tools.getOutputName(column); - if (name != '*') { - const clause = tools.compileRowValueExtractor(column.expr); - if (isClauseError(clause)) { - // Error logged already - continue; - } - extractors.push({ - extract: (tables, output) => { - output[name] = clause.evaluate(tables); - }, - getTypes(schema, into) { - const def = clause.getColumnDefinition(schema); - into[name] = { name, type: def?.type ?? ExpressionType.NONE, originalType: def?.originalType }; - } - }); - } else { - extractors.push({ - extract: (tables, output) => { - const row = tables[alias.nameInSchema]; - for (let key in row) { - if (key.startsWith('_')) { - continue; - } - output[key] ??= row[key]; - } - }, - getTypes(schema, into) { - for (let column of schema.getColumns(alias.nameInSchema)) { - into[column.name] ??= column; - } - } - }); - } - } - errors.push(...tools.errors); - - return new SqlEventSourceQuery({ - sourceTable, - table: alias, - sql, - columns: q.columns ?? [], - extractors: extractors, - tools, - bucketParameters: [], - errors: errors - }); - } - - constructor(options: BaseSqlDataQueryOptions) { - super(options); - } - - evaluateRowWithErrors(table: SourceTableRef, row: SqliteRow): EvaluatedEventRowWithErrors { - try { - const tables = { [this.table!.nameInSchema]: this.addSpecialParameters(table, row) }; - - const data = this.transformRow(tables); - return { - result: { - data - }, - errors: [] - }; - } catch (e) { - return { errors: [e.message ?? `Evaluating data query failed`] }; - } - } -} diff --git a/packages/sync-rules/src/from_yaml.ts b/packages/sync-rules/src/from_yaml.ts index e03501667..4080a3b67 100644 --- a/packages/sync-rules/src/from_yaml.ts +++ b/packages/sync-rules/src/from_yaml.ts @@ -9,13 +9,14 @@ import { import { ParsingErrorListener, SyncStreamsCompiler } from './compiler/compiler.js'; import { CommonTableExpression } from './compiler/sqlite.js'; import { SqlRuleError, SyncRulesErrors, YamlError } from './errors.js'; -import { SqlEventDescriptor } from './events/SqlEventDescriptor.js'; +import { PreparedEventDefinition } from './events/CompiledEventSourceQuery.js'; import { validateSyncRulesSchema } from './json_schema.js'; import { QueryParseResult, SqlBucketDescriptor } from './legacy/SqlBucketDescriptor.js'; import { syncStreamFromSql } from './legacy/streams/from_sql.js'; import { SqlSyncRules } from './SqlSyncRules.js'; import { validateStorageVersion } from './StorageVersion.js'; import { PrecompiledSyncConfig } from './sync_plan/evaluator/index.js'; +import { CompiledEventDescriptor } from './sync_plan/plan.js'; import { SyncConfig, SyncConfigWithErrors } from './SyncConfig.js'; import { TablePattern } from './TablePattern.js'; import { QueryParseOptions, SourceSchema, StreamParseOptions } from './types.js'; @@ -118,23 +119,24 @@ export class SyncConfigFromYaml { const bucketMap = rootState.get('bucket_definitions')?.requireMap(); const streamMap = rootState.get('streams')?.requireMap(); const globalCtes = rootState.get('with')?.requireMap(); + const eventMap = rootState.get('event_definitions')?.requireMap(); let result: SyncConfig; if (compatibility.edition >= CompatibilityEdition.COMPILED_STREAMS) { - result = this.#compileSyncPlan(bucketMap, streamMap, globalCtes, compatibility); + result = this.#compileSyncPlan(bucketMap, streamMap, globalCtes, eventMap, compatibility); this.#warnOnUnusedCtes(); } else { // We don't support CTEs at all in this compiler implementation. globalCtes?.reportError('Common table expressions require edition 3.'); - result = this.#legacyParseBucketDefinitionsAndStreams(bucketMap, streamMap, compatibility); + const eventCompiler = new SyncStreamsCompiler(this.options); + this.#compileEventDefinitions(eventMap, eventCompiler); + const eventPlan = eventCompiler.toSyncPlan(); + result = this.#legacyParseBucketDefinitionsAndStreams(bucketMap, streamMap, compatibility, eventPlan.events); } result.storageVersion = storageVersion; - const eventDefinitions = this.#parseEventDefinitions(rootState, compatibility); - result.eventDescriptors.push(...eventDefinitions); - return result; } @@ -190,6 +192,7 @@ export class SyncConfigFromYaml { bucketMap: YamlMapState | undefined, streamMap: YamlMapState | undefined, globalCtes: YamlMapState | undefined, + eventMap: YamlMapState | undefined, compatibility: CompatibilityContext ) { bucketMap?.reportError( @@ -286,8 +289,9 @@ export class SyncConfigFromYaml { streamCompiler.finish(); } - // We pass an empty array for eventDefinitions here because those will get parsed in #parseEventDefinitions. - return new PrecompiledSyncConfig(compiler.output.toSyncPlan(), compatibility, [], { + this.#compileEventDefinitions(eventMap, compiler); + + return new PrecompiledSyncConfig(compiler.toSyncPlan(), compatibility, { defaultSchema: this.options.defaultSchema, sourceText: this.yaml }); @@ -304,10 +308,14 @@ export class SyncConfigFromYaml { #legacyParseBucketDefinitionsAndStreams( bucketMap: YamlMapState | undefined, streamMap: YamlMapState | undefined, - compatibility: CompatibilityContext + compatibility: CompatibilityContext, + events: CompiledEventDescriptor[] ) { const rules = new SqlSyncRules(this.yaml); rules.compatibility = compatibility; + rules.eventDefinitions.push( + ...events.map((event) => new PreparedEventDefinition(event, this.options.defaultSchema)) + ); if (bucketMap == null && streamMap == null) { this.#errors.push(new YamlError(new Error(`'bucket_definitions' or 'streams' is required`))); @@ -438,31 +446,23 @@ export class SyncConfigFromYaml { return undefined; } - #parseEventDefinitions(parsed: YamlMapState, compatibility: CompatibilityContext) { - const eventMap = parsed.get('event_definitions')?.requireMap(); - const eventDescriptors: SqlEventDescriptor[] = []; - - for (const { key: name, keyScalar, value: maybeMap } of eventMap?.stringKeyedItems() ?? []) { + #compileEventDefinitions(eventMap: YamlMapState | undefined, compiler: SyncStreamsCompiler): void { + for (const { key: name, value: maybeMap } of eventMap?.stringKeyedItems() ?? []) { using value = maybeMap.requireMap(`Event definitions must be objects.`); if (value == null) continue; const payloads = value.get('payloads')?.requireSequence(`Event definition payloads must be an array.`); if (payloads == null) continue; - const eventDescriptor = new SqlEventDescriptor(name, compatibility); - for (let item of payloads.items) { + const eventCompiler = compiler.event(name); + for (const item of payloads.items) { const itemScalar = item.requireScalar(`Payload queries for events must be scalar.`); if (itemScalar == null) continue; - this.#withScalar(item, (q) => { - return eventDescriptor.addSourceQuery(q, this.options); - }); + const [sql, errorListener] = this.#scalarErrorListener(itemScalar.node); + eventCompiler.addSourceQuery(sql, errorListener); } - - eventDescriptors.push(eventDescriptor); } - - return eventDescriptors; } #checkUniqueName(name: string, literal: YamlState): boolean { diff --git a/packages/sync-rules/src/index.ts b/packages/sync-rules/src/index.ts index bcde2840b..d3073d508 100644 --- a/packages/sync-rules/src/index.ts +++ b/packages/sync-rules/src/index.ts @@ -4,8 +4,8 @@ export * from './BucketSource.js'; export * from './cast.js'; export * from './compatibility.js'; export * from './errors.js'; -export * from './events/SqlEventDescriptor.js'; -export * from './events/SqlEventSourceQuery.js'; +export * from './events/CompiledEventSourceQuery.js'; +export * from './events/EventDescriptor.js'; export * from './ExpressionType.js'; export * from './HydratedSyncConfig.js'; export * from './HydrationState.js'; diff --git a/packages/sync-rules/src/sync_plan/evaluator/bucket_data_source.ts b/packages/sync-rules/src/sync_plan/evaluator/bucket_data_source.ts index 914cda014..618f63087 100644 --- a/packages/sync-rules/src/sync_plan/evaluator/bucket_data_source.ts +++ b/packages/sync-rules/src/sync_plan/evaluator/bucket_data_source.ts @@ -3,26 +3,14 @@ import { idFromData } from '../../cast.js'; import { ColumnDefinition } from '../../ExpressionType.js'; import { SourceTableRef } from '../../SourceTableRef.js'; import { TablePattern } from '../../TablePattern.js'; -import { - EvaluateRowOptions, - SourceSchema, - SqliteJsonRow, - UnscopedEvaluatedRow, - UnscopedEvaluationResult -} from '../../types.js'; -import { filterJsonRow, isJsonValue, isValidParameterValue, JSONBucketNameSerialize } from '../../utils.js'; -import { - ScalarExpressionEngine, - ScalarStatement, - scalarStatementToSql, - TableValuedFunctionOutput -} from '../engine/scalar_expression_engine.js'; -import { SqlExpression } from '../expression.js'; +import { EvaluateRowOptions, SourceSchema, UnscopedEvaluatedRow, UnscopedEvaluationResult } from '../../types.js'; +import { isValidParameterValue, JSONBucketNameSerialize } from '../../utils.js'; +import { ScalarExpressionEngine } from '../engine/scalar_expression_engine.js'; import { ExpressionToSqlite } from '../expression_to_sql.js'; import * as plan from '../plan.js'; import { SyncPlanSchemaAnalyzer } from '../schema_inference.js'; import { StreamEvaluationContext } from './index.js'; -import { resolveRowMetadata, TableProcessorToSqlHelper } from './table_processor_to_sql.js'; +import { PendingRowProjection } from './row_projection.js'; export class PreparedStreamBucketDataSource implements BucketDataSource { private readonly sourceTables = new Set(); @@ -107,72 +95,27 @@ export class PreparedStreamBucketDataSource implements BucketDataSource { class PendingStreamDataSource { readonly tablePattern: TablePattern; - private readonly outputs: ('star' | { index: number; alias: string })[] = []; - private readonly numberOfOutputExpressions: number; - private readonly numberOfParameters: number; - private readonly evaluatorInputs: (plan.ColumnSqlParameterValue | plan.RowMetadataSqlValue)[]; - private readonly statement: ScalarStatement; + private readonly projection: PendingRowProjection; readonly fixedOutputTableName?: string; constructor(evaluator: plan.StreamDataSource, defaultSchema: string) { - const translationHelper = new TableProcessorToSqlHelper(evaluator); - const outputExpressions: SqlExpression[] = []; - - for (const column of evaluator.columns) { - if (column === 'star') { - this.outputs.push('star'); - } else { - const expressionIndex = outputExpressions.length; - outputExpressions.push(translationHelper.mapper.transform(column.expr)); - this.outputs.push({ index: expressionIndex, alias: column.alias }); - } - } - - this.numberOfOutputExpressions = outputExpressions.length; - for (const parameter of evaluator.parameters) { - outputExpressions.push(translationHelper.mapper.transform(parameter.expr)); - } - this.numberOfParameters = evaluator.parameters.length; - - this.statement = { - outputs: outputExpressions, - filters: translationHelper.filterExpressions, - tableValuedFunctions: translationHelper.tableValuedFunctions - }; + this.projection = new PendingRowProjection(evaluator, defaultSchema); this.fixedOutputTableName = evaluator.outputTableName; - this.tablePattern = evaluator.sourceTable.toTablePattern(defaultSchema); - this.evaluatorInputs = translationHelper.mapper.instantiation; + this.tablePattern = this.projection.tablePattern; } get debugSql(): string { - return scalarStatementToSql(this.statement); + return this.projection.debugSql; } instantiate(engine: ScalarExpressionEngine) { - const evaluator = engine.prepareEvaluator(this.statement); - const pattern = this.tablePattern; + const evaluate = this.projection.instantiate(engine); return (options: EvaluateRowOptions, results: UnscopedEvaluationResult[]) => { try { - const inputInstantiation = this.evaluatorInputs.map((input) => - 'column' in input ? options.record[input.column] : resolveRowMetadata(input, pattern, options.sourceTable) - ); - row: for (const source of evaluator.evaluate(inputInstantiation)) { - const record: SqliteJsonRow = {}; - for (const output of this.outputs) { - if (output === 'star') { - Object.assign(record, filterJsonRow(options.record)); - } else { - const value = source[output.index]; - if (isJsonValue(value)) { - record[output.alias] = value; - } - } - } - const id = idFromData(record); - // source is [...outputs, ...partitionValues] - const partitionValues = source.splice(this.numberOfOutputExpressions, this.numberOfParameters); - + row: for (const projected of evaluate(options)) { + const id = idFromData(projected.data); + const partitionValues = projected.partitionValues; for (const bucketParameter of partitionValues) { if (!isValidParameterValue(bucketParameter)) { continue row; @@ -181,7 +124,7 @@ class PendingStreamDataSource { results.push({ id, - data: record, + data: projected.data, table: this.fixedOutputTableName ?? options.sourceTable.name, serializedBucketParameters: JSONBucketNameSerialize.stringify(partitionValues) } satisfies UnscopedEvaluatedRow); diff --git a/packages/sync-rules/src/sync_plan/evaluator/index.ts b/packages/sync-rules/src/sync_plan/evaluator/index.ts index a779f5a09..2fbbdcb45 100644 --- a/packages/sync-rules/src/sync_plan/evaluator/index.ts +++ b/packages/sync-rules/src/sync_plan/evaluator/index.ts @@ -1,6 +1,6 @@ import { SyncConfig } from '../../SyncConfig.js'; import { CompatibilityContext } from '../../compatibility.js'; -import { SqlEventDescriptor } from '../../index.js'; +import { PreparedEventDefinition } from '../../events/CompiledEventSourceQuery.js'; import * as plan from '../plan.js'; import { PreparedStreamBucketDataSource } from './bucket_data_source.js'; import { StreamBucketSource, StreamInput } from './bucket_source.js'; @@ -17,24 +17,22 @@ export interface StreamEvaluationContext { } export class PrecompiledSyncConfig extends SyncConfig { - /** - * The default schema for this sync config. - * - * This is independent of the loaded {@link plan} (the same sync plan can be loaded with different default schemas). - */ + /** Default schema used to prepare unqualified source-table references from the compiled plan. */ readonly defaultSchema: string; constructor( readonly plan: plan.SyncPlan, compatibility: CompatibilityContext, - eventDefinitions: SqlEventDescriptor[], context: StreamEvaluationContext ) { super(context.sourceText); this.compatibility = compatibility; - this.eventDescriptors = eventDefinitions; this.defaultSchema = context.defaultSchema; + for (const event of plan.events) { + this.eventDefinitions.push(new PreparedEventDefinition(event, context.defaultSchema)); + } + const preparedBuckets = new Map(); const preparedLookups = new Map(); diff --git a/packages/sync-rules/src/sync_plan/evaluator/row_projection.ts b/packages/sync-rules/src/sync_plan/evaluator/row_projection.ts new file mode 100644 index 000000000..d6304252c --- /dev/null +++ b/packages/sync-rules/src/sync_plan/evaluator/row_projection.ts @@ -0,0 +1,98 @@ +import { TablePattern } from '../../TablePattern.js'; +import { EvaluateRowOptions, SqliteJsonRow, SqliteValue } from '../../types.js'; +import { filterJsonRow, isJsonValue } from '../../utils.js'; +import { + ScalarExpressionEngine, + ScalarStatement, + scalarStatementToSql, + TableValuedFunctionOutput +} from '../engine/scalar_expression_engine.js'; +import { SqlExpression } from '../expression.js'; +import * as plan from '../plan.js'; +import { resolveRowMetadata, TableProcessorToSqlHelper } from './table_processor_to_sql.js'; + +export interface EvaluatedRowProjection { + data: SqliteJsonRow; + partitionValues: SqliteValue[]; +} + +/** + * Prepares the shared projection/filter portion of bucket and event row evaluators. + */ +export class PendingRowProjection { + readonly tablePattern: TablePattern; + private readonly outputs: ('star' | { index: number; alias: string })[] = []; + private readonly numberOfOutputExpressions: number; + private readonly numberOfParameters: number; + private readonly evaluatorInputs: (plan.ColumnSqlParameterValue | plan.RowMetadataSqlValue)[]; + private readonly statement: ScalarStatement; + + constructor(evaluator: plan.RowProjection, defaultSchema: string) { + const translationHelper = new TableProcessorToSqlHelper(evaluator); + const outputExpressions: SqlExpression[] = []; + + for (const column of evaluator.columns) { + if (column === 'star') { + this.outputs.push('star'); + } else { + const expressionIndex = outputExpressions.length; + outputExpressions.push(translationHelper.mapper.transform(column.expr)); + this.outputs.push({ index: expressionIndex, alias: column.alias }); + } + } + + this.numberOfOutputExpressions = outputExpressions.length; + for (const parameter of evaluator.parameters) { + outputExpressions.push(translationHelper.mapper.transform(parameter.expr)); + } + this.numberOfParameters = evaluator.parameters.length; + + this.statement = { + outputs: outputExpressions, + filters: translationHelper.filterExpressions, + tableValuedFunctions: translationHelper.tableValuedFunctions + }; + this.tablePattern = evaluator.sourceTable.toTablePattern(defaultSchema); + this.evaluatorInputs = translationHelper.mapper.instantiation; + } + + get debugSql(): string { + return scalarStatementToSql(this.statement); + } + + instantiate(engine: ScalarExpressionEngine): (options: EvaluateRowOptions) => EvaluatedRowProjection[] { + const evaluator = engine.prepareEvaluator(this.statement); + const pattern = this.tablePattern; + + return (options) => { + const inputInstantiation = this.evaluatorInputs.map((input) => + 'column' in input ? options.record[input.column] : resolveRowMetadata(input, pattern, options.sourceTable) + ); + const results: EvaluatedRowProjection[] = []; + + for (const source of evaluator.evaluate(inputInstantiation)) { + const record: SqliteJsonRow = {}; + for (const output of this.outputs) { + if (output === 'star') { + Object.assign(record, filterJsonRow(options.record)); + } else { + const value = source[output.index]; + if (isJsonValue(value)) { + record[output.alias] = value; + } + } + } + + results.push({ + data: record, + partitionValues: source.slice( + this.numberOfOutputExpressions, + this.numberOfOutputExpressions + this.numberOfParameters + ) + }); + } + + return results; + }; + } +} diff --git a/packages/sync-rules/src/sync_plan/plan.ts b/packages/sync-rules/src/sync_plan/plan.ts index 23d39cd45..06cba0529 100644 --- a/packages/sync-rules/src/sync_plan/plan.ts +++ b/packages/sync-rules/src/sync_plan/plan.ts @@ -26,6 +26,7 @@ export interface SyncPlan { buckets: StreamBucketDataSource[]; parameterIndexes: StreamParameterIndexLookupCreator[]; streams: CompiledSyncStream[]; + events: CompiledEventDescriptor[]; } /** @@ -118,21 +119,53 @@ export interface PartitionKey { * duplicate bucket data but can still share the actual processing logic between multiple streams (to avoid evaluating * the same filters and expressions multiple times). */ -export interface StreamDataSource extends TableProcessor { +export interface RowProjection extends TableProcessor { + /** + * Output columns produced when a matching source row is evaluated. + */ + columns: ColumnSource[]; +} + +export interface StreamDataSource extends RowProjection { /** * The name of the output table for evaluated rows. * * If null, the name of the table being evaluated should be used instead. */ outputTableName?: string; +} +export type ColumnSource = 'star' | { expr: SqlExpression; alias: string }; + +/** + * A named replication event compiled from `event_definitions`. + */ +export interface CompiledEventDescriptor { + name: string; + sourceQueries: CompiledEventSourceQuery[]; +} + +/** + * A single payload query for an event. + * + * Event payload queries are restricted to one physical source table. A query can have multiple variants after its + * filter has been normalized to disjunctive normal form; evaluation stops after the first matching variant so one + * source row produces at most one event payload for this query. + */ +export interface CompiledEventSourceQuery { /** - * Output columns describing the row to store in buckets. + * Original SQL retained as a compatibility mirror for services using the legacy event evaluator. Compiled event + * evaluation uses the remaining fields. */ - columns: ColumnSource[]; + sql: string; + sourceTable: ImplicitSchemaTablePattern; + variants: EventRowEvaluator[]; } -export type ColumnSource = 'star' | { expr: SqlExpression; alias: string }; +/** + * A row projection used by one normalized filter variant of an event payload query. + */ +export interface EventRowEvaluator extends RowProjection {} /** * A mapping describing how {@link StreamDataSource}s are combined into buckets. diff --git a/packages/sync-rules/src/sync_plan/plan_equality_serialized.ts b/packages/sync-rules/src/sync_plan/plan_equality_serialized.ts index 4b321b5ce..63f721bbb 100644 --- a/packages/sync-rules/src/sync_plan/plan_equality_serialized.ts +++ b/packages/sync-rules/src/sync_plan/plan_equality_serialized.ts @@ -1,5 +1,5 @@ -import { Equality } from '../compiler/equality.js'; -import { +import type { Equality } from '../compiler/equality.js'; +import type { SerializedBucketDataSource, SerializedDataSource, SerializedParameterIndexLookupCreator diff --git a/packages/sync-rules/src/sync_plan/serialize.ts b/packages/sync-rules/src/sync_plan/serialize.ts index 7db39fd98..2cf94bb7c 100644 --- a/packages/sync-rules/src/sync_plan/serialize.ts +++ b/packages/sync-rules/src/sync_plan/serialize.ts @@ -5,8 +5,10 @@ import { MapSourceVisitor, visitExpr } from './expression_visitor.js'; import { ColumnSource, ColumnSqlParameterValue, + CompiledEventDescriptor, CompiledSyncStream, EvaluateTableValuedFunction, + EventRowEvaluator, ExpandingLookup, ParameterLookup, ParameterValue, @@ -25,18 +27,7 @@ import { TableProcessorTableValuedFunctionOutput } from './plan.js'; -/** - * Serializes a sync plan into a simple JSON object. - * - * While {@link SyncPlan}s are already serializable for the most part, it contains a graph of references from e.g. - * queriers to bucket creators. To represent this efficiently, we assign numbers to referenced elements while - * serializing instead of duplicating definitions. - */ -export function serializeSyncPlan(plan: SyncPlan): SerializedSyncPlan { - const dataSourceIndex = new Map(); - const bucketIndex = new Map(); - const parameterIndex = new Map(); - const expandingLookups = new Map(); +function createTableProcessorSerializer() { const addedTableValuedFunctions = new Map(); let usesRowMetadataSqlValue = false; @@ -88,6 +79,62 @@ export function serializeSyncPlan(plan: SyncPlan): SerializedSyncPlan { }); } + function serializeEventRowEvaluator(source: EventRowEvaluator): SerializedEventRowEvaluator { + return { + hash: source.hashCode, + table: serializeTablePattern(source.sourceTable), + tableValuedFunctions: serializeTableValued(source), + filters: source.filters.map(serializeTableProcessorDataExpr), + partitionBy: translateParameters(source), + columns: source.columns.map((column): SerializedColumnSource => { + if (column == 'star') { + return 'star'; + } + + return { expr: serializeTableProcessorDataExpr(column.expr), alias: column.alias }; + }) + }; + } + + function serializeEventDefinition(event: CompiledEventDescriptor): SerializedEventDescriptor { + return { + name: event.name, + sourceQueries: event.sourceQueries.map((query) => ({ + sql: query.sql, + table: serializeTablePattern(query.sourceTable), + variants: query.variants.map(serializeEventRowEvaluator) + })) + }; + } + + return { + get usesRowMetadataSqlValue() { + return usesRowMetadataSqlValue; + }, + serializeTableProcessorDataExpr, + serializeTablePattern, + serializeTableValued, + translateParameters, + serializeEventDefinition + }; +} + +/** + * Serializes a sync plan into a simple JSON object. + * + * While {@link SyncPlan}s are already serializable for the most part, it contains a graph of references from e.g. + * queriers to bucket creators. To represent this efficiently, we assign numbers to referenced elements while + * serializing instead of duplicating definitions. + */ +export function serializeSyncPlan(plan: SyncPlan): SerializedSyncPlan { + const dataSourceIndex = new Map(); + const bucketIndex = new Map(); + const parameterIndex = new Map(); + const expandingLookups = new Map(); + const tableProcessorSerializer = createTableProcessorSerializer(); + const { serializeTableProcessorDataExpr, serializeTablePattern, serializeTableValued, translateParameters } = + tableProcessorSerializer; + function serializeDataSources(): SerializedDataSource[] { return plan.dataSources.map((source, i) => { dataSourceIndex.set(source, i); @@ -120,7 +167,7 @@ export function serializeSyncPlan(plan: SyncPlan): SerializedSyncPlan { tableValuedFunctions: serializeTableValued(source), filters: source.filters.map(serializeTableProcessorDataExpr), partitionBy: translateParameters(source), - output: source.outputs.map((out) => visitExpr(replaceFunctionReferenceWithIndex, out, null)), + output: source.outputs.map(serializeTableProcessorDataExpr), lookupScope: source.defaultLookupScope } satisfies SerializedParameterIndexLookupCreator; }); @@ -178,7 +225,8 @@ export function serializeSyncPlan(plan: SyncPlan): SerializedSyncPlan { }; } - return { + const events = plan.events.map(tableProcessorSerializer.serializeEventDefinition); + const serialized: SerializedSyncPlan = { dataSources: serializeDataSources(), buckets: plan.buckets.map((bkt, index) => { bucketIndex.set(bkt, index); @@ -193,8 +241,16 @@ export function serializeSyncPlan(plan: SyncPlan): SerializedSyncPlan { stream: s.stream, queriers: s.queriers.map(serializeStreamQuerier) })), - version: usesRowMetadataSqlValue ? 2 : 1 + version: tableProcessorSerializer.usesRowMetadataSqlValue ? 2 : 1 }; + + // Compiled events are intentionally additive to plan versions 1 and 2. The service also persists their raw SQL in + // the legacy eventDescriptors field, so older readers can ignore this field and retain equivalent event behavior. + if (events.length != 0) { + serialized.events = events; + } + + return serialized; } export function deserializeSyncPlan(serialized: unknown): SyncPlan { @@ -282,6 +338,40 @@ export function deserializeSyncPlan(serialized: unknown): SyncPlan { }; }); + function deserializeEventRowEvaluator(source: SerializedEventRowEvaluator): EventRowEvaluator { + const functions = (tableValuedFunctionsInScope = source.tableValuedFunctions); + + return { + hashCode: source.hash, + sourceTable: deserializeTablePattern(source.table), + tableValuedFunctions: functions, + filters: source.filters.map(deserializeTableProcessorDataExpr), + parameters: deserializeParameters(source.partitionBy), + columns: source.columns.map((column): ColumnSource => { + if (column == 'star') { + return 'star'; + } + + return { expr: deserializeTableProcessorDataExpr(column.expr), alias: column.alias }; + }) + }; + } + + if (plan.events != null && !Array.isArray(plan.events)) { + throw new Error('Compiled sync plan events must be an array.'); + } + const serializedEvents = plan.events ?? []; + const events = serializedEvents.map((event): CompiledEventDescriptor => { + return { + name: event.name, + sourceQueries: event.sourceQueries.map((query) => ({ + sql: query.sql, + sourceTable: deserializeTablePattern(query.table), + variants: query.variants.map(deserializeEventRowEvaluator) + })) + }; + }); + function deserializeParameterValue(stages: ExpandingLookup[][], value: SerializedParameterValue): ParameterValue { switch (value.type) { case 'request': @@ -346,15 +436,19 @@ export function deserializeSyncPlan(serialized: unknown): SyncPlan { dataSources, buckets, parameterIndexes, - streams + streams, + events }; } /** - * Every change to the format of {@link SerializedSyncPlan} needs a version bump and a changelog entry in this - * documentation comment. Even for seemingly backward-compatible changes, like adding new fields, older services would - * be unaware of them and thus interpret the sync plan incorrectly. Increasing this version ensures that older services - * wouldn't even try to deserialize sync plans. + * Changes to {@link SerializedSyncPlan} require a version bump when older services would interpret the plan + * incorrectly. Optional additive fields are only safe without a bump when older readers can ignore them while another + * persisted representation preserves equivalent behavior. + * + * Compiled `events` are an explicit additive exception: service-core continues to persist raw event SQL alongside the + * plan for the legacy evaluator. Older readers ignore `events` and use that legacy mirror. Removing the mirror or + * relying on compiled-only event semantics will require a version bump. * * ### Version 2 * @@ -376,6 +470,11 @@ export interface SerializedSyncPlan { buckets: SerializedBucketDataSource[]; parameterIndexes: SerializedParameterIndexLookupCreator[]; streams: SerializedStream[]; + /** + * Optional additive compiled event definitions. Older readers safely ignore this because service-core dual-writes + * equivalent raw SQL in its legacy `eventDescriptors` field. + */ + events?: SerializedEventDescriptor[]; } export interface SerializedBucketDataSource { @@ -416,6 +515,31 @@ export interface SerializedDataSource { partitionBy: SerializedPartitionKey[]; } +export interface SerializedEventDescriptor { + name: string; + sourceQueries: SerializedEventSourceQuery[]; +} + +export interface SerializedEventSourceQuery { + /** Raw SQL retained for the legacy compatibility mirror. */ + sql: string; + table: SerializedTablePattern; + variants: SerializedEventRowEvaluator[]; +} + +export interface SerializedEventRowEvaluator { + table: SerializedTablePattern; + /** + * The compiler's structural hash, retained only for round-trip symmetry with data sources (which reuse the same + * projection shape). It has no behavioral meaning for events. + */ + hash: number; + columns: SerializedColumnSource[]; + filters: SqlExpression[]; + tableValuedFunctions: TableProcessorTableValuedFunction[]; + partitionBy: SerializedPartitionKey[]; +} + export interface SerializedParameterIndexLookupCreator { table: SerializedTablePattern; hash: number; diff --git a/packages/sync-rules/test/src/compiler/events.test.ts b/packages/sync-rules/test/src/compiler/events.test.ts new file mode 100644 index 000000000..abdcf593e --- /dev/null +++ b/packages/sync-rules/test/src/compiler/events.test.ts @@ -0,0 +1,160 @@ +import * as sqlite from 'node:sqlite'; +import { describe, expect, test } from 'vitest'; +import { + compiledEventDefinitionEquality, + compileEventDefinitionsToCompilerModel, + DEFAULT_HYDRATION_STATE, + deserializeSyncPlan, + nodeSqlite, + PrecompiledSyncConfig, + serializeSyncPlan, + SqlSyncRules +} from '../../../src/index.js'; +import { TestSourceTable } from '../util.js'; +import { yamlToSyncPlan } from './utils.js'; + +const CHECKPOINT_EVENT_YAML = ` +config: + edition: 3 + +streams: + checkpoints: + query: SELECT * FROM checkpoints + +event_definitions: + write_checkpoints: + payloads: + - SELECT user_id, checkpoint, client_id FROM checkpoints WHERE active = true +`; + +describe('compiled replication events', () => { + test('compiles, serializes and evaluates event payload queries', () => { + const { config, errors } = SqlSyncRules.fromYaml(CHECKPOINT_EVENT_YAML, { + defaultSchema: 'test_schema', + throwOnError: false + }); + expect(errors).toStrictEqual([]); + expect(config).toBeInstanceOf(PrecompiledSyncConfig); + + const compiled = config as PrecompiledSyncConfig; + const serialized = serializeSyncPlan(compiled.plan); + // Compiled events are additive and do not require a new plan version. + expect(serialized.version).toBe(1); + expect(serialized.events).toHaveLength(1); + const deserialized = deserializeSyncPlan(JSON.parse(JSON.stringify(serialized))); + expect(deserialized.events).toMatchObject(compiled.plan.events); + + const hydrated = compiled.hydrate({ hydrationState: DEFAULT_HYDRATION_STATE, sqlite: nodeSqlite(sqlite) }); + const event = hydrated.eventDescriptors[0]; + expect(event.name).toBe('write_checkpoints'); + const checkpoints = new TestSourceTable('checkpoints'); + + expect( + event.evaluateRowWithErrors({ + sourceTable: checkpoints, + record: { user_id: 'user-1', checkpoint: 42n, client_id: 'client-1', active: 1, ignored: 'value' } + }) + ).toEqual({ + result: { data: { user_id: 'user-1', checkpoint: 42n, client_id: 'client-1' } }, + errors: [] + }); + expect( + event.evaluateRowWithErrors({ + sourceTable: checkpoints, + record: { user_id: 'user-1', checkpoint: 42n, client_id: 'client-1', active: 0 } + }) + ).toEqual({ errors: [] }); + }); + + test('matches definitions independent of formatting, equivalent operand order and payload-query order', () => { + const first = eventDefinitionFromQueries( + 'SELECT user_id, checkpoint FROM checkpoints WHERE active = true AND checkpoint > 0', + 'SELECT user_id, checkpoint FROM archived_checkpoints' + ); + const equivalent = eventDefinitionFromQueries( + ' select "user_id", "checkpoint" from "archived_checkpoints" ', + 'select "user_id", "checkpoint" from "checkpoints" where 0 < "checkpoint" and true = "active"' + ); + const changed = eventDefinitionFromQueries( + 'SELECT user_id, checkpoint FROM checkpoints WHERE active = true AND checkpoint > 1', + 'SELECT user_id, checkpoint FROM archived_checkpoints' + ); + + expect(compiledEventDefinitionEquality.equals(first, equivalent)).toBe(true); + expect(compiledEventDefinitionEquality.equals(first, changed)).toBe(false); + }); + + test('excludes the raw SQL compatibility mirror from equality', () => { + const original = eventDefinitionFromQueries('SELECT user_id, checkpoint FROM checkpoints WHERE active = true'); + const differentSql = eventDefinitionFromQueries('SELECT user_id, checkpoint FROM checkpoints WHERE active = true'); + differentSql.sourceQueries[0].sql = 'raw sql is compatibility metadata only'; + + expect(compiledEventDefinitionEquality.equals(original, differentSql)).toBe(true); + }); + + test('matches independent of the default schema used to compile', () => { + const query = 'SELECT user_id, checkpoint FROM checkpoints WHERE active = true'; + + expect( + compiledEventDefinitionEquality.equals( + eventDefinitionForSchema('first_schema', query), + eventDefinitionForSchema('second_schema', query) + ) + ).toBe(true); + }); + + test.each([ + ['joins', 'SELECT c.user_id FROM checkpoints c JOIN users u ON u.id = c.user_id', 'single physical source table'], + [ + 'request parameters', + 'SELECT user_id FROM checkpoints WHERE user_id = auth.user_id()', + 'cannot depend on request parameters' + ] + ])('rejects %s', (_label, query, message) => { + const [errors] = yamlToSyncPlan(yamlWithEventQueries(query), { + defaultSchema: 'test_schema', + throwOnError: false + }); + + expect(errors.map((error) => error.message)).toContainEqual(expect.stringContaining(message)); + }); + + test('requires payload queries within an event to use unique source tables', () => { + const [errors] = yamlToSyncPlan( + yamlWithEventQueries( + 'SELECT user_id FROM checkpoints', + 'SELECT checkpoint FROM test_schema.checkpoints WHERE checkpoint > 0' + ), + { defaultSchema: 'test_schema', throwOnError: false } + ); + + expect(errors.map((error) => error.message)).toContain('Each payload query should query a unique table'); + }); +}); + +function eventDefinitionFromQueries(...queries: string[]) { + return eventDefinitionForSchema('test_schema', ...queries); +} + +function eventDefinitionForSchema(defaultSchema: string, ...queries: string[]) { + const compiled = compileEventDefinitionsToCompilerModel({ write_checkpoints: queries }, { defaultSchema }); + expect(compiled.errors.filter((error) => error.type == 'fatal')).toEqual([]); + + return compiled.events[0]; +} + +function yamlWithEventQueries(...queries: string[]): string { + return ` +config: + edition: 3 + +streams: + checkpoints: + query: SELECT * FROM checkpoints + +event_definitions: + write_checkpoints: + payloads: +${queries.map((query) => ` - ${query}`).join('\n')} +`; +} diff --git a/packages/sync-rules/test/src/compiler/utils.ts b/packages/sync-rules/test/src/compiler/utils.ts index 21cfd6ccb..16a9ab7f4 100644 --- a/packages/sync-rules/test/src/compiler/utils.ts +++ b/packages/sync-rules/test/src/compiler/utils.ts @@ -69,7 +69,7 @@ function compileSingleStream(...sql: string[]): [TranslationError[], SyncPlan] { } builder.finish(); - const originalPlan = compiler.output.toSyncPlan(); + const originalPlan = compiler.toSyncPlan(); // Add a serialization roundtrip to ensure sync plans are correctly evaluated even after being deserialized. const afterSerializationRoundtrip = deserializeSyncPlan(JSON.parse(JSON.stringify(serializeSyncPlan(originalPlan)))); diff --git a/packages/sync-rules/test/src/legacy/sync_rules.test.ts b/packages/sync-rules/test/src/legacy/sync_rules.test.ts index dc6be9efe..b0cf44b76 100644 --- a/packages/sync-rules/test/src/legacy/sync_rules.test.ts +++ b/packages/sync-rules/test/src/legacy/sync_rules.test.ts @@ -1108,7 +1108,19 @@ event_definitions: PARSE_OPTIONS ); expect(errors).toStrictEqual([]); - expect(rules.eventDescriptors).toHaveLength(1); + expect(rules).not.toHaveProperty('eventDescriptors'); + expect(rules.eventDefinitions).toHaveLength(1); + + const [event] = rules.hydrate(hydrationParams).eventDescriptors; + expect( + event.evaluateRowWithErrors({ + sourceTable: new TestSourceTable('checkpoints'), + record: { user_id: 'user-1', checkpoint: 2n, client_id: 'client-1' } + }) + ).toEqual({ + result: { data: { user_id: 'user-1', checkpoint: 2n, client_id: 'client-1' } }, + errors: [] + }); }); test('suggests upgrading for streams on edition 2', () => { diff --git a/packages/sync-rules/test/src/sync_plan/evaluator/table_valued.test.ts b/packages/sync-rules/test/src/sync_plan/evaluator/table_valued.test.ts index 9d1bc3519..d64da9a40 100644 --- a/packages/sync-rules/test/src/sync_plan/evaluator/table_valued.test.ts +++ b/packages/sync-rules/test/src/sync_plan/evaluator/table_valued.test.ts @@ -130,7 +130,8 @@ streams: dataSources: [source], buckets: [{ hashCode: 0, uniqueName: 'a', sources: [source] }], parameterIndexes: [], - streams: [] + streams: [], + events: [] }) ); diff --git a/packages/sync-rules/test/src/sync_plan/schema_inference.test.ts b/packages/sync-rules/test/src/sync_plan/schema_inference.test.ts index cf39c98da..8bb292e50 100644 --- a/packages/sync-rules/test/src/sync_plan/schema_inference.test.ts +++ b/packages/sync-rules/test/src/sync_plan/schema_inference.test.ts @@ -39,7 +39,7 @@ describe('schema inference', () => { function generateSchema(...queries: string[]) { const serializedPlan = compileSingleStreamAndSerialize(...queries); const plan = deserializeSyncPlan(serializedPlan); - const rules = new PrecompiledSyncConfig(plan, new CompatibilityContext({ edition: 3 }), [], { + const rules = new PrecompiledSyncConfig(plan, new CompatibilityContext({ edition: 3 }), { sourceText: '', defaultSchema: 'test_schema' });