diff --git a/.changeset/famous-cobras-give.md b/.changeset/famous-cobras-give.md
new file mode 100644
index 000000000..9a2d12069
--- /dev/null
+++ b/.changeset/famous-cobras-give.md
@@ -0,0 +1,39 @@
+---
+'@powersync/service-sync-rules': minor
+'@powersync/service-core': minor
+'@powersync/service-module-postgres-storage': patch
+'@powersync/service-module-mysql': minor
+---
+
+Add snapshot_filter support for initial table replication
+
+Users can now configure snapshot filters in sync_rules.yaml to apply WHERE clauses during initial table snapshots. This reduces storage and bandwidth for large tables where only a subset of rows match sync rules.
+
+Features:
+
+- Configure global filters for initial replication
+- CDC changes continue to work normally, only affecting rows in storage
+- Supports MySQL source with PostgreSQL storage
+
+Example:
+
+```yaml
+# EXPLICIT: "I only want data from the last 90 days, period"
+initial_snapshot_filters:
+  orders:
+    sql: 'created_at > DATE_SUB(NOW(), INTERVAL 90 DAY)'
+
+  logs:
+    sql: "timestamp > NOW() - INTERVAL '7 days'"
+
+bucket_definitions:
+  # This works - queries recent orders
+  recent_orders:
+    data:
+      - SELECT * FROM orders WHERE created_at > DATE_SUB(NOW(), INTERVAL 30 DAY)
+
+  # This will be EMPTY initially - and that's OK for logs
+  all_logs:
+    data:
+      - SELECT * FROM logs
+```
diff --git a/docs/initial-snapshot-filters.md b/docs/initial-snapshot-filters.md
new file mode 100644
index 000000000..dce9802dc
--- /dev/null
+++ b/docs/initial-snapshot-filters.md
@@ -0,0 +1,428 @@
+# Initial Snapshot Filters
+
+Initial snapshot filters allow you to limit which rows are replicated during the initial snapshot phase. This is useful for large tables where you only need to sync a subset of data.
+
+## Overview
+
+Without snapshot filters, PowerSync replicates **all rows** from source tables during initial replication, even if your sync rules only match a small subset. Initial snapshot filters solve this by allowing you to specify which rows to replicate upfront.
+
+**Important**: Filters are configured **globally** at the top level of your sync rules and apply to **all buckets** using that table. This means some buckets may end up empty if their queries don't align with the global filter.
+
+## Syntax
+
+Initial snapshot filters are defined at the **top level** of your `sync_rules.yaml` using an **object format** with database-specific filter syntax:
+
+```yaml
+initial_snapshot_filters:
+  users:
+    sql: "status = 'active'" # MySQL, PostgreSQL, SQL Server
+    mongo: { status: 'active' } # MongoDB (BSON/EJSON format)
+
+bucket_definitions:
+  active_users:
+    data:
+      - SELECT id, name, status FROM users
+```
+
+You can specify just `sql` or just `mongo` depending on your database:
+
+```yaml
+initial_snapshot_filters:
+  todos:
+    sql: 'archived = false'
+
+bucket_definitions:
+  my_todos:
+    data:
+      - SELECT id, title FROM todos
+```
+
+## Basic Examples
+
+### SQL Databases (MySQL, PostgreSQL, SQL Server)
+
+```yaml
+initial_snapshot_filters:
+  users:
+    sql: "status = 'active'"
+
+bucket_definitions:
+  active_users:
+    data:
+      - SELECT id, name, status FROM users
+```
+
+### MongoDB
+
+```yaml
+initial_snapshot_filters:
+  users:
+    mongo: { status: 'active' }
+
+bucket_definitions:
+  active_users:
+    data:
+      - SELECT id, name, status FROM users
+```
+
+### Multi-Database Support
+
+Specify filters for both SQL and MongoDB sources:
+
+```yaml
+initial_snapshot_filters:
+  users:
+    sql: "status = 'active'"
+    mongo: { status: 'active' }
+
+bucket_definitions:
+  active_users:
+    data:
+      - SELECT id, name, status FROM users
+```
+
+## Global Filter Behavior
+
+⚠️ **Critical**: Filters are **global** and apply to **all buckets** using that table. This can result in empty buckets if the bucket query doesn't match the filter.
+
+### Example: Misaligned Bucket Query
+
+```yaml
+initial_snapshot_filters:
+  users:
+    sql: "status = 'active'" # Only active users are replicated
+
+bucket_definitions:
+  active_users:
+    data:
+      - SELECT * FROM users WHERE status = 'active' # ✅ Will have data
+
+  admin_users:
+    data:
+      - SELECT * FROM users WHERE is_admin = true # ⚠️ Will be EMPTY if admin users aren't active
+```
+
+In this example:
+
+- Only users with `status = 'active'` are replicated (due to the global filter)
+- The `admin_users` bucket will be **empty** because no admin users are included in the initial snapshot
+- This is a **deliberate trade-off** between snapshot performance and bucket completeness
+
+### When to Use Global Filters
+
+✅ **Good use case**: All your buckets can work with a single filter
+
+```yaml
+initial_snapshot_filters:
+  orders:
+    sql: 'created_at > DATE_SUB(NOW(), INTERVAL 90 DAY)' # Only recent orders
+
+bucket_definitions:
+  my_orders:
+    data:
+      - SELECT * FROM orders WHERE user_id = token_parameters.user_id
+  pending_orders:
+    data:
+      - SELECT * FROM orders WHERE status = 'pending'
+```
+
+Both buckets work with recent orders only - the filter just improves snapshot performance.
+
+❌ **Bad use case**: Different buckets need different data
+
+```yaml
+initial_snapshot_filters:
+  users:
+    sql: "status = 'active'" # Only active users
+
+bucket_definitions:
+  active_users:
+    data:
+      - SELECT * FROM users WHERE status = 'active' # ✅ Works
+  archived_users:
+    data:
+      - SELECT * FROM users WHERE status = 'archived' # ⚠️ Will be EMPTY!
+```
+
+The `archived_users` bucket will never have data because archived users aren't included in the snapshot.
+
+## Complex Filters
+
+You can use any valid SQL WHERE clause syntax:
+
+```yaml
+initial_snapshot_filters:
+  orders:
+    sql: "created_at > DATE_SUB(NOW(), INTERVAL 30 DAY) AND status != 'cancelled'"
+
+bucket_definitions:
+  my_orders:
+    data:
+      - SELECT * FROM orders WHERE user_id = token_parameters.user_id
+```
+
+## Wildcard Table Names
+
+You can use wildcards (%) in table names to match multiple tables:
+
+```yaml
+initial_snapshot_filters:
+  logs_%:
+    sql: 'created_at > DATE_SUB(NOW(), INTERVAL 7 DAY)'
+
+bucket_definitions:
+  recent_logs:
+    data:
+      - SELECT * FROM logs_2024
+      - SELECT * FROM logs_2025
+```
+
+This applies the filter to `logs_2024`, `logs_2025`, etc.
+
+## Schema-Qualified Names
+
+You can specify schema-qualified table names:
+
+```yaml
+initial_snapshot_filters:
+  public.users:
+    sql: "status = 'active'"
+  analytics.events:
+    sql: "timestamp > NOW() - INTERVAL '1 day'"
+
+bucket_definitions:
+  active_users:
+    data:
+      - SELECT id, name FROM public.users
+  recent_events:
+    data:
+      - SELECT * FROM analytics.events
+```
+
+## How It Works
+
+1. **Initial Snapshot**: Only rows matching the filter are replicated during the initial snapshot phase
+2. **CDC Replication**: After the snapshot, Change Data Capture (CDC) continues to replicate changes to **only the rows that were included in the snapshot**
+3. **Storage**: Only filtered rows are stored in PostgreSQL/MongoDB storage
+4. **Buckets**: All buckets using the filtered table will only see the filtered data
+
+### Changing Filters Later
+
+If you widen a filter after the initial snapshot, CDC will **not** backfill older rows that were previously excluded because those rows did not change. To include them, run a **targeted resnapshot** of the affected tables or perform a **one-time backfill** job, then let CDC keep the data up to date.
+
+## Database-Specific Syntax
+
+### MySQL
+
+```yaml
+initial_snapshot_filters:
+  orders:
+    sql: 'DATE(created_at) > DATE_SUB(CURDATE(), INTERVAL 30 DAY)'
+```
+
+### PostgreSQL
+
+```yaml
+initial_snapshot_filters:
+  orders:
+    sql: "created_at > NOW() - INTERVAL '30 days'"
+```
+
+### SQL Server
+
+```yaml
+initial_snapshot_filters:
+  orders:
+    sql: 'created_at > DATEADD(day, -30, GETDATE())'
+```
+
+### MongoDB
+
+```yaml
+initial_snapshot_filters:
+  orders:
+    mongo:
+      created_at: { $gt: { $date: '2024-01-01T00:00:00Z' } }
+      status: { $in: ['active', 'pending'] }
+```
+
+## Use Cases
+
+### Large Historical Tables
+
+```yaml
+# Only sync recent orders instead of entire order history
+initial_snapshot_filters:
+  orders:
+    sql: 'created_at > DATE_SUB(NOW(), INTERVAL 90 DAY)'
+
+bucket_definitions:
+  my_orders:
+    data:
+      - SELECT * FROM orders WHERE user_id = token_parameters.user_id
+```
+
+### Active Records Only
+
+```yaml
+# Skip archived/deleted records
+initial_snapshot_filters:
+  users:
+    sql: 'archived = false AND deleted_at IS NULL'
+
+bucket_definitions:
+  all_users:
+    data:
+      - SELECT id, name FROM users
+```
+
+### Tenant/Organization Filtering
+
+```yaml
+# Multi-tenant app - only sync specific tenants
+initial_snapshot_filters:
+  records:
+    sql: "tenant_id IN ('tenant-a', 'tenant-b')"
+
+bucket_definitions:
+  my_records:
+    data:
+      - SELECT * FROM records WHERE user_id = token_parameters.user_id
+```
+
+### Partitioned Tables
+
+```yaml
+# Apply same filter to all partitions
+initial_snapshot_filters:
+  events_%:
+    sql: 'created_at > DATE_SUB(NOW(), INTERVAL 30 DAY)'
+
+bucket_definitions:
+  recent_events:
+    data:
+      - SELECT * FROM events_2024
+      - 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
+- CDC changes only affect rows that were **initially snapshotted**
+- Widening filters requires a **resnapshot or backfill** to pick up previously excluded rows
+- Filter syntax must be valid for your **source database**
+- Some buckets may be **empty** if their queries don't align with the global filter
+
+## Best Practices
+
+1. **Design filters carefully** - ensure all buckets can work with the same filter
+2. **Test filters** on your source database before deploying
+3. **Use indexes** on filtered columns for better snapshot performance
+4. **Be conservative** - if unsure whether a row is needed, include it
+5. **Document filters** in your sync_rules.yaml comments
+6. **Monitor snapshot progress** to ensure reasonable replication times
+7. **Verify bucket data** after applying filters to ensure no buckets are unexpectedly empty
+
+## Migration from Per-Bucket Filters
+
+If you're using the old per-bucket `source_tables` configuration, move filters to the top level:
+
+```yaml
+# Old (no longer supported)
+bucket_definitions:
+  active_users:
+    data:
+      - SELECT id, name FROM users
+    source_tables:
+      users:
+        initial_snapshot_filter:
+          sql: "status = 'active'"
+
+# New (global configuration)
+initial_snapshot_filters:
+  users:
+    sql: "status = 'active'"
+
+bucket_definitions:
+  active_users:
+    data:
+      - SELECT id, name FROM users
+```
+
+**Important**: With the old configuration, if multiple buckets specified different filters for the same table, they were ORed together. With the new configuration, there is **one global filter per table** that applies to all buckets.
diff --git a/modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts b/modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts
index 59021f9c0..e0c6dec3b 100644
--- a/modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts
+++ b/modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts
@@ -300,7 +300,8 @@ export abstract 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)
       });
       sourceTable.syncEvent = options.sync_rules.tableTriggersEvent(sourceTable);
       sourceTable.syncData = options.sync_rules.tableSyncsData(sourceTable);
diff --git a/modules/module-mongodb/src/replication/ChangeStream.ts b/modules/module-mongodb/src/replication/ChangeStream.ts
index c92954908..692247ca1 100644
--- a/modules/module-mongodb/src/replication/ChangeStream.ts
+++ b/modules/module-mongodb/src/replication/ChangeStream.ts
@@ -482,7 +482,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(
@@ -491,6 +492,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();
diff --git a/modules/module-mongodb/src/replication/MongoSnapshotQuery.ts b/modules/module-mongodb/src/replication/MongoSnapshotQuery.ts
index b92493fae..b6a857c29 100644
--- a/modules/module-mongodb/src/replication/MongoSnapshotQuery.ts
+++ b/modules/module-mongodb/src/replication/MongoSnapshotQuery.ts
@@ -15,12 +15,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<
@@ -39,8 +46,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,
diff --git a/modules/module-mssql/src/replication/CDCStream.ts b/modules/module-mssql/src/replication/CDCStream.ts
index a6b39f3d5..1d524e1ee 100644
--- a/modules/module-mssql/src/replication/CDCStream.ts
+++ b/modules/module-mssql/src/replication/CDCStream.ts
@@ -348,6 +348,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;
diff --git a/modules/module-mssql/src/replication/MSSQLSnapshotQuery.ts b/modules/module-mssql/src/replication/MSSQLSnapshotQuery.ts
index 96d7f3db9..3222aa8a7 100644
--- a/modules/module-mssql/src/replication/MSSQLSnapshotQuery.ts
+++ b/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
diff --git a/modules/module-mysql/src/replication/BinLogStream.ts b/modules/module-mysql/src/replication/BinLogStream.ts
index b2865f0fe..b707d8b47 100644
--- a/modules/module-mysql/src/replication/BinLogStream.ts
+++ b/modules/module-mysql/src/replication/BinLogStream.ts
@@ -305,12 +305,25 @@ 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
-    const query = connection.query(`SELECT /*+ MAX_EXECUTION_TIME(0) */ * FROM ${qualifiedMySQLTable(table)}`);
-    const stream = query.stream();
+    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})`;
+      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);
+    const stream = queryStream.stream();
 
     let columns: Map<string, ColumnDescriptor> | undefined = undefined;
     stream.on('fields', (fields: mysql.FieldPacket[]) => {
diff --git a/modules/module-mysql/test/src/BinLogStream.test.ts b/modules/module-mysql/test/src/BinLogStream.test.ts
index 245816412..3610431b0 100644
--- a/modules/module-mysql/test/src/BinLogStream.test.ts
+++ b/modules/module-mysql/test/src/BinLogStream.test.ts
@@ -403,4 +403,199 @@ function defineBinlogStreamTests(config: storage.TestStorageConfig) {
       ]);
     }
   });
+
+  // test('Snapshot filter - replicate only filtered rows', async () => {
+  //   await using context = await BinlogStreamTestContext.open(factory);
+  //   const { connectionManager } = context;
+  //   await context.updateSyncRules(`
+  // initial_snapshot_filters:
+  //   users:
+  //     sql: "status = 'active'"
+
+  // bucket_definitions:
+  //   active_users:
+  //     data:
+  //       - SELECT id, name, status FROM "users"`);
+
+  //   await connectionManager.query(`CREATE TABLE users (id CHAR(36) PRIMARY KEY, name TEXT, status VARCHAR(20))`);
+
+  //   // Insert rows before snapshot
+  //   const activeId = uuid();
+  //   const inactiveId = uuid();
+  //   await connectionManager.query(`INSERT INTO users(id, name, status) VALUES('${activeId}', 'Active User', 'active')`);
+  //   await connectionManager.query(
+  //     `INSERT INTO users(id, name, status) VALUES('${inactiveId}', 'Inactive User', 'inactive')`
+  //   );
+
+  //   await context.replicateSnapshot();
+
+  //   const data = await context.getBucketData('active_users[]');
+
+  //   // Should only have the active user, not the inactive one
+  //   expect(data).toMatchObject([putOp('users', { id: activeId, name: 'Active User', status: 'active' })]);
+  //   expect(data.length).toBe(1);
+  // });
+
+  // test('Snapshot filter - only specified filter applied globally', async () => {
+  //   await using context = await BinlogStreamTestContext.open(factory);
+  //   const { connectionManager } = context;
+  //   await context.updateSyncRules(`
+  // initial_snapshot_filters:
+  //   users:
+  //     sql: "status = 'active'"
+
+  // bucket_definitions:
+  //   active_users:
+  //     data:
+  //       - SELECT id, name, status FROM "users" WHERE status = 'active'
+
+  //   admin_users:
+  //     data:
+  //       - SELECT id, name, is_admin FROM "users" WHERE is_admin = true`);
+
+  //   await connectionManager.query(
+  //     `CREATE TABLE users (id CHAR(36) PRIMARY KEY, name TEXT, status VARCHAR(20), is_admin BOOLEAN)`
+  //   );
+
+  //   // Insert test data
+  //   const activeUserId = uuid();
+  //   const adminUserId = uuid();
+  //   const regularUserId = uuid();
+
+  //   await connectionManager.query(
+  //     `INSERT INTO users(id, name, status, is_admin) VALUES('${activeUserId}', 'Active User', 'active', false)`
+  //   );
+  //   await connectionManager.query(
+  //     `INSERT INTO users(id, name, status, is_admin) VALUES('${adminUserId}', 'Admin User', 'inactive', true)`
+  //   );
+  //   await connectionManager.query(
+  //     `INSERT INTO users(id, name, status, is_admin) VALUES('${regularUserId}', 'Regular User', 'inactive', false)`
+  //   );
+
+  //   await context.replicateSnapshot();
+
+  //   const activeData = await context.getBucketData('active_users[]');
+  //   const adminData = await context.getBucketData('admin_users[]');
+
+  //   // Active bucket should have the active user
+  //   expect(activeData).toMatchObject([
+  //     putOp('users', { id: activeUserId, name: 'Active User', status: 'active', is_admin: 0n })
+  //   ]);
+
+  //   // Admin bucket is empty because global filter only allows active users, not admin users
+  //   expect(adminData).toMatchObject([]);
+
+  //   // Regular user should not be in either bucket (filtered out by snapshot filter)
+  // });
+
+  // test('Snapshot filter - CDC changes only affect filtered rows', async () => {
+  //   await using context = await BinlogStreamTestContext.open(factory);
+  //   const { connectionManager } = context;
+  //   await context.updateSyncRules(`
+  // initial_snapshot_filters:
+  //   users:
+  //     sql: "status = 'active'"
+
+  // bucket_definitions:
+  //   active_users:
+  //     data:
+  //       - SELECT id, name, status FROM "users" WHERE status = 'active'`);
+
+  //   await connectionManager.query(`CREATE TABLE users (id CHAR(36) PRIMARY KEY, name TEXT, status VARCHAR(20))`);
+
+  //   // Insert an active user before snapshot
+  //   const activeId = uuid();
+  //   await connectionManager.query(`INSERT INTO users(id, name, status) VALUES('${activeId}', 'Active User', 'active')`);
+
+  //   await context.replicateSnapshot();
+  //   await context.startStreaming();
+
+  //   // Insert an inactive user - should not appear in bucket
+  //   const inactiveId = uuid();
+  //   await connectionManager.query(
+  //     `INSERT INTO users(id, name, status) VALUES('${inactiveId}', 'Inactive User', 'inactive')`
+  //   );
+
+  //   // Update the active user - should appear in bucket
+  //   await connectionManager.query(`UPDATE users SET name = 'Updated Active' WHERE id = '${activeId}'`);
+
+  //   const data = await context.getBucketData('active_users[]');
+
+  //   // Should only have the active user with updated name
+  //   expect(data).toMatchObject([putOp('users', { id: activeId, name: 'Updated Active', status: 'active' })]);
+  //   expect(data.length).toBe(1);
+  // });
+
+  // test('Snapshot filter - complex WHERE clause', async () => {
+  //   await using context = await BinlogStreamTestContext.open(factory);
+  //   const { connectionManager } = context;
+  //   await context.updateSyncRules(`
+  // initial_snapshot_filters:
+  //   users:
+  //     sql: "created_at > DATE_SUB(NOW(), INTERVAL 7 DAY) AND status = 'active'"
+
+  // bucket_definitions:
+  //   recent_active_users:
+  //     data:
+  //       - SELECT id, name, created_at FROM "users"`);
+
+  //   await connectionManager.query(
+  //     `CREATE TABLE users (id CHAR(36) PRIMARY KEY, name TEXT, status VARCHAR(20), created_at DATETIME)`
+  //   );
+
+  //   // Insert recent active user
+  //   const recentActiveId = uuid();
+  //   await connectionManager.query(
+  //     `INSERT INTO users(id, name, status, created_at) VALUES('${recentActiveId}', 'Recent Active', 'active', NOW())`
+  //   );
+
+  //   // Insert old active user
+  //   const oldActiveId = uuid();
+  //   await connectionManager.query(
+  //     `INSERT INTO users(id, name, status, created_at) VALUES('${oldActiveId}', 'Old Active', 'active', DATE_SUB(NOW(), INTERVAL 30 DAY))`
+  //   );
+
+  //   // Insert recent inactive user
+  //   const recentInactiveId = uuid();
+  //   await connectionManager.query(
+  //     `INSERT INTO users(id, name, status, created_at) VALUES('${recentInactiveId}', 'Recent Inactive', 'inactive', NOW())`
+  //   );
+
+  //   await context.replicateSnapshot();
+
+  //   const data = await context.getBucketData('recent_active_users[]');
+
+  //   // Should only have the recent active user
+  //   expect(data.length).toBe(1);
+  //   expect(data[0]).toMatchObject({
+  //     op: 'PUT',
+  //     object_type: 'users',
+  //     object_id: recentActiveId
+  //   });
+  // });
+
+  // test('Snapshot filter - no filter means all rows replicated', async () => {
+  //   await using context = await BinlogStreamTestContext.open(factory);
+  //   const { connectionManager } = context;
+  //   await context.updateSyncRules(`
+  // bucket_definitions:
+  //   all_users:
+  //     data:
+  //       - SELECT id, name FROM "users"`);
+
+  //   await connectionManager.query(`CREATE TABLE users (id CHAR(36) PRIMARY KEY, name TEXT)`);
+
+  //   // Insert multiple users
+  //   const user1Id = uuid();
+  //   const user2Id = uuid();
+  //   await connectionManager.query(`INSERT INTO users(id, name) VALUES('${user1Id}', 'User 1')`);
+  //   await connectionManager.query(`INSERT INTO users(id, name) VALUES('${user2Id}', 'User 2')`);
+
+  //   await context.replicateSnapshot();
+
+  //   const data = await context.getBucketData('all_users[]');
+
+  //   // Should have both users when no filter is specified
+  //   expect(data.length).toBe(2);
+  // });
 }
diff --git a/modules/module-postgres-storage/src/storage/PostgresSyncRulesStorage.ts b/modules/module-postgres-storage/src/storage/PostgresSyncRulesStorage.ts
index 807aeee55..5144b613e 100644
--- a/modules/module-postgres-storage/src/storage/PostgresSyncRulesStorage.ts
+++ b/modules/module-postgres-storage/src/storage/PostgresSyncRulesStorage.ts
@@ -267,7 +267,8 @@ export class PostgresSyncRulesStorage
         schema: schema,
         name: table,
         replicaIdColumns: replicaIdColumns,
-        snapshotComplete: sourceTableRow!.snapshot_done ?? true
+        snapshotComplete: sourceTableRow!.snapshot_done ?? true,
+        initialSnapshotFilter: options.sync_rules.definition.getInitialSnapshotFilter(connection_tag, schema, table)
       });
       if (!sourceTable.snapshotComplete) {
         sourceTable.snapshotStatus = {
diff --git a/modules/module-postgres/src/replication/SnapshotQuery.ts b/modules/module-postgres/src/replication/SnapshotQuery.ts
index 799b484ba..171813fe5 100644
--- a/modules/module-postgres/src/replication/SnapshotQuery.ts
+++ b/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,
diff --git a/modules/module-postgres/src/replication/WalStream.ts b/modules/module-postgres/src/replication/WalStream.ts
index 6d2a54b54..4b3c62360 100644
--- a/modules/module-postgres/src/replication/WalStream.ts
+++ b/modules/module-postgres/src/replication/WalStream.ts
@@ -717,6 +717,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;
diff --git a/packages/service-core/src/storage/SourceTable.ts b/packages/service-core/src/storage/SourceTable.ts
index c9559b2d7..720e391a8 100644
--- a/packages/service-core/src/storage/SourceTable.ts
+++ b/packages/service-core/src/storage/SourceTable.ts
@@ -8,6 +8,17 @@ import { ColumnDescriptor, SourceEntityDescriptor } from './SourceEntity.js';
  */
 export type SourceTableId = string | bson.ObjectId;
 
+/**
+ * Filter definition for initial snapshot replication.
+ * Object with database-specific filters:
+ *   - sql: MySQL, PostgreSQL, SQL Server syntax
+ *   - mongo: MongoDB query syntax (BSON/EJSON format)
+ */
+export type InitialSnapshotFilter = {
+  sql?: string;
+  mongo?: any;
+};
+
 export interface SourceTableOptions {
   id: SourceTableId;
   connectionTag: string;
@@ -16,6 +27,7 @@ export interface SourceTableOptions {
   name: string;
   replicaIdColumns: ColumnDescriptor[];
   snapshotComplete: boolean;
+  initialSnapshotFilter?: InitialSnapshotFilter;
 }
 
 export interface TableSnapshotStatus {
@@ -90,6 +102,10 @@ export class SourceTable implements SourceEntityDescriptor {
     return this.options.replicaIdColumns;
   }
 
+  get initialSnapshotFilter() {
+    return this.options.initialSnapshotFilter;
+  }
+
   /**
    *  Sanitized name of the entity in the format of "{schema}.{entity name}"
    *  Suitable for safe use in Postgres queries.
@@ -113,7 +129,8 @@ export class SourceTable implements SourceEntityDescriptor {
       schema: this.schema,
       name: this.name,
       replicaIdColumns: this.replicaIdColumns,
-      snapshotComplete: this.snapshotComplete
+      snapshotComplete: this.snapshotComplete,
+      initialSnapshotFilter: this.initialSnapshotFilter
     });
     copy.syncData = this.syncData;
     copy.syncParameters = this.syncParameters;
diff --git a/packages/sync-rules/src/SyncConfig.ts b/packages/sync-rules/src/SyncConfig.ts
index 637ca747b..384944cbc 100644
--- a/packages/sync-rules/src/SyncConfig.ts
+++ b/packages/sync-rules/src/SyncConfig.ts
@@ -8,6 +8,17 @@ import { TablePattern } from './TablePattern.js';
 import { SqliteInputValue, SqliteRow, SqliteValue } from './types.js';
 import { applyRowContext } from './utils.js';
 
+/**
+ * Filter definition for initial snapshot replication.
+ * Object with database-specific filters.
+ */
+export type InitialSnapshotFilter = {
+  sql?: string;
+  mongo?: any;
+};
+
+export type DatabaseType = 'sql' | 'mongo';
+
 /**
  * A class describing how the sync process has been configured (i.e. which buckets and parameters to create and how to
  * resolve buckets for connections).
@@ -25,6 +36,13 @@ export abstract class SyncConfig {
   storageVersion: number | undefined;
   eventDescriptors: SqlEventDescriptor[] = [];
 
+  /**
+   * Global initial snapshot filters for source tables.
+   * Map structure: tableName -> initialSnapshotFilter
+   * Filters are applied globally during initial snapshot, regardless of bucket definitions.
+   */
+  initialSnapshotFilters: Map<string, InitialSnapshotFilter> = new Map();
+
   /**
    * The (YAML-based) source contents from which these sync rules have been derived.
    */
@@ -124,6 +142,56 @@ export abstract class SyncConfig {
   debugRepresentation() {
     return this.bucketSources.map((rules) => rules.debugRepresentation());
   }
+
+  /**
+   * Get the initial snapshot filter for a given table.
+   * Filters are applied globally and support wildcard matching.
+   *
+   * When called without `dbType`, returns the full filter object.
+   * When called with `dbType`, returns only the filter value for that database type.
+   *
+   * @param connectionTag Connection tag for the active source connection
+   * @param schema Schema name
+   * @param tableName Concrete table name (wildcards are allowed in the filter patterns, not here)
+   * @param dbType Optional database type ('sql' or 'mongo'). When provided, extracts the specific filter value.
+   */
+  getInitialSnapshotFilter(connectionTag: string, schema: string, tableName: string): InitialSnapshotFilter | undefined;
+  getInitialSnapshotFilter(connectionTag: string, schema: string, tableName: string, dbType: DatabaseType): any;
+  getInitialSnapshotFilter(
+    connectionTag: string,
+    schema: string,
+    tableName: string,
+    dbType?: DatabaseType
+  ): InitialSnapshotFilter | any | undefined {
+    for (const [pattern, filterDef] of this.initialSnapshotFilters) {
+      const tablePattern = this.parseTablePattern(connectionTag, schema, pattern);
+      if (tablePattern.matches({ connectionTag, schema, name: tableName })) {
+        if (dbType !== undefined) {
+          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 {
+    const parts = pattern.split('.');
+    if (parts.length === 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 {
   config: SyncConfig;
diff --git a/packages/sync-rules/src/from_yaml.ts b/packages/sync-rules/src/from_yaml.ts
index ec5b90798..a32bcd305 100644
--- a/packages/sync-rules/src/from_yaml.ts
+++ b/packages/sync-rules/src/from_yaml.ts
@@ -12,6 +12,9 @@ import { SqlRuleError, SyncRulesErrors, YamlError } from './errors.js';
 import { SqlEventDescriptor } from './events/SqlEventDescriptor.js';
 import { validateSyncRulesSchema } from './json_schema.js';
 import { QueryParseResult, SqlBucketDescriptor } from './SqlBucketDescriptor.js';
+import { QueryParseOptions, SourceSchema, StreamParseOptions } from './types.js';
+import { InitialSnapshotFilter, SyncConfig, SyncConfigWithErrors } from './SyncConfig.js';
+import { ParsingErrorListener, SyncStreamsCompiler } from './compiler/compiler.js';
 import { SqlSyncRules } from './SqlSyncRules.js';
 import { validateStorageVersion } from './StorageVersion.js';
 import { syncStreamFromSql } from './streams/from_sql.js';
@@ -24,6 +27,42 @@ import { buildParsedToSourceValueMap, isBlockScalar, isQuotedScalar } from './ya
 
 const ACCEPT_POTENTIALLY_DANGEROUS_QUERIES = Symbol('ACCEPT_POTENTIALLY_DANGEROUS_QUERIES');
 
+/**
+ * Parse initial_snapshot_filter from YAML object with sql and/or mongo properties
+ */
+function parseInitialSnapshotFilter(value: unknown): InitialSnapshotFilter | undefined {
+  if (value instanceof YAMLMap) {
+    // Object format with db-specific filters
+    const result: { sql?: string; mongo?: any } = {};
+
+    const sqlValue = value.get('sql', true);
+    if (sqlValue) {
+      if (sqlValue instanceof Scalar) {
+        result.sql = sqlValue.value as string;
+      } else {
+        // If it's not a scalar, try to convert it
+        result.sql = String(sqlValue);
+      }
+    }
+
+    const mongoValue = value.get('mongo', true) as any;
+    if (mongoValue instanceof Scalar) {
+      result.mongo = mongoValue.value;
+    } else if (mongoValue instanceof YAMLMap || mongoValue instanceof YAMLSeq) {
+      result.mongo = mongoValue.toJSON();
+    } else if (mongoValue !== null && mongoValue !== undefined) {
+      result.mongo = mongoValue;
+    }
+
+    // Only return if at least one property is set
+    if (result.sql !== undefined || result.mongo !== undefined) {
+      return result;
+    }
+  }
+
+  return undefined;
+}
+
 /**
  * Reads `sync_rules.yaml` files containing a sync configuration.
  *
@@ -101,6 +140,11 @@ export class SyncConfigFromYaml {
       result = this.#legacyParseBucketDefinitionsAndStreams(bucketMap, streamMap, compatibility);
     }
 
+    // Parse global initial_snapshot_filters
+    const filtersMap = parsed.get('initial_snapshot_filters') as YAMLMap | null;
+    if (filtersMap instanceof YAMLMap) {
+      this.#parseInitialSnapshotFilters(filtersMap, result);
+    }
     result.storageVersion = storageVersion;
 
     const eventDefinitions = this.#parseEventDefinitions(parsed, compatibility);
@@ -126,6 +170,45 @@ export class SyncConfigFromYaml {
     }
   }
 
+  /**
+   * Parse the global initial_snapshot_filters configuration
+   */
+  #parseInitialSnapshotFilters(filtersMap: YAMLMap, result: SyncConfig) {
+    for (const entry of filtersMap.items) {
+      const { key: tableKey, value: filterValue } = entry as { key: Scalar; value: unknown };
+      const tableName = tableKey.value as string;
+
+      if (!filterValue) {
+        this.#errors.push(
+          this.#tokenError(tableKey, `Initial snapshot filter for table '${tableName}' cannot be empty`)
+        );
+        continue;
+      }
+
+      if (!(filterValue instanceof YAMLMap)) {
+        this.#errors.push(
+          this.#tokenError(
+            filterValue as any,
+            `Initial snapshot filter for table '${tableName}' must be an object with 'sql' and/or 'mongo' properties`
+          )
+        );
+        continue;
+      }
+
+      const filter = parseInitialSnapshotFilter(filterValue);
+      if (filter) {
+        result.initialSnapshotFilters.set(tableName, filter);
+      } else {
+        this.#errors.push(
+          this.#tokenError(
+            filterValue as any,
+            `Initial snapshot filter for table '${tableName}' must have at least 'sql' or 'mongo' property`
+          )
+        );
+      }
+    }
+  }
+
   /**
    * Parses the `config` block of a sync configuration.
    *
diff --git a/packages/sync-rules/src/json_schema.ts b/packages/sync-rules/src/json_schema.ts
index ade38bdbe..5444aa956 100644
--- a/packages/sync-rules/src/json_schema.ts
+++ b/packages/sync-rules/src/json_schema.ts
@@ -8,6 +8,38 @@ const ajv = new Ajv({ allErrors: true, verbose: true });
 export const syncRulesSchema: ajvModule.Schema = {
   type: 'object',
   properties: {
+    initial_snapshot_filters: {
+      type: 'object',
+      description: 'Global filters applied during initial snapshot replication for specific tables',
+      examples: [
+        {
+          users: {
+            sql: "status = 'active'",
+            mongo: { status: 'active' }
+          },
+          orders: {
+            sql: 'created_at > DATE_SUB(NOW(), INTERVAL 90 DAY)'
+          }
+        }
+      ],
+      patternProperties: {
+        '.*': {
+          type: 'object',
+          description: 'Database-specific filter syntax for a table',
+          properties: {
+            sql: {
+              type: 'string',
+              description: 'SQL WHERE clause for MySQL, PostgreSQL, SQL Server'
+            },
+            mongo: {
+              description: 'MongoDB query filter (BSON/EJSON format) - can be any valid MongoDB query object'
+            }
+          },
+          additionalProperties: false,
+          minProperties: 1
+        }
+      }
+    },
     bucket_definitions: {
       type: 'object',
       description: 'List of bucket definitions',