From 3cc72bc4e0fcb88b8f2ee5e917c450d53d50bb1e Mon Sep 17 00:00:00 2001
From: harveyyue <yw_yuewei@126.com>
Date: Mon, 4 Dec 2023 11:39:17 +0800
Subject: [PATCH] DBZ-7167 Provide a public API from the connector
 implementations to retrieve the list of matching collections or tables based
 on the different include-/exclude lists

---
 .../debezium/connector/vitess/VitessConnector.java | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/src/main/java/io/debezium/connector/vitess/VitessConnector.java b/src/main/java/io/debezium/connector/vitess/VitessConnector.java
index da6e8920..f22bb414 100644
--- a/src/main/java/io/debezium/connector/vitess/VitessConnector.java
+++ b/src/main/java/io/debezium/connector/vitess/VitessConnector.java
@@ -32,6 +32,7 @@
 import io.debezium.connector.common.RelationalBaseSourceConnector;
 import io.debezium.connector.vitess.connection.VitessReplicationConnection;
 import io.debezium.relational.RelationalDatabaseConnectorConfig;
+import io.debezium.relational.TableId;
 import io.debezium.util.Strings;
 import io.grpc.StatusRuntimeException;
 import io.vitess.proto.Query;
@@ -391,4 +392,17 @@ protected Map<String, ConfigValue> validateAllFields(Configuration config) {
         }
         return results;
     }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public List<TableId> getMatchingCollections(Configuration configuration) {
+        VitessConnectorConfig vitessConnectorConfig = new VitessConnectorConfig(configuration);
+        String keyspace = vitessConnectorConfig.getKeyspace();
+        List<String> allTables = getKeyspaceTables(vitessConnectorConfig);
+        List<String> includedTables = getIncludedTables(keyspace,
+                vitessConnectorConfig.tableIncludeList(), allTables);
+        return includedTables.stream()
+                .map(table -> new TableId(keyspace, null, table))
+                .collect(Collectors.toList());
+    }
 }