update

Author: Hisoka-X

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.java

@@ -56,7 +56,7 @@
  * {@link JdbcConnection} connection extension used for connecting to Postgres instances.
  *
  * @author Horia Chiorean
- *     <p>Copied from Debezium 1.9.8-Final with three additional methods:
+ *     <p>Copied from Debezium 1.9.8-Final with the following modifications:
  *     <ul>
  *       <li>Constructor PostgresConnection( Configuration config, PostgresValueConverterBuilder
  *           valueConverterBuilder, ConnectionFactory factory) to allow passing a custom
@@ -66,6 +66,13 @@
  *       <li>override isTableUniqueIndexIncluded: Copied DBZ-5398 from Debezium 2.0.0.Final to fix
  *           https://github.com/ververica/flink-cdc-connectors/issues/2710. Remove this comment
  *           after bumping debezium version to 2.0.0.Final.
+ *       <li>FLINK-38965: Modified doReadTableColumn to filter out columns from other tables that
+ *           might be returned due to PostgreSQL LIKE wildcard matching. The underscore '_' matches
+ *           any single character, and '%' matches any sequence of characters. For example, when
+ *           querying table 'user_sink', the LIKE pattern may also match 'userbsink' (due to '_');
+ *           when querying table 'user%data' (where % is a literal character in the table name), the
+ *           LIKE pattern may also match 'user_test_data' (due to '%'). See also:
+ *           https://github.com/debezium/debezium/blob/main/debezium-core/src/main/java/io/debezium/jdbc/JdbcConnection.java#L1327
  *     </ul>
  */
 public class PostgresConnection extends JdbcConnection {
@@ -697,9 +704,11 @@ private Optional<ColumnEditor> doReadTableColumn(
             ResultSet columnMetadata, TableId tableId, Tables.ColumnNameFilter columnFilter)
             throws SQLException {
         // FLINK-38965: Filter out columns from other tables that might be returned due to
-        // PostgreSQL LIKE wildcard matching (underscore '_' matches any single character).
-        // For example, when querying 'ndi_pg_user_sink_1', the LIKE pattern may also match
-        // 'ndi_pg_userbsink_1' because '_' acts as a wildcard.
+        // PostgreSQL LIKE wildcard matching. The underscore '_' matches any single character,
+        // and '%' matches any sequence of characters. For example:
+        // - When querying 'user_sink', the pattern may also match 'userbsink' (due to '_')
+        // - When querying 'user%data' (where % is literal), it may match 'user_test_data' (due to
+        // '%')
         final String resultTableName = columnMetadata.getString(3);
         if (!tableId.table().equals(resultTableName)) {
             return Optional.empty();

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/SimilarTableNamesITCase.java

@@ -48,7 +48,18 @@
 
 /**
  * IT tests for FLINK-38965: Fix PostgreSQL CDC connector issue when table names contain underscore
- * that matches other tables due to LIKE wildcard behavior.
+ * or percent characters that match other tables due to LIKE wildcard behavior.
+ *
+ * <p>PostgreSQL LIKE wildcards:
+ *
+ * <ul>
+ *   <li>'_' (underscore) matches any single character. E.g., 'user_sink' matches 'userbsink'
+ *   <li>'%' (percent) matches any sequence of characters. E.g., 'user%data' matches
+ *       'user_test_data'
+ * </ul>
+ *
+ * <p>When table names contain these special characters, JDBC metadata queries using LIKE may return
+ * columns from unintended tables.
  */
 @Timeout(value = 300, unit = TimeUnit.SECONDS)
 class SimilarTableNamesITCase extends PostgresTestBase {
@@ -135,17 +146,64 @@ void testReadTableWithSimilarNameUnderscore() throws Exception {
     }
 
     /**
-     * Test reading from both similar-named tables to verify they can be captured independently
+     * Test that when capturing CDC events for table 'user%data' (which contains a literal '%'
+     * character), we don't accidentally capture events from 'user_test_data' which would match due
+     * to PostgreSQL's LIKE wildcard behavior (percent '%' matches any sequence of characters).
+     *
+     * <p>When querying for table 'user%data', the LIKE pattern may also match 'user_test_data'
+     * because '%' matches the '_test_' sequence.
+     */
+    @Test
+    void testReadTableWithSimilarNamePercent() throws Exception {
+        StreamTableEnvironment tEnv = createTableEnv();
+
+        // Only capture events from 'user%data' table (note: % is a literal character in table name)
+        tEnv.executeSql(createSourceDDL("target_table", "user%data"));
+
+        TableResult result = tEnv.executeSql("SELECT * FROM target_table");
+        CloseableIterator<Row> iterator = result.collect();
+
+        try {
+            // Verify snapshot data (3 rows from user%data only)
+            List<String> expectedSnapshotData =
+                    Arrays.asList(
+                            "+I[201, percent_1, Tianjin]",
+                            "+I[202, percent_2, Dalian]",
+                            "+I[203, percent_3, Qingdao]");
+            assertRowsEquals(collectRows(iterator, 3), expectedSnapshotData);
+
+            // Perform DML on user_test_data table - these should NOT be captured
+            executeDmlOperationsOnTestDataTable();
+
+            // Perform DML on target table - these SHOULD be captured
+            try (Connection conn =
+                            getJdbcConnection(
+                                    POSTGRES_CONTAINER, similarNamesDatabase.getDatabaseName());
+                    Statement stmt = conn.createStatement()) {
+                stmt.execute(
+                        "INSERT INTO similar_names.\"user%data\" VALUES (204, 'percent_4', 'Xiamen')");
+            }
+
+            // Should only see the insert from target table, not user_test_data table
+            List<String> expectedStreamData = Arrays.asList("+I[204, percent_4, Xiamen]");
+            assertRowsEquals(collectRows(iterator, 1), expectedStreamData);
+        } finally {
+            closeResourcesAndWaitForJobTermination(iterator, result);
+        }
+    }
+
+    /**
+     * Test reading from all similar-named tables to verify they can be captured independently
      * without interference.
      */
     @Test
-    void testReadBothSimilarNamedTables() throws Exception {
+    void testReadAllSimilarNamedTables() throws Exception {
         StreamTableEnvironment tEnv = createTableEnv();
 
-        // Capture events from both tables using regex pattern
-        tEnv.executeSql(createSourceDDL("all_tables", "ndi_pg_user.*"));
+        // Capture events from underscore-related tables using regex pattern
+        tEnv.executeSql(createSourceDDL("underscore_tables", "ndi_pg_user.*"));
 
-        TableResult result = tEnv.executeSql("SELECT * FROM all_tables");
+        TableResult result = tEnv.executeSql("SELECT * FROM underscore_tables");
         CloseableIterator<Row> iterator = result.collect();
 
         try {
@@ -229,7 +287,7 @@ private void assertRowsEquals(List<String> actual, List<String> expected) {
                         expected.stream().sorted().collect(Collectors.toList()));
     }
 
-    /** Executes DML operations on both tables for streaming phase testing. */
+    /** Executes DML operations on target and similar-named tables for streaming phase testing. */
     private void executeDmlOperations() throws Exception {
         try (Connection conn =
                         getJdbcConnection(
@@ -248,6 +306,22 @@ private void executeDmlOperations() throws Exception {
         }
     }
 
+    /** Executes DML operations on user_test_data table for '%' wildcard testing. */
+    private void executeDmlOperationsOnTestDataTable() throws Exception {
+        try (Connection conn =
+                        getJdbcConnection(
+                                POSTGRES_CONTAINER, similarNamesDatabase.getDatabaseName());
+                Statement stmt = conn.createStatement()) {
+            // Insert into user_test_data table (should NOT be captured when only listening to
+            // user%data)
+            stmt.execute(
+                    "INSERT INTO similar_names.user_test_data VALUES (304, 'test_4', 'Harbin')");
+            // Update user_test_data table
+            stmt.execute(
+                    "UPDATE similar_names.user_test_data SET address = 'Changchun' WHERE id = 301");
+        }
+    }
+
     /** Properly closes resources and waits for job termination. */
     private void closeResourcesAndWaitForJobTermination(
             CloseableIterator<Row> iterator, TableResult result) throws Exception {

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/resources/ddl/similar_names.sql

@@ -13,10 +13,12 @@
 -- See the License for the specific language governing permissions and
 -- limitations under the License.
 
--- FLINK-38965: Test case for similar table names with underscore
--- This tests the fix for PostgreSQL LIKE wildcard matching issue
--- where underscore '_' matches any single character, causing
--- 'ndi_pg_user_sink_1' to also match 'ndi_pg_userbsink_1'
+-- FLINK-38965: Test case for similar table names with underscore or percent characters
+-- This tests the fix for PostgreSQL LIKE wildcard matching issue:
+-- - underscore '_' matches any single character
+-- - percent '%' matches any sequence of characters
+-- For example, 'user_sink' may match 'userbsink' (due to '_')
+-- and 'user%sink' may match 'user_test_sink' (due to '%')
 
 DROP SCHEMA IF EXISTS similar_names CASCADE;
 CREATE SCHEMA similar_names;
@@ -49,3 +51,34 @@ INSERT INTO ndi_pg_userbsink_1
 VALUES (101, 'userb_1', 'Guangzhou'),
        (102, 'userb_2', 'Shenzhen'),
        (103, 'userb_3', 'Chengdu');
+
+-- Table 3: user%data (tests '%' wildcard scenario)
+-- The table name contains '%' character which acts as a wildcard in LIKE pattern.
+-- When querying for table 'user%data', the LIKE pattern may also match
+-- 'user_test_data' because '%' matches any sequence of characters.
+CREATE TABLE "user%data" (
+  id INTEGER NOT NULL PRIMARY KEY,
+  name VARCHAR(255) NOT NULL,
+  address VARCHAR(1024)
+);
+ALTER TABLE "user%data" REPLICA IDENTITY FULL;
+
+INSERT INTO "user%data"
+VALUES (201, 'percent_1', 'Tianjin'),
+       (202, 'percent_2', 'Dalian'),
+       (203, 'percent_3', 'Qingdao');
+
+-- Table 4: user_test_data (similar to 'user%data' when % is treated as wildcard)
+-- This table name would match the LIKE pattern for 'user%data'
+-- because '%' matches '_test_' sequence.
+CREATE TABLE user_test_data (
+  id INTEGER NOT NULL PRIMARY KEY,
+  name VARCHAR(255) NOT NULL,
+  address VARCHAR(1024)
+);
+ALTER TABLE user_test_data REPLICA IDENTITY FULL;
+
+INSERT INTO user_test_data
+VALUES (301, 'test_1', 'Harbin'),
+       (302, 'test_2', 'Changchun'),
+       (303, 'test_3', 'Shenyang');