Granular sync rule parsing (#432)

* Initial split of data vs parameter sources.

* Split out sync rule definitions from compiled ones.

* Update APIs to use hydrated sync rules.

* Refactor parameter queries.

* Fix tests.

* Remove redundant functions.

* Split out parameter lookup sources from parameter querier sources.

* Bring back BucketSource.

* Cleanup.

* Cleanup imports.

* Split out SyncStream into its components.

* Support multiple data sources.

* Use separate sources per sync stream variant.

* Move debugRepresentation() back to BucketSource.

* Split out dataSource on SqlBucketDescriptor.

* Move SubqueryParameterLookupSource deeper to be specific to variants.

* Use HydrationState for bucket names.

* Remove some unused descriptorName references.

* Refactor parameter lookups to make the names configurable.

* More minor refactoring.

* Some comments.

* Import cleanup.

* Remove BucketIdTransformer.

* Add a stream test with custom hydrationState.

* Add changeset.

* Remove need for hydration on BucketDataSource.

* Rename defaultBucketPrefix -> uniqueName.

* Remove hydration for ParameterLookupSource.

* Rename BucketParameterLookupSource -> ParameterIndexLookupCreator.

* Merge BucketParameterQuerierSourceDefinition into HydratedBucketSource.

* Remove HydrationState generics.

We can re-add it if we actually need it later.

* Add some hydration tests.

* Add "end-to-end" sync rules test.

Author: rkistner

.changeset/fresh-geckos-develop.md

@@ -0,0 +1,13 @@
+---
+'@powersync/service-module-postgres-storage': minor
+'@powersync/service-module-mongodb-storage': minor
+'@powersync/service-core-tests': minor
+'@powersync/service-module-postgres': minor
+'@powersync/service-module-mongodb': minor
+'@powersync/service-core': minor
+'@powersync/service-module-mssql': minor
+'@powersync/service-module-mysql': minor
+'@powersync/service-sync-rules': minor
+---
+
+[Internal] Refactor sync rule representation to split out the parsed definitions from the hydrated state.

modules/module-mongodb-storage/src/storage/implementation/MongoBucketBatch.ts

@@ -1,14 +1,14 @@
 import { mongo } from '@powersync/lib-service-mongodb';
-import { SqlEventDescriptor, SqliteRow, SqliteValue, SqlSyncRules } from '@powersync/service-sync-rules';
+import { SqlEventDescriptor, SqliteRow, SqliteValue, HydratedSyncRules } from '@powersync/service-sync-rules';
 import * as bson from 'bson';
 
 import {
   BaseObserver,
   container,
+  logger as defaultLogger,
   ErrorCode,
   errors,
   Logger,
-  logger as defaultLogger,
   ReplicationAssertionError,
   ServiceError
 } from '@powersync/lib-services-framework';
@@ -22,13 +22,13 @@ import {
   utils
 } from '@powersync/service-core';
 import * as timers from 'node:timers/promises';
+import { idPrefixFilter } from '../../utils/util.js';
 import { PowerSyncMongo } from './db.js';
 import { CurrentBucket, CurrentDataDocument, SourceKey, SyncRuleDocument } from './models.js';
 import { MongoIdSequence } from './MongoIdSequence.js';
 import { batchCreateCustomWriteCheckpoints } from './MongoWriteCheckpointAPI.js';
 import { cacheKey, OperationBatch, RecordOperation } from './OperationBatch.js';
 import { PersistedBatch } from './PersistedBatch.js';
-import { idPrefixFilter } from '../../utils/util.js';
 
 /**
  * 15MB
@@ -44,7 +44,7 @@ const replicationMutex = new utils.Mutex();
 
 export interface MongoBucketBatchOptions {
   db: PowerSyncMongo;
-  syncRules: SqlSyncRules;
+  syncRules: HydratedSyncRules;
   groupId: number;
   slotName: string;
   lastCheckpointLsn: string | null;
@@ -71,7 +71,7 @@ export class MongoBucketBatch
   private readonly client: mongo.MongoClient;
   public readonly db: PowerSyncMongo;
   public readonly session: mongo.ClientSession;
-  private readonly sync_rules: SqlSyncRules;
+  private readonly sync_rules: HydratedSyncRules;
 
   private readonly group_id: number;
 
@@ -474,8 +474,7 @@ export class MongoBucketBatch
       if (sourceTable.syncData) {
         const { results: evaluated, errors: syncErrors } = this.sync_rules.evaluateRowWithErrors({
           record: after,
-          sourceTable,
-          bucketIdTransformer: SqlSyncRules.versionedBucketIdTransformer(`${this.group_id}`)
+          sourceTable
         });
 
         for (let error of syncErrors) {

modules/module-mongodb-storage/src/storage/implementation/MongoPersistedSyncRules.ts

@@ -1,6 +1,7 @@
-import { SqlSyncRules } from '@powersync/service-sync-rules';
+import { SqlSyncRules, HydratedSyncRules } from '@powersync/service-sync-rules';
 
 import { storage } from '@powersync/service-core';
+import { versionedHydrationState } from '@powersync/service-sync-rules/src/HydrationState.js';
 
 export class MongoPersistedSyncRules implements storage.PersistedSyncRules {
   public readonly slot_name: string;
@@ -13,4 +14,8 @@ export class MongoPersistedSyncRules implements storage.PersistedSyncRules {
   ) {
     this.slot_name = slot_name ?? `powersync_${id}`;
   }
+
+  hydratedSyncRules(): HydratedSyncRules {
+    return this.sync_rules.hydrate({ hydrationState: versionedHydrationState(this.id) });
+  }
 }

modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts

@@ -10,7 +10,6 @@ import {
   BroadcastIterable,
   CHECKPOINT_INVALIDATE_ALL,
   CheckpointChanges,
-  CompactOptions,
   deserializeParameterLookup,
   GetCheckpointChangesOptions,
   InternalOpId,
@@ -25,10 +24,11 @@ import {
   WatchWriteCheckpointOptions
 } from '@powersync/service-core';
 import { JSONBig } from '@powersync/service-jsonbig';
-import { ParameterLookup, SqliteJsonRow, SqlSyncRules } from '@powersync/service-sync-rules';
+import { HydratedSyncRules, ScopedParameterLookup, SqliteJsonRow } from '@powersync/service-sync-rules';
 import * as bson from 'bson';
 import { LRUCache } from 'lru-cache';
 import * as timers from 'timers/promises';
+import { idPrefixFilter, mapOpEntry, readSingleBatch, setSessionSnapshotTime } from '../../utils/util.js';
 import { MongoBucketStorage } from '../MongoBucketStorage.js';
 import { PowerSyncMongo } from './db.js';
 import { BucketDataDocument, BucketDataKey, BucketStateDocument, SourceKey, SourceTableDocument } from './models.js';
@@ -37,7 +37,6 @@ import { MongoChecksumOptions, MongoChecksums } from './MongoChecksums.js';
 import { MongoCompactor } from './MongoCompactor.js';
 import { MongoParameterCompactor } from './MongoParameterCompactor.js';
 import { MongoWriteCheckpointAPI } from './MongoWriteCheckpointAPI.js';
-import { idPrefixFilter, mapOpEntry, readSingleBatch, setSessionSnapshotTime } from '../../utils/util.js';
 
 export interface MongoSyncBucketStorageOptions {
   checksumOptions?: MongoChecksumOptions;
@@ -61,7 +60,7 @@ export class MongoSyncBucketStorage
   private readonly db: PowerSyncMongo;
   readonly checksums: MongoChecksums;
 
-  private parsedSyncRulesCache: { parsed: SqlSyncRules; options: storage.ParseSyncRulesOptions } | undefined;
+  private parsedSyncRulesCache: { parsed: HydratedSyncRules; options: storage.ParseSyncRulesOptions } | undefined;
   private writeCheckpointAPI: MongoWriteCheckpointAPI;
 
   constructor(
@@ -101,14 +100,14 @@ export class MongoSyncBucketStorage
     });
   }
 
-  getParsedSyncRules(options: storage.ParseSyncRulesOptions): SqlSyncRules {
+  getParsedSyncRules(options: storage.ParseSyncRulesOptions): HydratedSyncRules {
     const { parsed, options: cachedOptions } = this.parsedSyncRulesCache ?? {};
     /**
      * Check if the cached sync rules, if present, had the same options.
      * Parse sync rules if the options are different or if there is no cached value.
      */
     if (!parsed || options.defaultSchema != cachedOptions?.defaultSchema) {
-      this.parsedSyncRulesCache = { parsed: this.sync_rules.parsed(options).sync_rules, options };
+      this.parsedSyncRulesCache = { parsed: this.sync_rules.parsed(options).hydratedSyncRules(), options };
     }
 
     return this.parsedSyncRulesCache!.parsed;
@@ -170,7 +169,7 @@ export class MongoSyncBucketStorage
     await using batch = new MongoBucketBatch({
       logger: options.logger,
       db: this.db,
-      syncRules: this.sync_rules.parsed(options).sync_rules,
+      syncRules: this.sync_rules.parsed(options).hydratedSyncRules(),
       groupId: this.group_id,
       slotName: this.slot_name,
       lastCheckpointLsn: checkpoint_lsn,
@@ -293,7 +292,10 @@ export class MongoSyncBucketStorage
     return result!;
   }
 
-  async getParameterSets(checkpoint: MongoReplicationCheckpoint, lookups: ParameterLookup[]): Promise<SqliteJsonRow[]> {
+  async getParameterSets(
+    checkpoint: MongoReplicationCheckpoint,
+    lookups: ScopedParameterLookup[]
+  ): Promise<SqliteJsonRow[]> {
     return this.db.client.withSession({ snapshot: true }, async (session) => {
       // Set the session's snapshot time to the checkpoint's snapshot time.
       // An alternative would be to create the session when the checkpoint is created, but managing
@@ -1025,7 +1027,7 @@ class MongoReplicationCheckpoint implements ReplicationCheckpoint {
     public snapshotTime: mongo.Timestamp
   ) {}
 
-  async getParameterSets(lookups: ParameterLookup[]): Promise<SqliteJsonRow[]> {
+  async getParameterSets(lookups: ScopedParameterLookup[]): Promise<SqliteJsonRow[]> {
     return this.storage.getParameterSets(this, lookups);
   }
 }
@@ -1034,7 +1036,7 @@ class EmptyReplicationCheckpoint implements ReplicationCheckpoint {
   readonly checkpoint: InternalOpId = 0n;
   readonly lsn: string | null = null;
 
-  async getParameterSets(lookups: ParameterLookup[]): Promise<SqliteJsonRow[]> {
+  async getParameterSets(lookups: ScopedParameterLookup[]): Promise<SqliteJsonRow[]> {
     return [];
   }
 }

modules/module-mongodb/src/replication/ChangeStream.ts

@@ -10,15 +10,20 @@ import {
   ServiceError
 } from '@powersync/lib-services-framework';
 import {
-  InternalOpId,
   MetricsEngine,
   RelationCache,
   SaveOperationTag,
   SourceEntityDescriptor,
   SourceTable,
   storage
 } from '@powersync/service-core';
-import { DatabaseInputRow, SqliteInputRow, SqliteRow, SqlSyncRules, TablePattern } from '@powersync/service-sync-rules';
+import {
+  DatabaseInputRow,
+  SqliteInputRow,
+  SqliteRow,
+  HydratedSyncRules,
+  TablePattern
+} from '@powersync/service-sync-rules';
 import { ReplicationMetric } from '@powersync/service-types';
 import { MongoLSN } from '../common/MongoLSN.js';
 import { PostImagesOption } from '../types/types.js';
@@ -75,7 +80,7 @@ export class ChangeStreamInvalidatedError extends DatabaseConnectionError {
 }
 
 export class ChangeStream {
-  sync_rules: SqlSyncRules;
+  sync_rules: HydratedSyncRules;
   group_id: number;
 
   connection_id = 1;

modules/module-mssql/src/replication/CDCStream.ts

@@ -10,7 +10,13 @@ import {
 } from '@powersync/lib-services-framework';
 import { getUuidReplicaIdentityBson, MetricsEngine, SourceEntityDescriptor, storage } from '@powersync/service-core';
 
-import { SqliteInputRow, SqliteRow, SqlSyncRules, TablePattern } from '@powersync/service-sync-rules';
+import {
+  SqliteInputRow,
+  SqliteRow,
+  SqlSyncRules,
+  HydratedSyncRules,
+  TablePattern
+} from '@powersync/service-sync-rules';
 
 import { ReplicationMetric } from '@powersync/service-types';
 import { BatchedSnapshotQuery, MSSQLSnapshotQuery, SimpleSnapshotQuery } from './MSSQLSnapshotQuery.js';
@@ -82,7 +88,7 @@ export class CDCDataExpiredError extends DatabaseConnectionError {
 }
 
 export class CDCStream {
-  private readonly syncRules: SqlSyncRules;
+  private readonly syncRules: HydratedSyncRules;
   private readonly storage: storage.SyncRulesBucketStorage;
   private readonly connections: MSSQLConnectionManager;
   private readonly abortSignal: AbortSignal;

modules/module-mysql/src/replication/BinLogStream.ts

@@ -61,7 +61,7 @@ function createTableId(schema: string, tableName: string): string {
 }
 
 export class BinLogStream {
-  private readonly syncRules: sync_rules.SqlSyncRules;
+  private readonly syncRules: sync_rules.HydratedSyncRules;
   private readonly groupId: number;
 
   private readonly storage: storage.SyncRulesBucketStorage;

modules/module-postgres-storage/src/storage/PostgresSyncRulesStorage.ts

@@ -4,7 +4,6 @@ import {
   BucketChecksum,
   CHECKPOINT_INVALIDATE_ALL,
   CheckpointChanges,
-  CompactOptions,
   GetCheckpointChangesOptions,
   InternalOpId,
   internalToExternalOpId,
@@ -60,7 +59,9 @@ export class PostgresSyncRulesStorage
   protected writeCheckpointAPI: PostgresWriteCheckpointAPI;
 
   //   TODO we might be able to share this in an abstract class
-  private parsedSyncRulesCache: { parsed: sync_rules.SqlSyncRules; options: storage.ParseSyncRulesOptions } | undefined;
+  private parsedSyncRulesCache:
+    | { parsed: sync_rules.HydratedSyncRules; options: storage.ParseSyncRulesOptions }
+    | undefined;
   private _checksumCache: storage.ChecksumCache | undefined;
 
   constructor(protected options: PostgresSyncRulesStorageOptions) {
@@ -96,14 +97,14 @@ export class PostgresSyncRulesStorage
   }
 
   //   TODO we might be able to share this in an abstract class
-  getParsedSyncRules(options: storage.ParseSyncRulesOptions): sync_rules.SqlSyncRules {
+  getParsedSyncRules(options: storage.ParseSyncRulesOptions): sync_rules.HydratedSyncRules {
     const { parsed, options: cachedOptions } = this.parsedSyncRulesCache ?? {};
     /**
      * Check if the cached sync rules, if present, had the same options.
      * Parse sync rules if the options are different or if there is no cached value.
      */
     if (!parsed || options.defaultSchema != cachedOptions?.defaultSchema) {
-      this.parsedSyncRulesCache = { parsed: this.sync_rules.parsed(options).sync_rules, options };
+      this.parsedSyncRulesCache = { parsed: this.sync_rules.parsed(options).hydratedSyncRules(), options };
     }
 
     return this.parsedSyncRulesCache!.parsed;
@@ -349,7 +350,7 @@ export class PostgresSyncRulesStorage
     const batch = new PostgresBucketBatch({
       logger: options.logger ?? framework.logger,
       db: this.db,
-      sync_rules: this.sync_rules.parsed(options).sync_rules,
+      sync_rules: this.sync_rules.parsed(options).hydratedSyncRules(),
       group_id: this.group_id,
       slot_name: this.slot_name,
       last_checkpoint_lsn: checkpoint_lsn,
@@ -374,7 +375,7 @@ export class PostgresSyncRulesStorage
 
   async getParameterSets(
     checkpoint: ReplicationCheckpoint,
-    lookups: sync_rules.ParameterLookup[]
+    lookups: sync_rules.ScopedParameterLookup[]
   ): Promise<sync_rules.SqliteJsonRow[]> {
     const rows = await this.db.sql`
       SELECT DISTINCT
@@ -879,7 +880,7 @@ class PostgresReplicationCheckpoint implements storage.ReplicationCheckpoint {
     public readonly lsn: string | null
   ) {}
 
-  getParameterSets(lookups: sync_rules.ParameterLookup[]): Promise<sync_rules.SqliteJsonRow[]> {
+  getParameterSets(lookups: sync_rules.ScopedParameterLookup[]): Promise<sync_rules.SqliteJsonRow[]> {
     return this.storage.getParameterSets(this, lookups);
   }
 }

modules/module-postgres-storage/src/storage/batch/PostgresBucketBatch.ts

@@ -24,7 +24,7 @@ import { PostgresPersistedBatch } from './PostgresPersistedBatch.js';
 export interface PostgresBucketBatchOptions {
   logger: Logger;
   db: lib_postgres.DatabaseClient;
-  sync_rules: sync_rules.SqlSyncRules;
+  sync_rules: sync_rules.HydratedSyncRules;
   group_id: number;
   slot_name: string;
   last_checkpoint_lsn: string | null;
@@ -72,7 +72,7 @@ export class PostgresBucketBatch
   protected persisted_op: InternalOpId | null;
 
   protected write_checkpoint_batch: storage.CustomWriteCheckpointOptions[];
-  protected readonly sync_rules: sync_rules.SqlSyncRules;
+  protected readonly sync_rules: sync_rules.HydratedSyncRules;
   protected batch: OperationBatch | null;
   private lastWaitingLogThrottled = 0;
   private markRecordUnavailable: BucketStorageMarkRecordUnavailable | undefined;
@@ -840,8 +840,7 @@ export class PostgresBucketBatch
       if (sourceTable.syncData) {
         const { results: evaluated, errors: syncErrors } = this.sync_rules.evaluateRowWithErrors({
           record: after,
-          sourceTable,
-          bucketIdTransformer: sync_rules.SqlSyncRules.versionedBucketIdTransformer(`${this.group_id}`)
+          sourceTable
         });
 
         for (const error of syncErrors) {

modules/module-postgres-storage/src/storage/sync-rules/PostgresPersistedSyncRulesContent.ts

@@ -4,6 +4,7 @@ import { storage } from '@powersync/service-core';
 import { SqlSyncRules } from '@powersync/service-sync-rules';
 
 import { models } from '../../types/types.js';
+import { versionedHydrationState } from '@powersync/service-sync-rules/src/HydrationState.js';
 
 export class PostgresPersistedSyncRulesContent implements storage.PersistedSyncRulesContent {
   public readonly slot_name: string;
@@ -35,7 +36,12 @@ export class PostgresPersistedSyncRulesContent implements storage.PersistedSyncR
     return {
       id: this.id,
       slot_name: this.slot_name,
-      sync_rules: SqlSyncRules.fromYaml(this.sync_rules_content, options)
+      sync_rules: SqlSyncRules.fromYaml(this.sync_rules_content, options),
+      hydratedSyncRules() {
+        return this.sync_rules.hydrate({
+          hydrationState: versionedHydrationState(this.id)
+        });
+      }
     };
   }