[FLINK-36165][source-connector/mysql] Support capturing snapshot data with conditions #4104
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
github-actions
bot
added
docs
… with conditions [FLINK-36165][source-connector/mysql] add docs [FLINK-36165][source-connector/mysql] Implement snapshot filter for MySQL table source [FLINK-36165][source-connector/mysql] Escape dot [FLINK-36165 ] fixed supported escape like 'city != 'China:beijing''
Mrart
force-pushed
the
FLINK-36165
branch
3 times, most recently
from
d0738af to
c3005b6
Compare
fixed checkstyle fixed test fixed MySqlTableSourceFactoryTest test error.
SML0127
reviewed
Sep 7, 2025
Comment on lines
+296
to
+306
| <tr> | ||
| <td>scan.snapshot.filters</td> | ||
| <td>optional</td> | ||
| <td style="word-wrap: break-word;">(none)</td> | ||
| <td>String</td> | ||
| <td>When reading a table snapshot, the rows of captured tables will be filtered using the specified filter expression (AKA a SQL WHERE clause). <br> | ||
| By default, no filter is applied, meaning the entire table will be synchronized. <br> | ||
| A colon (:) separates table name and filter expression, while a semicolon (;) separate multiple filters, | ||
| e.g. `db1.user_table_[0-9]+:id > 100;db[1-2].[app|web]_order_\\.*:id < 0;`. | ||
| </td> | ||
| </tr> |
There was a problem hiding this comment.
How about explicitly stating that the filter conditions are combined using AND and that it has nothing to do with the binlog step?
SML0127
reviewed
Sep 7, 2025
Comment on lines
+58
to
+67
| Map<Selectors, String> snapshotFilters = toSelector(filters); | ||
|
|
||
| String filter = null; | ||
| for (Selectors selector : snapshotFilters.keySet()) { | ||
| if (selector.isMatch( | ||
| org.apache.flink.cdc.common.event.TableId.tableId( | ||
| tableId.catalog(), tableId.table()))) { | ||
| filter = snapshotFilters.get(selector); | ||
| break; | ||
| } |
There was a problem hiding this comment.
Can we fail fast on unknown/nonexiststent columns in filters?
SML0127
reviewed
Sep 7, 2025
Comment on lines
+64
to
+88
| public static Long queryRowCnt( | ||
| JdbcConnection jdbc, TableId tableId, String columnName, @Nullable String filter) | ||
| throws SQLException { | ||
|
|
||
| if (filter == null) { | ||
| return queryApproximateRowCnt(jdbc, tableId); | ||
| } | ||
|
|
||
| final String cntQuery = | ||
| String.format( | ||
| "SELECT COUNT(%s) FROM %s WHERE %s", | ||
| quote(columnName), quote(tableId), filter); | ||
| return jdbc.queryAndMap( | ||
| cntQuery, | ||
| rs -> { | ||
| if (!rs.next()) { | ||
| // this should never happen | ||
| throw new SQLException( | ||
| String.format( | ||
| "No result returned after running query [%s]", cntQuery)); | ||
| } | ||
| return rs.getLong(1); | ||
| }); | ||
| } | ||
|
|
There was a problem hiding this comment.
Because the filter is applied during split planning, the distribution factor may fall outside the configured bounds(0.05 ~1,000), affecting snapshot performance. I’d appreciate your thoughts on this.
SML0127
reviewed
Sep 7, 2025
|
It seems the author of #3776 is actively working on their PR, and there's some duplicated work... It would be nice if we can discuss & implement this nice feature at one place. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
issue link: https://issues.apache.org/jira/browse/FLINK-36165
fixed #3776