Adds snapshot filtering for binlog replication

suika · suika · commit 881e732ced79 · 2026-02-11T21:17:20.000Z
Enables filtering of rows during the initial snapshot phase of binlog replication, based on a configurable SQL WHERE clause.

This allows for partial snapshots, replicating only a subset of data based on specified criteria, which is particularly useful for large tables or scenarios where only recent data is needed.

The commit also includes tests to verify the functionality of snapshot filtering, including handling of CDC changes and multiple bucket filters.

Only for source: Mysql and PostgreSQL storage
diff --git a/modules/module-mysql/src/replication/BinLogStream.ts b/modules/module-mysql/src/replication/BinLogStream.ts
@@ -311,8 +311,17 @@ export class BinLogStream {
     // 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 ${qualifiedMySQLTable(table)}`;
+
+    // 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.snapshotFilter) {
+      query += ` WHERE ${table.snapshotFilter}`;
+      this.logger.info(`Applying snapshot filter: ${table.snapshotFilter}`);
+    }
+  
+    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
@@ -401,4 +401,208 @@ function defineBinlogStreamTests(factory: storage.TestStorageFactory) {
       ]);
     }
   });
+
+  test('Snapshot filter - replicate only filtered rows', async () => {
+    await using context = await BinlogStreamTestContext.open(factory);
+    const { connectionManager } = context;
+    await context.updateSyncRules(`
+  bucket_definitions:
+    active_users:
+      data:
+        - SELECT id, name, status FROM "users"
+      source_tables:
+        users:
+          snapshot_filter: "status = 'active'"`);
+
+    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 - ORs multiple bucket filters', async () => {
+    await using context = await BinlogStreamTestContext.open(factory);
+    const { connectionManager } = context;
+    await context.updateSyncRules(`
+  bucket_definitions:
+    active_users:
+      data:
+        - SELECT id, name, status FROM "users" WHERE status = 'active'
+      source_tables:
+        users:
+          snapshot_filter: "status = 'active'"
+    
+    admin_users:
+      data:
+        - SELECT id, name, is_admin FROM "users" WHERE is_admin = true
+      source_tables:
+        users:
+          snapshot_filter: "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 should have the admin user
+    expect(adminData).toMatchObject([
+      putOp('users', { id: adminUserId, name: 'Admin User', status: 'inactive', is_admin: 1n })
+    ]);
+
+    // 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(`
+  bucket_definitions:
+    active_users:
+      data:
+        - SELECT id, name, status FROM "users" WHERE status = 'active'
+      source_tables:
+        users:
+          snapshot_filter: "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(`
+  bucket_definitions:
+    recent_active_users:
+      data:
+        - SELECT id, name, created_at FROM "users"
+      source_tables:
+        users:
+          snapshot_filter: "created_at > DATE_SUB(NOW(), INTERVAL 7 DAY) AND status = 'active'"`);
+
+    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
@@ -249,7 +249,8 @@ export class PostgresSyncRulesStorage
         schema: schema,
         name: table,
         replicaIdColumns: replicaIdColumns,
-        snapshotComplete: sourceTableRow!.snapshot_done ?? true
+        snapshotComplete: sourceTableRow!.snapshot_done ?? true,
+        snapshotFilter: options.sync_rules.definition.getSnapshotFilter(connection_tag, schema, table)
       });
       if (!sourceTable.snapshotComplete) {
         sourceTable.snapshotStatus = {
diff --git a/packages/service-core/src/storage/SourceTable.ts b/packages/service-core/src/storage/SourceTable.ts
@@ -10,6 +10,7 @@ export interface SourceTableOptions {
   name: string;
   replicaIdColumns: ColumnDescriptor[];
   snapshotComplete: boolean;
+  snapshotFilter?: string; // SQL WHERE clause in source DB syntax
 }
 
 export interface TableSnapshotStatus {
@@ -84,6 +85,10 @@ export class SourceTable implements SourceEntityDescriptor {
     return this.options.replicaIdColumns;
   }
 
+  get snapshotFilter() {
+    return this.options.snapshotFilter;
+  }
+
   /**
    *  Sanitized name of the entity in the format of "{schema}.{entity name}"
    *  Suitable for safe use in Postgres queries.
@@ -107,7 +112,8 @@ export class SourceTable implements SourceEntityDescriptor {
       schema: this.schema,
       name: this.name,
       replicaIdColumns: this.replicaIdColumns,
-      snapshotComplete: this.snapshotComplete
+      snapshotComplete: this.snapshotComplete,
+      snapshotFilter: this.snapshotFilter
     });
     copy.syncData = this.syncData;
     copy.syncParameters = this.syncParameters;
diff --git a/packages/sync-rules/src/SyncConfig.ts b/packages/sync-rules/src/SyncConfig.ts
@@ -21,6 +21,13 @@ export abstract class SyncConfig {
   compatibility: CompatibilityContext = CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY;
   eventDescriptors: SqlEventDescriptor[] = [];
 
+  /**
+   * Snapshot filters for source tables, organized by bucket/stream.
+   * Map structure: bucketName -> tableName -> snapshotFilter
+   * Multiple buckets can specify different filters for the same table.
+   */
+  snapshotFilters: Map<string, Map<string, string>> = new Map();
+
   /**
    * The (YAML-based) source contents from which these sync rules have been derived.
    */
@@ -122,6 +129,57 @@ export abstract class SyncConfig {
   debugRepresentation() {
     return this.bucketSources.map((rules) => rules.debugRepresentation());
   }
+
+  /**
+   * Get the aggregated snapshot filter for a given table.
+   * If multiple buckets specify filters for the same table, they are ORed together.
+   * 
+   * @param connectionTag Connection tag (e.g., 'default')
+   * @param schema Schema name
+   * @param tableName Table name (without wildcard %)
+   * @returns Aggregated SQL WHERE clause, or undefined if no filters are specified
+   */
+  getSnapshotFilter(connectionTag: string, schema: string, tableName: string): string | undefined {
+    const filters: string[] = [];
+    const fullTableName = `${schema}.${tableName}`;
+    
+    // Collect all filters for this specific table from all buckets/streams
+    for (const [_bucketName, tableFilters] of this.snapshotFilters) {
+      // Check for filters matching this table (could be exact match or wildcard)
+      for (const [pattern, filter] of tableFilters) {
+        const tablePattern = this.parseTablePattern(connectionTag, schema, pattern);
+        if (tablePattern.matches({ connectionTag, schema, name: tableName })) {
+          filters.push(filter);
+        }
+      }
+    }
+    
+    if (filters.length === 0) {
+      return undefined;
+    }
+    
+    if (filters.length === 1) {
+      return filters[0];
+    }
+    
+    // OR multiple filters together, wrapping each in parentheses
+    return filters.map(f => `(${f})`).join(' OR ');
+  }
+
+  /**
+   * 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]);
+    }
+  }
 }
 export interface SyncConfigWithErrors {
   config: SyncConfig;
diff --git a/packages/sync-rules/src/from_yaml.ts b/packages/sync-rules/src/from_yaml.ts
@@ -266,6 +266,25 @@ export class SyncConfigFromYaml {
       };
       const parameters = value.get('parameters', true) as unknown;
       const dataQueries = value.get('data', true) as unknown;
+      const source_tables = value.get('source_tables', true) as unknown;
+
+      // Parse source_tables configuration for this bucket
+      if (source_tables instanceof YAMLMap) {
+        const tableFilters = new Map<string, string>();
+        for (const entry of source_tables.items) {
+          const { key: tableKey, value: tableValue } = entry as { key: Scalar; value: YAMLMap };
+          const tableName = tableKey.toString();
+          if (tableValue instanceof YAMLMap) {
+            const snapshot_filter = tableValue.get('snapshot_filter', true) as Scalar | null;
+            if (snapshot_filter) {
+              tableFilters.set(tableName, snapshot_filter.toString());
+            }
+          }
+        }
+        if (tableFilters.size > 0) {
+          rules.snapshotFilters.set(key, tableFilters);
+        }
+      }
 
       const descriptor = new SqlBucketDescriptor(key);
 
@@ -333,6 +352,25 @@ export class SyncConfigFromYaml {
         compatibility
       };
 
+      // Parse source_tables configuration for this stream
+      const source_tables = value.get('source_tables', true) as unknown;
+      if (source_tables instanceof YAMLMap) {
+        const tableFilters = new Map<string, string>();
+        for (const entry of source_tables.items) {
+          const { key: tableKey, value: tableValue } = entry as { key: Scalar; value: YAMLMap };
+          const tableName = tableKey.toString();
+          if (tableValue instanceof YAMLMap) {
+            const snapshot_filter = tableValue.get('snapshot_filter', true) as Scalar | null;
+            if (snapshot_filter) {
+              tableFilters.set(tableName, snapshot_filter.toString());
+            }
+          }
+        }
+        if (tableFilters.size > 0) {
+          rules.snapshotFilters.set(key, tableFilters);
+        }
+      }
+
       const data = value.get('query', true) as unknown;
       if (data instanceof Scalar) {
         this.#withScalar(data, (q) => {
diff --git a/packages/sync-rules/src/json_schema.ts b/packages/sync-rules/src/json_schema.ts
diff --git a/packages/sync-rules/test/src/snapshot_filter.test.ts b/packages/sync-rules/test/src/snapshot_filter.test.ts
