[FLINK-36165][source-connector/mysql] Support capturing snapshot data with conditions

[uicosp]

Hey, this is an implementation designed to capture snapshot data with filtering conditions.

For example, by specifying scan.snapshot.filters: db.user_table:id > 200;, we can synchronize only the user data where the id is greater than 200.

issue link: https://issues.apache.org/jira/browse/FLINK-36165

[yuxiqian]

Thanks for @uicosp's great work! It's indeed a long awaited feature.

Seems Debezium has a similar option called snapshot.select.statement.overrides, which allows users to project out unwanted columns and filter rows based on custom predicates.

As these options aren't available in incremental framework, it would be nice if we could support both row and column level pushdown with similar syntax, since they're both related to tweaking snapshot querying SQL. WDYT?

[Thorne-coder]

Thanks for @uicosp's great work! It's indeed a long awaited feature.

Seems Debezium has a similar option called snapshot.select.statement.overrides, which allows users to project out unwanted columns and filter rows based on custom predicates.

As these options aren't available in incremental framework, it would be nice if we could support both row and column level pushdown with similar syntax, since they're both related to tweaking snapshot querying SQL. WDYT?

it is also support table api and datastream api with incremental framework

[Thorne-coder]

Should we consider this pr with only snapshot startoption and back fill ?

[uicosp]

Column pruning is certainly a feature worth considering. In theory, it would only require minor adjustments to the current design, such as implementing a syntax like scan.snapshot.filters: db.user:id>200:id,name,age.

Taking this a step further, since the pipeline config already includes a transform option, we could leverage it to automatically extract filter conditions and projection fields, dynamically generating scan.snapshot.filters and thereby simplifying the config.

That said, I’m currently focused on other tasks and don't have time to work on this. Once I have time, I’d be happy to revisit and enhance this feature.