POC of prefiltering.

Author: rkistner

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

@@ -105,19 +105,19 @@ export class MergedSyncRules implements RowProcessor {
    * @param table The source database table definition, _not_ the individually derived SourceTables.
    * @returns
    */
-  getMatchingSources(table: SourceTableInterface): {
+  getMatchingSources(pattern: TablePattern): {
     bucketDataSources: BucketDataSource[];
     parameterIndexLookupCreators: ParameterIndexLookupCreator[];
   } {
     const bucketDataSources = [...this.resolvedDataSources.values()]
       .map((dataSource) => dataSource.source)
-      .filter((ds) => ds.tableSyncsData(table));
+      .filter((ds) => ds.getSourceTables().some((table) => table.equals(pattern)));
 
     const parameterIndexLookupCreators: ParameterIndexLookupCreator[] = [
       ...this.resolvedParameterLookupSources.values()
     ]
       .map((dataSource) => dataSource.source)
-      .filter((ds) => ds.tableSyncsParameters(table));
+      .filter((ds) => ds.getSourceTables().some((table) => table.equals(pattern)));
     return {
       bucketDataSources,
       parameterIndexLookupCreators
@@ -128,7 +128,6 @@ export class MergedSyncRules implements RowProcessor {
   compatibility: CompatibilityContext = CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY;
 
   getSourceTables(): TablePattern[] {
-    console.log('tables', this.sourcePatterns);
     return this.sourcePatterns;
   }
 

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

@@ -284,7 +284,8 @@ export class MongoBucketDataWriter implements storage.BucketDataWriter {
       replicaIdColumns: ref.replicaIdColumns,
       snapshotComplete: doc.snapshot_done ?? true,
       bucketDataSourceIds: doc.bucket_data_source_ids ?? [],
-      parameterLookupSourceIds: doc.parameter_lookup_source_ids ?? []
+      parameterLookupSourceIds: doc.parameter_lookup_source_ids ?? [],
+      pattern: ref.pattern
     });
     sourceTable.snapshotStatus =
       doc.snapshot_status == null
@@ -302,11 +303,7 @@ export class MongoBucketDataWriter implements storage.BucketDataWriter {
   }
 
   async resolveTables(options: storage.ResolveTablesOptions): Promise<storage.ResolveTablesResult> {
-    const sources = this.rowProcessor.getMatchingSources({
-      connectionTag: options.connection_tag,
-      name: options.entity_descriptor.name,
-      schema: options.entity_descriptor.schema
-    });
+    const sources = this.rowProcessor.getMatchingSources(options.pattern);
     const bucketDataSourceIds = sources.bucketDataSources.map((source) => this.mapping.bucketSourceId(source));
     const parameterLookupSourceIds = sources.parameterIndexLookupCreators.map((source) =>
       this.mapping.parameterLookupId(source)
@@ -390,7 +387,8 @@ export class MongoBucketDataWriter implements storage.BucketDataWriter {
           replicaIdColumns: replicaIdColumns,
           snapshotComplete: doc.snapshot_done ?? true,
           bucketDataSourceIds: doc.bucket_data_source_ids ?? [],
-          parameterLookupSourceIds: doc.parameter_lookup_source_ids ?? []
+          parameterLookupSourceIds: doc.parameter_lookup_source_ids ?? [],
+          pattern: options.pattern
         });
         sourceTable.snapshotStatus =
           doc.snapshot_status == null

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

@@ -34,6 +34,8 @@ import {
 } from './MongoRelation.js';
 import { MongoSnapshotter } from './MongoSnapshotter.js';
 import { CHECKPOINTS_COLLECTION, timestampToDate } from './replication-utils.js';
+import { staticFilterToMongoExpression } from './staticFilters.js';
+import { inspect } from 'node:util';
 
 export interface ChangeStreamOptions {
   connections: MongoManager;
@@ -212,6 +214,7 @@ export class ChangeStream {
   private getSourceNamespaceFilters(writer: BucketDataWriter): {
     $match: any;
     multipleDatabases: boolean;
+    filters: any[];
   } {
     const sourceTables = writer.rowProcessor.getSourceTables();
 
@@ -230,20 +233,31 @@ export class ChangeStream {
         multipleDatabases = true;
       }
 
+      let filterExpression =
+        tablePattern.filter == null
+          ? { $literal: true }
+          : staticFilterToMongoExpression(tablePattern.filter, { columnPrefix: '$fullDocument.' });
+
       if (tablePattern.isWildcard) {
         $refilters.push({
           'ns.db': tablePattern.schema,
           'ns.coll': new RegExp('^' + escapeRegExp(tablePattern.tablePrefix))
         });
         filters.push({
           'ns.db': tablePattern.schema,
-          'ns.coll': new RegExp('^' + escapeRegExp(tablePattern.tablePrefix))
+          'ns.coll': new RegExp('^' + escapeRegExp(tablePattern.tablePrefix)),
+          $expr: filterExpression
         });
       } else {
         $inFilters.push({
           db: tablePattern.schema,
           coll: tablePattern.name
         });
+        filters.push({
+          'ns.db': tablePattern.schema,
+          'ns.coll': tablePattern.name,
+          $expr: filterExpression
+        });
       }
     }
 
@@ -265,9 +279,9 @@ export class ChangeStream {
       : // collection-level: filter on coll only
         { 'ns.coll': { $in: $inFilters.map((ns) => ns.coll) } };
     if ($refilters.length > 0) {
-      return { $match: { $or: [nsFilter, ...$refilters] }, multipleDatabases };
+      return { $match: { $or: [nsFilter, ...$refilters] }, multipleDatabases, filters };
     }
-    return { $match: nsFilter, multipleDatabases };
+    return { $match: nsFilter, multipleDatabases, filters };
   }
 
   private async checkPostImages(db: string, collectionInfo: mongo.CollectionInfo) {
@@ -420,6 +434,12 @@ export class ChangeStream {
       {
         $match: filters.$match
       },
+      // Not working currently - getting "resumeToken not found"
+      // {
+      //   $match: {
+      //     $or: filters.filters
+      //   }
+      // },
       { $changeStreamSplitLargeEvent: {} }
     ];
 
@@ -485,37 +505,56 @@ export class ChangeStream {
       // Ignore the postImages check in this case.
     }
 
-    const result = await writer.resolveTables({
-      connection_id: this.connection_id,
-      connection_tag: this.connections.connectionTag,
-      entity_descriptor: descriptor
+    // What happens here:
+    // 1. We see a new collection that we haven't observed before.
+    // 2. We check which table pattern(s) match this collection, _regardless of specific row filters_.
+    // 3. We resolve the tables for those patterns.
+
+    // FIXME: don't scan through it all
+    // FIXME: handle wildcards
+    const patterns = writer.rowProcessor.getSourceTables().filter((t) => {
+      return (
+        t.connectionTag == this.connections.connectionTag && t.name == descriptor.name && t.schema == descriptor.schema
+      );
     });
 
-    const snapshot = options.snapshot;
-    this.relationCache.set(getCacheIdentifier(descriptor), result.tables);
+    let allTables: SourceTable[] = [];
+    for (let pattern of patterns) {
+      const result = await writer.resolveTables({
+        connection_id: this.connection_id,
+        connection_tag: this.connections.connectionTag,
+        entity_descriptor: descriptor,
+        pattern: pattern
+      });
 
-    // Drop conflicting collections.
-    // This is generally not expected for MongoDB source dbs, so we log an error.
-    if (result.dropTables.length > 0) {
-      this.logger.error(
-        `Conflicting collections found for ${JSON.stringify(descriptor)}. Dropping: ${result.dropTables.map((t) => t.id).join(', ')}`
-      );
-      await writer.drop(result.dropTables);
-    }
+      const snapshot = options.snapshot;
+      this.relationCache.set(getCacheIdentifier(descriptor), result.tables);
 
-    // Snapshot if:
-    // 1. Snapshot is requested (false for initial snapshot, since that process handles it elsewhere)
-    // 2. Snapshot is not already done, AND:
-    // 3. The table is used in sync rules.
-    for (let table of result.tables) {
-      const shouldSnapshot = snapshot && !table.snapshotComplete && table.syncAny;
-      if (shouldSnapshot) {
-        this.logger.info(`New collection: ${descriptor.schema}.${descriptor.name}`);
-        await this.snapshotter.queueSnapshot(writer, table);
+      // Drop conflicting collections.
+      // This is generally not expected for MongoDB source dbs, so we log an error.
+      if (result.dropTables.length > 0) {
+        this.logger.error(
+          `Conflicting collections found for ${JSON.stringify(descriptor)}. Dropping: ${result.dropTables.map((t) => t.id).join(', ')}`
+        );
+        await writer.drop(result.dropTables);
       }
+
+      // Snapshot if:
+      // 1. Snapshot is requested (false for initial snapshot, since that process handles it elsewhere)
+      // 2. Snapshot is not already done, AND:
+      // 3. The table is used in sync rules.
+      for (let table of result.tables) {
+        const shouldSnapshot = snapshot && !table.snapshotComplete && table.syncAny;
+        if (shouldSnapshot) {
+          this.logger.info(`New collection: ${descriptor.schema}.${descriptor.name}`);
+          await this.snapshotter.queueSnapshot(writer, table);
+        }
+      }
+
+      allTables.push(...result.tables);
     }
 
-    return result.tables;
+    return allTables;
   }
 
   private async drop(writer: storage.BucketDataWriter, entity: SourceEntityDescriptor): Promise<void> {

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

@@ -1,6 +1,8 @@
 import { mongo } from '@powersync/lib-service-mongodb';
 import { ReplicationAssertionError } from '@powersync/lib-services-framework';
 import { bson } from '@powersync/service-core';
+import { MongoExpression, StaticFilter } from '@powersync/service-sync-rules';
+import { staticFilterToMongoExpression } from './staticFilters.js';
 
 /**
  * Performs a collection snapshot query, chunking by ranges of _id.
@@ -13,12 +15,21 @@ export class ChunkedSnapshotQuery implements AsyncDisposable {
   private lastCursor: mongo.FindCursor | null = null;
   private collection: mongo.Collection;
   private batchSize: number;
+  private filter: MongoExpression | null = null;
 
-  public constructor(options: { collection: mongo.Collection; batchSize: number; key?: Uint8Array | null }) {
+  public constructor(options: {
+    collection: mongo.Collection;
+    batchSize: number;
+    key?: Uint8Array | null;
+    filter?: StaticFilter;
+  }) {
     this.lastKey = options.key ? bson.deserialize(options.key, { useBigInt64: true })._id : null;
     this.lastCursor = null;
     this.collection = options.collection;
     this.batchSize = options.batchSize;
+    if (options.filter) {
+      this.filter = staticFilterToMongoExpression(options.filter);
+    }
   }
 
   async nextChunk(): Promise<{ docs: mongo.Document[]; lastKey: Uint8Array } | { docs: []; lastKey: null }> {
@@ -35,8 +46,17 @@ export class ChunkedSnapshotQuery implements AsyncDisposable {
       // any parsing as an operator.
       // Starting in MongoDB 5.0, this filter can use the _id index. Source:
       // https://www.mongodb.com/docs/manual/release-notes/5.0/#general-aggregation-improvements
-      const filter: mongo.Filter<mongo.Document> =
-        this.lastKey == null ? {} : { $expr: { $gt: ['$_id', { $literal: this.lastKey }] } };
+      let filter: mongo.Filter<mongo.Document>;
+      if (this.lastKey == null) {
+        filter = this.filter == null ? {} : { $expr: this.filter };
+      } else {
+        if (this.filter == null) {
+          filter = { $expr: { $gt: ['$_id', { $literal: this.lastKey }] } };
+        } else {
+          filter = { $and: [{ $expr: { $gt: ['$_id', { $literal: this.lastKey }] } }, { $expr: this.filter }] };
+        }
+      }
+
       cursor = this.collection.find(filter, {
         readConcern: 'majority',
         limit: this.batchSize,

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

@@ -266,7 +266,8 @@ export class MongoSnapshotter {
       const sourceTables = await writer.resolveTables({
         connection_id: this.connection_id,
         connection_tag: this.connections.connectionTag,
-        entity_descriptor: getMongoRelation({ db: schema, coll: collection.name })
+        entity_descriptor: getMongoRelation({ db: schema, coll: collection.name }),
+        pattern: tablePattern
       });
       // TODO: dropTables?
       result.push(...sourceTables.tables);
@@ -280,10 +281,12 @@ export class MongoSnapshotter {
     let at = table.snapshotStatus?.replicatedCount ?? 0;
     const db = this.client.db(table.schema);
     const collection = db.collection(table.name);
+    console.log('snapshot with filter', table.pattern?.filter, table.pattern);
     await using query = new ChunkedSnapshotQuery({
       collection,
       key: table.snapshotStatus?.lastKey,
-      batchSize: this.snapshotChunkLength
+      batchSize: this.snapshotChunkLength,
+      filter: table.pattern?.filter
     });
     if (query.lastKey != null) {
       this.logger.info(