This is an automated email from the ASF dual-hosted git repository.
ruanhang1993 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 91040dece [FLINK-39824][flink-connector-debezium] Cache relational
table filter results (#4422)
91040dece is described below
commit 91040dece074c18640a58f74a6391cf7613f8063
Author: Ran Tao <[email protected]>
AuthorDate: Wed Aug 12 14:32:35 2026 +0800
[FLINK-39824][flink-connector-debezium] Cache relational table filter
results (#4422)
Introduce a bounded CachedTableFilter in flink-connector-debezium and
use it as the default relational table filter.
---
.../io/debezium/relational/CachedTableFilter.java | 71 +++++++++++++
.../relational/RelationalTableFilters.java | 9 +-
.../debezium/relational/CachedTableFilterTest.java | 112 +++++++++++++++++++++
.../relational/RelationalTableFiltersTest.java | 49 +++++++++
.../mysql/source/config/MySqlSourceConfig.java | 19 ++--
.../mysql/source/config/MySqlSourceConfigTest.java | 54 ++++++++++
6 files changed, 303 insertions(+), 11 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/CachedTableFilter.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/CachedTableFilter.java
new file mode 100644
index 000000000..18e9ef93a
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/CachedTableFilter.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.debezium.relational;
+
+import org.apache.flink.shaded.guava31.com.google.common.cache.CacheBuilder;
+import org.apache.flink.shaded.guava31.com.google.common.cache.CacheLoader;
+import org.apache.flink.shaded.guava31.com.google.common.cache.LoadingCache;
+
+import io.debezium.relational.Tables.TableFilter;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** A bounded cache for table filter results. */
+public class CachedTableFilter implements TableFilter {
+
+ private static final long TABLE_FILTER_CACHE_MAXIMUM_SIZE = 32 * 1024;
+
+ private final TableFilter rawTableFilter;
+ private final LoadingCache<TableId, Boolean> tableFilterCache;
+
+ private CachedTableFilter(TableFilter rawTableFilter) {
+ this.rawTableFilter = rawTableFilter;
+ this.tableFilterCache =
+ CacheBuilder.newBuilder()
+ .maximumSize(TABLE_FILTER_CACHE_MAXIMUM_SIZE)
+ .build(
+ new CacheLoader<TableId, Boolean>() {
+ @Override
+ public Boolean load(TableId tableId) {
+ return
rawTableFilter.isIncluded(tableId);
+ }
+ });
+ }
+
+ /** Wraps the given filter in a cache, or returns it unchanged if it is
already cached. */
+ public static CachedTableFilter from(TableFilter tableFilter) {
+ checkNotNull(tableFilter);
+ if (tableFilter instanceof CachedTableFilter) {
+ return (CachedTableFilter) tableFilter;
+ }
+ return new CachedTableFilter(tableFilter);
+ }
+
+ /** Returns a new cached filter that also requires the additional filter
to match. */
+ public CachedTableFilter withAdditionalFilter(TableFilter
additionalFilter) {
+ checkNotNull(additionalFilter);
+ TableFilter rawFilter = rawTableFilter;
+ return new CachedTableFilter(
+ tableId -> rawFilter.isIncluded(tableId) &&
additionalFilter.isIncluded(tableId));
+ }
+
+ @Override
+ public boolean isIncluded(TableId tableId) {
+ return tableFilterCache.getUnchecked(tableId);
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/RelationalTableFilters.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/RelationalTableFilters.java
index e36c936d5..4d91ff83c 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/RelationalTableFilters.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/RelationalTableFilters.java
@@ -20,7 +20,9 @@ import static
io.debezium.relational.RelationalDatabaseConnectorConfig.COLUMN_EX
/**
* Copied from Debezium 1.9.8.Final.
*
- * <p>Line 146: add a method to update the tableFilter variable.
+ * <p>Line 98: cache table filter results.
+ *
+ * <p>Line 148: add a method to update the tableFilter variable.
*/
public class RelationalTableFilters implements DataCollectionFilters {
@@ -92,7 +94,8 @@ public class RelationalTableFilters implements
DataCollectionFilters {
? tablePredicate.and(systemTablesFilter::isIncluded)
: tablePredicate;
- this.tableFilter = finalTablePredicate::test;
+ TableFilter initialTableFilter = finalTablePredicate::test;
+ this.tableFilter = CachedTableFilter.from(initialTableFilter);
// Define the database filter using the include and exclude lists for
database names ...
this.databaseFilter =
@@ -114,7 +117,7 @@ public class RelationalTableFilters implements
DataCollectionFilters {
this.schemaSnapshotFilter =
config.getBoolean(DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL)
- ?
eligibleSchemaPredicate.and(tableFilter::isIncluded)::test
+ ?
eligibleSchemaPredicate.and(initialTableFilter::isIncluded)::test
: eligibleSchemaPredicate::test;
this.excludeColumns =
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/test/java/io/debezium/relational/CachedTableFilterTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/test/java/io/debezium/relational/CachedTableFilterTest.java
new file mode 100644
index 000000000..7b4f65c3e
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/test/java/io/debezium/relational/CachedTableFilterTest.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.debezium.relational;
+
+import io.debezium.relational.Tables.TableFilter;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class CachedTableFilterTest {
+
+ @Test
+ void testCachesPositiveAndNegativeResultsForEquivalentTableIds() {
+ AtomicInteger invocationCount = new AtomicInteger();
+ CachedTableFilter cachedTableFilter =
+ CachedTableFilter.from(
+ tableId -> {
+ invocationCount.incrementAndGet();
+ return tableId.table().startsWith("orders");
+ });
+
+ TableId includedTable = new TableId("test_db", null, "orders_1");
+ TableId sameIncludedTable = new TableId("test_db", null, "orders_1");
+ TableId unmatchedTable = new TableId("test_db", null, "customers");
+ TableId sameUnmatchedTable = new TableId("test_db", null, "customers");
+
+ assertThat(cachedTableFilter.isIncluded(includedTable)).isTrue();
+ assertThat(cachedTableFilter.isIncluded(sameIncludedTable)).isTrue();
+ assertThat(cachedTableFilter.isIncluded(unmatchedTable)).isFalse();
+ assertThat(cachedTableFilter.isIncluded(sameUnmatchedTable)).isFalse();
+ assertThat(invocationCount).hasValue(2);
+ }
+
+ @Test
+ void testFromReturnsCachedFilterUnchanged() {
+ CachedTableFilter cachedTableFilter = CachedTableFilter.from(tableId
-> true);
+
+
assertThat(CachedTableFilter.from(cachedTableFilter)).isSameAs(cachedTableFilter);
+ }
+
+ @Test
+ void testAdditionalFilterCreatesSingleCacheOverRawDelegate() {
+ AtomicInteger rawFilterInvocationCount = new AtomicInteger();
+ AtomicInteger additionalFilterInvocationCount = new AtomicInteger();
+ CachedTableFilter cachedTableFilter =
+ CachedTableFilter.from(
+ tableId -> {
+ rawFilterInvocationCount.incrementAndGet();
+ return true;
+ });
+ TableId tableId = new TableId("test_db", null, "orders_1");
+
+ assertThat(cachedTableFilter.isIncluded(tableId)).isTrue();
+
+ CachedTableFilter combinedFilter =
+ cachedTableFilter.withAdditionalFilter(
+ ignored -> {
+ additionalFilterInvocationCount.incrementAndGet();
+ return false;
+ });
+
+ assertThat(combinedFilter).isNotSameAs(cachedTableFilter);
+ assertThat(combinedFilter.isIncluded(tableId)).isFalse();
+ assertThat(combinedFilter.isIncluded(tableId)).isFalse();
+ assertThat(rawFilterInvocationCount).hasValue(2);
+ assertThat(additionalFilterInvocationCount).hasValue(1);
+
+ assertThat(cachedTableFilter.isIncluded(tableId)).isTrue();
+ assertThat(rawFilterInvocationCount).hasValue(2);
+ }
+
+ @Test
+ void testCacheSizeIsBounded() {
+ AtomicInteger invocationCount = new AtomicInteger();
+ int numberOfTableIds = 32 * 1024 + 1;
+ TableFilter cachedTableFilter =
+ CachedTableFilter.from(
+ tableId -> {
+ invocationCount.incrementAndGet();
+ return true;
+ });
+ List<TableId> tableIds = new ArrayList<>();
+ for (int i = 0; i < numberOfTableIds; i++) {
+ tableIds.add(new TableId("test_db", null, "table_" + i));
+ }
+
+ tableIds.forEach(cachedTableFilter::isIncluded);
+ assertThat(invocationCount).hasValue(numberOfTableIds);
+
+ tableIds.forEach(cachedTableFilter::isIncluded);
+ assertThat(invocationCount).hasValueGreaterThan(numberOfTableIds);
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/test/java/io/debezium/relational/RelationalTableFiltersTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/test/java/io/debezium/relational/RelationalTableFiltersTest.java
new file mode 100644
index 000000000..836bc9122
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/test/java/io/debezium/relational/RelationalTableFiltersTest.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.debezium.relational;
+
+import io.debezium.config.Configuration;
+import io.debezium.relational.Tables.TableFilter;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class RelationalTableFiltersTest {
+
+ @Test
+ void testInitialTableFilterIsCached() {
+ RelationalTableFilters tableFilters = createTableFilters();
+
+
assertThat(tableFilters.dataCollectionFilter()).isInstanceOf(CachedTableFilter.class);
+ }
+
+ @Test
+ void testSetDataCollectionFiltersRetainsReplacement() {
+ RelationalTableFilters tableFilters = createTableFilters();
+ TableFilter replacementFilter = tableId -> false;
+
+ tableFilters.setDataCollectionFilters(replacementFilter);
+
+
assertThat(tableFilters.dataCollectionFilter()).isSameAs(replacementFilter);
+ }
+
+ private static RelationalTableFilters createTableFilters() {
+ return new RelationalTableFilters(
+ Configuration.empty(), tableId -> true, TableId::toString);
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java
index cf456fcae..6b87dfef9 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java
@@ -24,6 +24,7 @@ import org.apache.flink.table.catalog.ObjectPath;
import io.debezium.config.Configuration;
import io.debezium.connector.mysql.MySqlConnectorConfig;
+import io.debezium.relational.CachedTableFilter;
import io.debezium.relational.RelationalTableFilters;
import io.debezium.relational.TableId;
import io.debezium.relational.Tables;
@@ -143,14 +144,16 @@ public class MySqlSourceConfig implements Serializable {
(excludeTableList == null
? null
: new
Selectors.SelectorsBuilder().includeTables(excludeTableList).build());
- Tables.TableFilter tableFilter =
dbzMySqlConfig.getTableFilters().dataCollectionFilter();
- dbzMySqlConfig
- .getTableFilters()
- .setDataCollectionFilters(
- (TableId tableId) ->
- tableFilter.isIncluded(tableId)
- && (excludeTableFilter == null
- ||
!excludeTableFilter.isMatch(tableId)));
+ if (excludeTableFilter != null) {
+ Tables.TableFilter tableFilter =
+ dbzMySqlConfig.getTableFilters().dataCollectionFilter();
+ dbzMySqlConfig
+ .getTableFilters()
+ .setDataCollectionFilters(
+ CachedTableFilter.from(tableFilter)
+ .withAdditionalFilter(
+ tableId ->
!excludeTableFilter.isMatch(tableId)));
+ }
this.jdbcProperties = jdbcProperties;
this.chunkKeyColumns = chunkKeyColumns;
this.skipSnapshotBackfill = skipSnapshotBackfill;
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigTest.java
new file mode 100644
index 000000000..5a88eada4
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigTest.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.mysql.source.config;
+
+import io.debezium.relational.CachedTableFilter;
+import io.debezium.relational.TableId;
+import org.junit.jupiter.api.Test;
+
+import java.util.function.Predicate;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link MySqlSourceConfig}. */
+class MySqlSourceConfigTest {
+
+ @Test
+ void testTableFilterWithExcludeTableList() {
+ MySqlSourceConfig config =
+ new MySqlSourceConfigFactory()
+ .hostname("localhost")
+ .username("user")
+ .password("password")
+ .databaseList("test_db")
+ .tableList("test_db\\.orders_.*")
+ .excludeTableList("test_db.orders_skip")
+ .createConfig(0);
+
+ Predicate<TableId> tableFilter = config.getTableFilter();
+ TableId includedTable = new TableId("test_db", null, "orders_1");
+ TableId excludedTable = new TableId("test_db", null, "orders_skip");
+ TableId unmatchedTable = new TableId("test_db", null, "customers");
+
+ assertThat(config.getTableFilters().dataCollectionFilter())
+ .isInstanceOf(CachedTableFilter.class);
+ assertThat(tableFilter.test(includedTable)).isTrue();
+ assertThat(tableFilter.test(excludedTable)).isFalse();
+ assertThat(tableFilter.test(unmatchedTable)).isFalse();
+ }
+}
