feat: implement global initial_snapshot_filters for all database modules

Author: raffidahmad

docs/initial-snapshot-filters.md

@@ -288,6 +288,79 @@ bucket_definitions:
       - SELECT * FROM events_2025
 ```
 
+## Special Characters and Identifiers
+
+⚠️ **Important**: Filter expressions are embedded directly into SQL/MongoDB queries. You must properly quote identifiers and escape string literals according to your database's syntax rules.
+
+### SQL Identifier Quoting
+
+**PostgreSQL** - Use double quotes for identifiers with spaces or special characters:
+```yaml
+initial_snapshot_filters:
+  users:
+    sql: "\"User Status\" = 'active' AND \"created-at\" > NOW() - INTERVAL '30 days'"
+```
+
+**MySQL** - Use backticks for identifiers with spaces or special characters:
+```yaml
+initial_snapshot_filters:
+  users:
+    sql: "`User Status` = 'active' AND `created-at` > NOW() - INTERVAL 30 DAY"
+```
+
+**SQL Server** - Use square brackets for identifiers with spaces or special characters:
+```yaml
+initial_snapshot_filters:
+  users:
+    sql: "[User Status] = 'active' AND [created-at] > DATEADD(day, -30, GETDATE())"
+```
+
+### String Literal Escaping
+
+Always use proper escaping for string literals containing quotes:
+
+```yaml
+initial_snapshot_filters:
+  comments:
+    # Single quotes must be escaped as '' in SQL
+    sql: "content NOT LIKE '%can''t%' AND status = 'approved'"
+```
+
+### Complex Expressions with OR Operators
+
+PowerSync automatically wraps your filter in parentheses to prevent operator precedence issues:
+
+```yaml
+initial_snapshot_filters:
+  users:
+    # This is wrapped as: WHERE (status = 'active' OR status = 'pending')
+    sql: "status = 'active' OR status = 'pending'"
+```
+
+### MongoDB Filters
+
+MongoDB filters use native BSON query syntax, which is safer than string concatenation:
+
+```yaml
+initial_snapshot_filters:
+  users:
+    mongo:
+      $or:
+        - status: 'active'
+        - status: 'pending'
+      "special field": { $exists: true }
+```
+
+### Security Considerations
+
+- Filters are defined in `sync_rules.yaml` by administrators, not by end users
+- Filters are static configuration, not dynamic user input
+- Still follow security best practices:
+  - Avoid including sensitive data in filters
+  - Test filters in development before production
+  - Review filter changes carefully during deployment
+- PowerSync does not parameterize filters since they are arbitrary SQL expressions, similar to bucket query definitions
+
 ## Limitations
 
 - Filters are applied **globally** across all buckets using that table

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

@@ -239,7 +239,8 @@ export class MongoSyncBucketStorage
         schema: schema,
         name: name,
         replicaIdColumns: replicaIdColumns,
-        snapshotComplete: doc.snapshot_done ?? true
+        snapshotComplete: doc.snapshot_done ?? true,
+        initialSnapshotFilter: options.sync_rules.definition.getInitialSnapshotFilter(connection_tag, schema, name, 'mongo')
       });
       sourceTable.syncEvent = options.sync_rules.tableTriggersEvent(sourceTable);
       sourceTable.syncData = options.sync_rules.tableSyncsData(sourceTable);

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

@@ -483,7 +483,8 @@ export class ChangeStream {
     await using query = new ChunkedSnapshotQuery({
       collection,
       key: table.snapshotStatus?.lastKey,
-      batchSize: this.snapshotChunkLength
+      batchSize: this.snapshotChunkLength,
+      filter: table.initialSnapshotFilter?.mongo
     });
     if (query.lastKey != null) {
       this.logger.info(
@@ -492,6 +493,9 @@ export class ChangeStream {
     } else {
       this.logger.info(`Replicating ${table.qualifiedName} ${table.formatSnapshotProgress()}`);
     }
+    if (table.initialSnapshotFilter?.mongo) {
+      this.logger.info(`Applying initial snapshot filter: ${JSON.stringify(table.initialSnapshotFilter.mongo)}`);
+    }
 
     let lastBatch = performance.now();
     let nextChunkPromise = query.nextChunk();

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

@@ -13,12 +13,19 @@ export class ChunkedSnapshotQuery implements AsyncDisposable {
   private lastCursor: mongo.FindCursor | null = null;
   private collection: mongo.Collection;
   private batchSize: number;
+  private snapshotFilter: any;
 
-  public constructor(options: { collection: mongo.Collection; batchSize: number; key?: Uint8Array | null }) {
+  public constructor(options: {
+    collection: mongo.Collection;
+    batchSize: number;
+    key?: Uint8Array | null;
+    filter?: any;
+  }) {
     this.lastKey = options.key ? bson.deserialize(options.key, { useBigInt64: true })._id : null;
     this.lastCursor = null;
     this.collection = options.collection;
     this.batchSize = options.batchSize;
+    this.snapshotFilter = options.filter;
   }
 
   async nextChunk(): Promise<{ docs: mongo.Document[]; lastKey: Uint8Array } | { docs: []; lastKey: null }> {
@@ -35,8 +42,23 @@ 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> =
+      
+      // Build base filter for _id
+      const idFilter: mongo.Filter<mongo.Document> =
         this.lastKey == null ? {} : { $expr: { $gt: ['$_id', { $literal: this.lastKey }] } };
+      
+      // Combine with snapshot filter if present
+      let filter: mongo.Filter<mongo.Document>;
+      if (this.snapshotFilter) {
+        if (this.lastKey == null) {
+          filter = this.snapshotFilter;
+        } else {
+          filter = { $and: [idFilter, this.snapshotFilter] };
+        }
+      } else {
+        filter = idFilter;
+      }
+      
       cursor = this.collection.find(filter, {
         readConcern: 'majority',
         limit: this.batchSize,

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

@@ -356,6 +356,11 @@ export class CDCStream {
       query = new SimpleSnapshotQuery(transaction, table);
       replicatedCount = 0;
     }
+    
+    if (table.sourceTable.initialSnapshotFilter?.sql) {
+      this.logger.info(`Applying initial snapshot filter: ${table.sourceTable.initialSnapshotFilter.sql}`);
+    }
+    
     await query.initialize();
 
     let hasRemainingData = true;

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

@@ -44,7 +44,11 @@ export class SimpleSnapshotQuery implements MSSQLSnapshotQuery {
     const request = this.transaction.request();
     const stream = request.toReadableStream();
 
-    request.query(`SELECT * FROM ${this.table.toQualifiedName()}`);
+    let query = `SELECT * FROM ${this.table.toQualifiedName()}`;
+    if (this.table.sourceTable.initialSnapshotFilter?.sql) {
+      query += ` WHERE (${this.table.sourceTable.initialSnapshotFilter.sql})`;
+    }
+    request.query(query);
 
     // MSSQL only streams one row at a time
     for await (const row of stream) {
@@ -141,17 +145,26 @@ export class BatchedSnapshotQuery implements MSSQLSnapshotQuery {
 
     const request = this.transaction.request();
     const stream = request.toReadableStream();
+    const snapshotFilter = this.table.sourceTable.initialSnapshotFilter?.sql;
+    
     if (this.lastKey == null) {
-      request.query(`SELECT TOP(${this.batchSize}) * FROM ${this.table.toQualifiedName()} ORDER BY ${escapedKeyName}`);
+      let query = `SELECT TOP(${this.batchSize}) * FROM ${this.table.toQualifiedName()}`;
+      if (snapshotFilter) {
+        query += ` WHERE (${snapshotFilter})`;
+      }
+      query += ` ORDER BY ${escapedKeyName}`;
+      request.query(query);
     } else {
       if (this.key.typeId == null) {
         throw new Error(`typeId required for primary key ${this.key.name}`);
       }
-      request
-        .input('lastKey', this.lastKey)
-        .query(
-          `SELECT TOP(${this.batchSize}) * FROM ${this.table.toQualifiedName()} WHERE ${escapedKeyName} > @lastKey ORDER BY ${escapedKeyName}`
-        );
+      request.input('lastKey', this.lastKey);
+      let query = `SELECT TOP(${this.batchSize}) * FROM ${this.table.toQualifiedName()} WHERE ${escapedKeyName} > @lastKey`;
+      if (snapshotFilter) {
+        query += ` AND (${snapshotFilter})`;
+      }
+      query += ` ORDER BY ${escapedKeyName}`;
+      request.query(query);
     }
 
     // MSSQL only streams one row at a time

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

@@ -307,17 +307,21 @@ export class BinLogStream {
     batch: storage.BucketStorageBatch,
     table: storage.SourceTable
   ) {
-    this.logger.info(`Replicating ${qualifiedMySQLTable(table)}`);
+    const qualifiedMySQLTableName = qualifiedMySQLTable(table);
+
+    this.logger.info(`Replicating ${qualifiedMySQLTableName}`);
     // TODO count rows and log progress at certain batch sizes
 
     // MAX_EXECUTION_TIME(0) hint disables execution timeout for this query
-    let query = `SELECT /*+ MAX_EXECUTION_TIME(0) */ * FROM ${qualifiedMySQLTable(table)}`;
+    let query = `SELECT /*+ MAX_EXECUTION_TIME(0) */ * FROM ${qualifiedMySQLTableName}`;
 
     // Apply snapshot filter if it exists. This allows us to do partial snapshots,
     // for example for large tables where we only want to snapshot recent data.
     if (table.initialSnapshotFilter?.sql) {
-      query += ` WHERE ${table.initialSnapshotFilter.sql}`;
+      query += ` WHERE (${table.initialSnapshotFilter.sql})`;
       this.logger.info(`Applying initial snapshot filter: ${table.initialSnapshotFilter.sql}`);
+    } else {
+      this.logger.info(`No initial snapshot filter applied for ${qualifiedMySQLTableName}`);
     }
 
     const queryStream = connection.query(query);

modules/module-postgres/src/replication/SnapshotQuery.ts

@@ -36,7 +36,11 @@ export class SimpleSnapshotQuery implements SnapshotQuery {
   ) {}
 
   public async initialize(): Promise<void> {
-    await this.connection.query(`DECLARE snapshot_cursor CURSOR FOR SELECT * FROM ${this.table.qualifiedName}`);
+    let query = `SELECT * FROM ${this.table.qualifiedName}`;
+    if (this.table.initialSnapshotFilter?.sql) {
+      query += ` WHERE (${this.table.initialSnapshotFilter.sql})`;
+    }
+    await this.connection.query(`DECLARE snapshot_cursor CURSOR FOR ${query}`);
   }
 
   public nextChunk(): AsyncIterableIterator<PgChunk> {
@@ -119,17 +123,27 @@ export class ChunkedSnapshotQuery implements SnapshotQuery {
   public async *nextChunk(): AsyncIterableIterator<PgChunk> {
     let stream: AsyncIterableIterator<PgChunk>;
     const escapedKeyName = escapeIdentifier(this.key.name);
+    const snapshotFilter = this.table.initialSnapshotFilter?.sql;
+    
     if (this.lastKey == null) {
-      stream = this.connection.stream(
-        `SELECT * FROM ${this.table.qualifiedName} ORDER BY ${escapedKeyName} LIMIT ${this.chunkSize}`
-      );
+      let query = `SELECT * FROM ${this.table.qualifiedName}`;
+      if (snapshotFilter) {
+        query += ` WHERE (${snapshotFilter})`;
+      }
+      query += ` ORDER BY ${escapedKeyName} LIMIT ${this.chunkSize}`;
+      stream = this.connection.stream(query);
     } else {
       if (this.key.typeId == null) {
         throw new Error(`typeId required for primary key ${this.key.name}`);
       }
       const type = Number(this.key.typeId);
+      let query = `SELECT * FROM ${this.table.qualifiedName} WHERE ${escapedKeyName} > $1`;
+      if (snapshotFilter) {
+        query += ` AND (${snapshotFilter})`;
+      }
+      query += ` ORDER BY ${escapedKeyName} LIMIT ${this.chunkSize}`;
       stream = this.connection.stream({
-        statement: `SELECT * FROM ${this.table.qualifiedName} WHERE ${escapedKeyName} > $1 ORDER BY ${escapedKeyName} LIMIT ${this.chunkSize}`,
+        statement: query,
         params: [{ value: this.lastKey, type }]
       });
     }
@@ -200,8 +214,14 @@ export class IdSnapshotQuery implements SnapshotQuery {
     if (type == null) {
       throw new Error(`Cannot determine primary key array type for ${JSON.stringify(keyDefinition)}`);
     }
+    
+    let query = `SELECT * FROM ${this.table.qualifiedName} WHERE ${escapeIdentifier(keyDefinition.name)} = ANY($1)`;
+    if (this.table.initialSnapshotFilter?.sql) {
+      query += ` AND (${this.table.initialSnapshotFilter.sql})`;
+    }
+    
     yield* this.connection.stream({
-      statement: `SELECT * FROM ${this.table.qualifiedName} WHERE ${escapeIdentifier(keyDefinition.name)} = ANY($1)`,
+      statement: query,
       params: [
         {
           type: type,

modules/module-postgres/src/replication/WalStream.ts

@@ -558,6 +558,11 @@ WHERE  oid = $1::regclass`,
       q = new SimpleSnapshotQuery(db, table, this.snapshotChunkLength);
       at = 0;
     }
+    
+    if (table.initialSnapshotFilter?.sql) {
+      this.logger.info(`Applying initial snapshot filter: ${table.initialSnapshotFilter.sql}`);
+    }
+    
     await q.initialize();
 
     let hasRemainingData = true;

packages/sync-rules/src/SyncConfig.ts

@@ -144,41 +144,44 @@ export abstract class SyncConfig {
   /**
    * Get the initial snapshot filter for a given table.
    * Filters are applied globally and support wildcard matching.
-   * 
-   * @param connectionTag Connection tag (e.g., 'default')
+   *
+   * @param connectionTag Connection tag for the active source connection
    * @param schema Schema name
-   * @param tableName Table name (without wildcard %)
-   * @param dbType Database type ('sql' for MySQL/Postgres/MSSQL, 'mongo' for MongoDB)
-   * @returns WHERE clause/query, or undefined if no filter is specified
+   * @param tableName Concrete table name (wildcards are allowed in the filter patterns, not here)
+   * @param dbType Database type ('sql' for MySQL/Postgres/MSSQL, 'mongo' for MongoDB) - currently unused, returns full filter object
+   * @returns Filter object with sql/mongo properties, or undefined if no filter is specified
    */
-  getInitialSnapshotFilter(connectionTag: string, schema: string, tableName: string, dbType: DatabaseType = 'sql'): string | any | undefined {
-    const fullTableName = `${schema}.${tableName}`;
-    
-    // Check for exact matches first
+  getInitialSnapshotFilter(
+    connectionTag: string,
+    schema: string,
+    tableName: string,
+    dbType: DatabaseType = 'sql'
+  ): InitialSnapshotFilter | undefined {
     for (const [pattern, filterDef] of this.initialSnapshotFilters) {
       const tablePattern = this.parseTablePattern(connectionTag, schema, pattern);
       if (tablePattern.matches({ connectionTag, schema, name: tableName })) {
-        // Return the appropriate filter based on database type
-        return filterDef[dbType];
+        return filterDef;
       }
     }
-    
+
     return undefined;
   }
 
   /**
    * Helper to parse a table pattern string into a TablePattern object
    */
   private parseTablePattern(connectionTag: string, defaultSchema: string, pattern: string): TablePattern {
-    // Split on '.' to extract schema and table parts
     const parts = pattern.split('.');
     if (parts.length === 1) {
-      // Just table name, use default schema
-      return new TablePattern(defaultSchema, parts[0]);
-    } else {
-      // schema.table format
-      return new TablePattern(parts[0], parts[1]);
+      return new TablePattern(`${connectionTag}.${defaultSchema}`, parts[0]);
+    }
+    if (parts.length === 2) {
+      return new TablePattern(`${connectionTag}.${parts[0]}`, parts[1]);
     }
+    const tag = parts[0];
+    const schema = parts[1];
+    const tableName = parts.slice(2).join('.');
+    return new TablePattern(`${tag}.${schema}`, tableName);
   }
 }
 export interface SyncConfigWithErrors {