Use versioned bucket ids

Author: simolus3

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

@@ -45,6 +45,7 @@ const replicationMutex = new utils.Mutex();
 export interface MongoBucketBatchOptions {
   db: PowerSyncMongo;
   syncRules: SqlSyncRules;
+  syncRulesId: string;
   groupId: number;
   slotName: string;
   lastCheckpointLsn: string | null;
@@ -72,6 +73,7 @@ export class MongoBucketBatch
   public readonly db: PowerSyncMongo;
   public readonly session: mongo.ClientSession;
   private readonly sync_rules: SqlSyncRules;
+  private readonly syncRulesId: string;
 
   private readonly group_id: number;
 
@@ -127,6 +129,7 @@ export class MongoBucketBatch
     this.session = this.client.startSession();
     this.slot_name = options.slotName;
     this.sync_rules = options.syncRules;
+    this.syncRulesId = options.syncRulesId;
     this.storeCurrentData = options.storeCurrentData;
     this.skipExistingRows = options.skipExistingRows;
     this.markRecordUnavailable = options.markRecordUnavailable;
@@ -461,7 +464,8 @@ export class MongoBucketBatch
       if (sourceTable.syncData) {
         const { results: evaluated, errors: syncErrors } = this.sync_rules.evaluateRowWithErrors({
           record: after,
-          sourceTable
+          sourceTable,
+          bucketIdTransformer: SqlSyncRules.versionedBucketIdTransformer(this.syncRulesId)
         });
 
         for (let error of syncErrors) {

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

@@ -154,6 +154,7 @@ export class MongoSyncBucketStorage
       logger: options.logger,
       db: this.db,
       syncRules: this.sync_rules.parsed(options).sync_rules,
+      syncRulesId: `${this.sync_rules.id}`,
       groupId: this.group_id,
       slotName: this.slot_name,
       lastCheckpointLsn: checkpoint_lsn,

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

@@ -330,6 +330,7 @@ export class PostgresSyncRulesStorage
       logger: options.logger ?? framework.logger,
       db: this.db,
       sync_rules: this.sync_rules.parsed(options).sync_rules,
+      syncRulesId: `${this.sync_rules.id}`,
       group_id: this.group_id,
       slot_name: this.slot_name,
       last_checkpoint_lsn: checkpoint_lsn,

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

@@ -25,6 +25,7 @@ export interface PostgresBucketBatchOptions {
   logger: Logger;
   db: lib_postgres.DatabaseClient;
   sync_rules: sync_rules.SqlSyncRules;
+  syncRulesId: string;
   group_id: number;
   slot_name: string;
   last_checkpoint_lsn: string | null;
@@ -73,6 +74,7 @@ export class PostgresBucketBatch
 
   protected write_checkpoint_batch: storage.CustomWriteCheckpointOptions[];
   protected readonly sync_rules: sync_rules.SqlSyncRules;
+  private readonly syncRulesId: string;
   protected batch: OperationBatch | null;
   private lastWaitingLogThrottled = 0;
   private markRecordUnavailable: BucketStorageMarkRecordUnavailable | undefined;
@@ -88,6 +90,7 @@ export class PostgresBucketBatch
     this.resumeFromLsn = options.resumeFromLsn;
     this.write_checkpoint_batch = [];
     this.sync_rules = options.sync_rules;
+    this.syncRulesId = options.syncRulesId;
     this.markRecordUnavailable = options.markRecordUnavailable;
     this.batch = null;
     this.persisted_op = null;
@@ -825,7 +828,8 @@ export class PostgresBucketBatch
       if (sourceTable.syncData) {
         const { results: evaluated, errors: syncErrors } = this.sync_rules.evaluateRowWithErrors({
           record: after,
-          sourceTable
+          sourceTable,
+          bucketIdTransformer: sync_rules.SqlSyncRules.versionedBucketIdTransformer(this.syncRulesId)
         });
 
         for (const error of syncErrors) {

packages/sync-rules/src/BaseSqlDataQuery.ts

@@ -5,6 +5,7 @@ import { SourceTableInterface } from './SourceTableInterface.js';
 import { SqlTools } from './sql_filters.js';
 import { TablePattern } from './TablePattern.js';
 import {
+  BucketIdTransformer,
   EvaluationResult,
   QueryParameters,
   QuerySchema,
@@ -25,6 +26,7 @@ export interface EvaluateRowOptions {
   table: SourceTableInterface;
   row: SqliteRow;
   bucketIds: (params: QueryParameters) => string[];
+  bucketIdTransformer: BucketIdTransformer | null;
 }
 
 export interface BaseSqlDataQueryOptions {
@@ -177,7 +179,7 @@ export class BaseSqlDataQuery {
 
   evaluateRowWithOptions(options: EvaluateRowOptions): EvaluationResult[] {
     try {
-      const { table, row, bucketIds } = options;
+      const { table, row, bucketIds, bucketIdTransformer } = options;
 
       const tables = { [this.table]: this.addSpecialParameters(table, row) };
       const resolvedBucketIds = bucketIds(tables);
@@ -197,7 +199,7 @@ export class BaseSqlDataQuery {
 
       return resolvedBucketIds.map((bucketId) => {
         return {
-          bucket: bucketId,
+          bucket: bucketIdTransformer ? bucketIdTransformer(bucketId) : bucketId,
           table: outputTable,
           id: id,
           data

packages/sync-rules/src/SqlBucketDescriptor.ts

@@ -104,7 +104,13 @@ export class SqlBucketDescriptor implements BucketSource {
         continue;
       }
 
-      results.push(...query.evaluateRow(options.sourceTable, applyRowContext(options.record, this.compatibility)));
+      results.push(
+        ...query.evaluateRow(
+          options.sourceTable,
+          applyRowContext(options.record, this.compatibility),
+          options.bucketIdTransformer
+        )
+      );
     }
     return results;
   }

packages/sync-rules/src/SqlDataQuery.ts

@@ -10,7 +10,7 @@ import { checkUnsupportedFeatures, isClauseError } from './sql_support.js';
 import { SyncRulesOptions } from './SqlSyncRules.js';
 import { TablePattern } from './TablePattern.js';
 import { TableQuerySchema } from './TableQuerySchema.js';
-import { EvaluationResult, ParameterMatchClause, QuerySchema, SqliteRow } from './types.js';
+import { BucketIdTransformer, EvaluationResult, ParameterMatchClause, QuerySchema, SqliteRow } from './types.js';
 import { getBucketId, isSelectStatement } from './utils.js';
 
 export interface SqlDataQueryOptions extends BaseSqlDataQueryOptions {
@@ -185,10 +185,15 @@ export class SqlDataQuery extends BaseSqlDataQuery {
     this.filter = options.filter;
   }
 
-  evaluateRow(table: SourceTableInterface, row: SqliteRow): EvaluationResult[] {
+  evaluateRow(
+    table: SourceTableInterface,
+    row: SqliteRow,
+    bucketIdTransformer: BucketIdTransformer | null
+  ): EvaluationResult[] {
     return this.evaluateRowWithOptions({
       table,
       row,
+      bucketIdTransformer,
       bucketIds: (tables) => {
         const bucketParameters = this.filter.filterRow(tables);
         return bucketParameters.map((params) => getBucketId(this.descriptorName, this.bucketParameters, params));

packages/sync-rules/src/SqlSyncRules.ts

@@ -85,6 +85,7 @@ export interface GetBucketParameterQuerierResult {
 export class SqlSyncRules implements SyncRules {
   bucketSources: BucketSource[] = [];
   eventDescriptors: SqlEventDescriptor[] = [];
+  compatibility: CompatibilityContext = CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY;
 
   content: string;
 
@@ -157,6 +158,7 @@ export class SqlSyncRules implements SyncRules {
       }
 
       compatibility = new CompatibilityContext(edition, options);
+      rules.compatibility = compatibility;
     }
 
     // Bucket definitions using explicit parameter and data queries.
@@ -384,9 +386,17 @@ export class SqlSyncRules implements SyncRules {
   }
 
   evaluateRowWithErrors(options: EvaluateRowOptions): { results: EvaluatedRow[]; errors: EvaluationError[] } {
+    const resolvedOptions = this.compatibility.isEnabled(CompatibilityOption.versionedBucketIds)
+      ? options
+      : {
+          ...options,
+          // Disable bucket id transformer when the option is unused.
+          transformBucketIds: null
+        };
+
     let rawResults: EvaluationResult[] = [];
     for (let source of this.bucketSources) {
-      rawResults.push(...source.evaluateRow(options));
+      rawResults.push(...source.evaluateRow(resolvedOptions));
     }
 
     const results = rawResults.filter(isEvaluatedRow) as EvaluatedRow[];
@@ -502,4 +512,8 @@ export class SqlSyncRules implements SyncRules {
       }
     }
   }
+
+  static versionedBucketIdTransformer(version: string) {
+    return (bucketId: string) => `${version}#${bucketId}`;
+  }
 }

packages/sync-rules/src/compatibility.ts

@@ -22,8 +22,15 @@ export class CompatibilityOption {
     CompatibilityEdition.SYNC_STREAMS
   );
 
+  static versionedBucketIds = new CompatibilityOption(
+    'versioned_bucket_ids',
+    'Encode the version of sync rules in generated bucket ids, which avoids clients downloading data twice and improves client-side progress estimates.',
+    CompatibilityEdition.SYNC_STREAMS
+  );
+
   static byName: Record<string, CompatibilityOption> = Object.freeze({
-    timestamps_iso8601: this.timestampsIso8601
+    timestamps_iso8601: this.timestampsIso8601,
+    versioned_bucket_ids: this.versionedBucketIds
   });
 }
 

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

@@ -143,8 +143,6 @@ export class SyncStream implements BucketSource {
         columns: this.data.columnOutputNames()
       }
     };
-
-    throw new Error('Method not implemented.');
   }
 
   debugWriteOutputTables(result: Record<string, { query: string }[]>): void {
@@ -181,6 +179,7 @@ export class SyncStream implements BucketSource {
     return this.data.evaluateRowWithOptions({
       table: options.sourceTable,
       row: applyRowContext(options.record, this.compatibility),
+      bucketIdTransformer: options.bucketIdTransformer,
       bucketIds() {
         const bucketIds: string[] = [];
         for (const variant of stream.variants) {