[FLINK-38188][pipeline-connector][postgres] Fix database name validation logic in PostgresDataSourceFactory (#4075)

---------

Co-authored-by: lvyanquan <lvyanquan.lyq@alibaba-inc.com>

Author: tchivs

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java

@@ -369,23 +369,19 @@ private Optional<String> getValidateDatabaseName(String tables) {
                     String.format(
                             "Tables format must db.schema.table, can not 'tables' = %s",
                             TABLES.key()));
-            if (tableNameParts.length == 3) {
-                String currentDbName = tableNameParts[0];
+            String currentDbName = tableNameParts[0];
 
+            checkState(
+                    isValidPostgresDbName(currentDbName),
+                    String.format("%s is not a valid PostgreSQL database name", currentDbName));
+            if (dbName == null) {
+                dbName = currentDbName;
+            } else {
                 checkState(
-                        isValidPostgresDbName(currentDbName),
-                        String.format(
-                                "The value of option %s does not conform to PostgresSQL database name naming conventions",
-                                TABLES.key()));
-                if (dbName == null) {
-                    dbName = currentDbName;
-                } else {
-                    checkState(
-                            !dbName.equals(currentDbName),
-                            String.format(
-                                    "The value of option %s all table names must have the same database name",
-                                    TABLES.key()));
-                }
+                        dbName.equals(currentDbName),
+                        "The value of option `%s` is `%s`, but not all table names have the same database name",
+                        TABLES.key(),
+                        String.join(",", tableNames));
             }
         }
 

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactoryTest.java

@@ -257,6 +257,54 @@ public void testPrefixRequireOption() {
                 .isEqualTo(Arrays.asList("inventory.products"));
     }
 
+    @Test
+    public void testTableValidationWithDifferentDatabases() {
+        Map<String, String> options = new HashMap<>();
+        options.put(HOSTNAME.key(), POSTGRES_CONTAINER.getHost());
+        options.put(
+                PG_PORT.key(), String.valueOf(POSTGRES_CONTAINER.getMappedPort(POSTGRESQL_PORT)));
+        options.put(USERNAME.key(), TEST_USER);
+        options.put(PASSWORD.key(), TEST_PASSWORD);
+        options.put(
+                TABLES.key(),
+                "aia_test.public.aia_t_icc_jjdb,different_db.public.aia_t_icc_jjdb_extend");
+        options.put(SLOT_NAME.key(), slotName);
+
+        PostgresDataSourceFactory factory = new PostgresDataSourceFactory();
+        Factory.Context context = new MockContext(Configuration.fromMap(options));
+
+        assertThatThrownBy(() -> factory.createDataSource(context))
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining(
+                        "The value of option `tables` is `aia_test.public.aia_t_icc_jjdb,different_db.public.aia_t_icc_jjdb_extend`, but not all table names have the same database name");
+    }
+
+    @Test
+    public void testTableValidationWithOriginalBugScenario() {
+        Map<String, String> options = new HashMap<>();
+        options.put(HOSTNAME.key(), POSTGRES_CONTAINER.getHost());
+        options.put(
+                PG_PORT.key(), String.valueOf(POSTGRES_CONTAINER.getMappedPort(POSTGRESQL_PORT)));
+        options.put(USERNAME.key(), TEST_USER);
+        options.put(PASSWORD.key(), TEST_PASSWORD);
+        String tables =
+                POSTGRES_CONTAINER.getDatabaseName()
+                        + ".public.aia_t_icc_jjdb,"
+                        + POSTGRES_CONTAINER.getDatabaseName()
+                        + ".public.aia_t_icc_jjdb_\\\\d{6},"
+                        + POSTGRES_CONTAINER.getDatabaseName()
+                        + ".public.aia_t_icc_jjdb_extend";
+        options.put(TABLES.key(), tables);
+        options.put(SLOT_NAME.key(), slotName);
+
+        PostgresDataSourceFactory factory = new PostgresDataSourceFactory();
+        Factory.Context context = new MockContext(Configuration.fromMap(options));
+
+        assertThatThrownBy(() -> factory.createDataSource(context))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("Cannot find any table by the option 'tables'");
+    }
+
     class MockContext implements Factory.Context {
 
         Configuration factoryConfiguration;

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/PostgresE2eITCase.java

@@ -111,7 +111,7 @@ void testSyncWholeDatabase() throws Exception {
                                 + "  port: %d\n"
                                 + "  username: %s\n"
                                 + "  password: %s\n"
-                                + "  tables: %s.inventory.\\.*\n"
+                                + "  tables: %s.inventory.products,%s.inventory.customers\n"
                                 + "  slot.name: %s\n"
                                 + "  scan.startup.mode: initial\n"
                                 + "  server-time-zone: UTC\n"
@@ -127,6 +127,7 @@ void testSyncWholeDatabase() throws Exception {
                         POSTGRES_TEST_USER,
                         POSTGRES_TEST_PASSWORD,
                         postgresInventoryDatabase.getDatabaseName(),
+                        postgresInventoryDatabase.getDatabaseName(),
                         slotName,
                         parallelism);
         Path postgresCdcJar = TestUtils.getResource("postgres-cdc-pipeline-connector.jar");