What if the filter in sql contains ; or :, like city != 'China:beijing'?

Sorry I’m late, I was busy with something else. I have modified the snapshotFilters parsing logic in MySqlDataSourceFactory to correctly handle semicolons and colons within filter conditions.

Key improvements:

1. parseSnapshotFilters method:

• Intelligently parses semicolon delimiters while recognizing quoted content
• Supports both single and double quotes
• Semicolons within quotes are not treated as delimiters

2. parseTableAndCondition method:

• Splits only on the first colon to avoid misinterpreting colons in condition values
• Ensures proper separation of table names and conditions

I think a better solution to this problem should allow passing nested YAML structures (LIST and MAP should be enough) as config option values instead of inventing more customized syntaxes, as they're hard to write and parse.

I think a better solution to this problem should allow passing nested YAML structures (LIST and MAP should be enough) as config option values instead of inventing more customized syntaxes, as they're hard to write and parse.

I agree with u , but i think that we should consider the difference with snapshot filters or cdc filters

You are right. Actually, at the beginning I tried using nested YAML structures (LIST and MAP). However, I found that the YamlPipelineDefinitionParser.toSourceDef() method doesn’t support that; it only supports STRING values (because it uses Map<String, String> in Java).

I’m not sure whether this was intentional or simply a limitation of the current implementation. If I could use Map<String, Object>, I could switch to YAML structures, which is clearly more reasonable than parsing everything from strings.

snapshotFilter is only forwarded into .snapshotFilters(...) when enableParallelRead is true. If a user sets scan.snapshot.filter while parallel snapshot is disabled, the option will currently be silently ignored. Consider either (1) validating and failing fast when the option is set but parallel read is disabled, or (2) implementing equivalent filtering for the non-parallel Debezium source path, and documenting the supported modes.

fixed 3d360c9

scan.snapshot.filters is parsed via Collectors.toMap(...) without validating required keys (table/filter) and without handling duplicates. This can throw a generic NullPointerException/IllegalStateException on malformed or duplicate entries and also drops the original list order (which matters if patterns overlap). Consider validating each entry and building a LinkedHashMap with explicit duplicate handling (e.g., reject with a clear message).

fixed 67fae6d

getSnapshotFilter picks the first matching selector from a HashMap keySet iteration, which is non-deterministic when multiple patterns match the same table. This can cause different filters to be applied across runs/JVMs. Please make matching deterministic (e.g., preserve user-defined order via LinkedHashMap/list, or define precedence like exact match > regex, or fail fast on multiple matches).

fixed b7f13fd

We need more tests for this feature(IT, E2E test).

Added two tests in commit 85647b2:

• MySqlSourceITCase#testSnapshotFiltersMultipleTables — IT test
  verifying per-table filters work independently across two tables
  via the DataStream API.
• MysqlE2eITCase#testSnapshotFilters — E2E test verifying the
  full pipeline path using the YAML-based scan.snapshot.filters
  configuration.

I am considering whether we should implement SupportsFilterPushDown for the sql connector instead of a new config.

Thanks for the suggestion!
We actually discussed this with @ThorneANN earlier — the conclusion was that SupportsFilterPushDown is not suitable here because of the Incremental Snapshot framework's design.
The current config-based approach (scan.snapshot.filter) is intentional.

It is better to add a new use example for this feature.

Added usage examples in commit cc477d5:

• docs/connectors/pipeline-connectors/mysql.md — Added "Snapshot
  Filters" section with a YAML pipeline example demonstrating the
  scan.snapshot.filters list syntax.
• docs/connectors/flink-sources/mysql-cdc.md — Added "Snapshot
  Filter" section with both SQL DDL and DataStream API examples
  showing the scan.snapshot.filter option usage.

Could we store the filters as a class member variable instead of constructing it on every invocation?

Optimized filter lookup in commit 9e8b5f5:

• Moved snapshot filter resolution from
  createDataEventsForTable() to the constructor of
  MySqlSnapshotSplitReadTask.
• The filter is now computed once and stored as a class member
  variable (snapshotFilter), avoiding repeated lookups on every
  method invocation.