ruanhang1993 commented on PR #4422:
URL: https://github.com/apache/flink-cdc/pull/4422#issuecomment-5248267854
Hi, @taoran92. Thanks for working on this issue. The profiling result and
the caching direction look good, and the added tests clearly demonstrate the
performance improvement.
Before merging, could we move the cache implementation to the common
relational Debezium layer instead of keeping it in `MySqlSourceConfig`?
The repeated regular-expression evaluation happens in
`RelationalTableFilters.dataCollectionFilter()`, which is shared by multiple
relational connectors, including MySQL, PostgreSQL, SQL Server, Oracle, and
DB2. Therefore, this is not MySQL-specific. Keeping `createCachedTableFilter()`
in `MySqlSourceConfig` would only fix MySQL and may lead to similar cache
implementations being added independently by other connectors later.
A more general approach would be:
1. Introduce a reusable `CachedTableFilter` in `flink-connector-debezium`.
2. Wrap the initial table filter when `RelationalTableFilters` is
constructed, so all relational connectors benefit from the cache.
3. Keep `setDataCollectionFilters()` as a direct assignment:
```java
public void setDataCollectionFilters(TableFilter tableFilter) {
this.tableFilter = tableFilter;
}
```
If a caller needs caching, it should pass an already cached filter to
this method. This preserves the existing setter semantics.
4. Let `CachedTableFilter.from(filter)` return the existing instance when
the filter is already cached, avoiding accidental repeated wrapping.
MySQL needs a little extra care because `excludeTableList` is an additional
Flink CDC filter on top of Debezium's include/exclude rules. Simply wrapping
the existing cached filter again could produce:
```text
Cache B
-> Cache A
-> original Debezium filter
```
It would be better for the cached filter to retain its raw delegate and
provide something like `withAdditionalFilter(...)`, which combines the raw
delegate with the MySQL exclude filter and then creates one new cache:
```text
Cache
-> original Debezium filter
-> MySQL exclude filter
```
A possible API shape is:
```java
public static CachedTableFilter from(Tables.TableFilter filter);
public CachedTableFilter withAdditionalFilter(
Tables.TableFilter additionalFilter);
```
When implementing `withAdditionalFilter`, it would also be good to capture
the raw delegate in a local variable rather than capturing the old
`CachedTableFilter` instance, so that the previous cache is not retained by the
new filter.
WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
