[FLINK-38514][postgres] Add support for UUID array type in PostgreSQL CDC connector

Author: Hisoka-X

docs/content.zh/docs/connectors/flink-sources/postgres-cdc.md

@@ -736,6 +736,7 @@ Data Type Mapping
         CHARACTER(n)<br>
         VARCHAR(n)<br>
         CHARACTER VARYING(n)<br>
+        UUID<br>
         TEXT</td>
       <td>STRING</td>
     </tr>

docs/content/docs/connectors/flink-sources/postgres-cdc.md

@@ -740,6 +740,7 @@ Since the order of processing these records cannot be guaranteed, the final valu
         CHARACTER(n)<br>
         VARCHAR(n)<br>
         CHARACTER VARYING(n)<br>
+        UUID<br>
         TEXT</td>
       <td>STRING</td>
     </tr>

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/utils/PostgresTypeUtils.java

@@ -61,6 +61,7 @@ public class PostgresTypeUtils {
     private static final String PG_CHARACTER_VARYING = "varchar";
     private static final String PG_CHARACTER_VARYING_ARRAY = "_varchar";
     private static final String PG_UUID = "uuid";
+    private static final String PG_UUID_ARRAY = "_uuid";
 
     /** Returns a corresponding Flink data type from a debezium {@link Column}. */
     public static DataType fromDbzColumn(Column column) {
@@ -140,6 +141,7 @@ private static DataType convertFromColumn(Column column) {
             case PG_UUID:
                 return DataTypes.STRING();
             case PG_TEXT_ARRAY:
+            case PG_UUID_ARRAY:
                 return DataTypes.ARRAY(DataTypes.STRING());
             case PG_TIMESTAMP:
                 return DataTypes.TIMESTAMP(scale);

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

@@ -968,6 +968,85 @@ void testUpsertMode(boolean parallelismSnapshot) throws Exception {
         result.getJobClient().get().cancel().get();
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    void testArrayTypes(boolean parallelismSnapshot) throws Throwable {
+        setup(parallelismSnapshot);
+        initializePostgresTable(POSTGIS_CONTAINER, "column_type_test");
+
+        String sourceDDL =
+                String.format(
+                        "CREATE TABLE array_types ("
+                                + "    id INTEGER NOT NULL,"
+                                + "    text_a1 ARRAY<STRING>,"
+                                + "    int_a1 ARRAY<INT>,"
+                                + "    int_s1 ARRAY<INT>,"
+                                + "    uuid_a1 ARRAY<STRING>"
+                                + ") WITH ("
+                                + " 'connector' = 'postgres-cdc',"
+                                + " 'hostname' = '%s',"
+                                + " 'port' = '%s',"
+                                + " 'username' = '%s',"
+                                + " 'password' = '%s',"
+                                + " 'database-name' = '%s',"
+                                + " 'schema-name' = '%s',"
+                                + " 'table-name' = '%s',"
+                                + " 'scan.incremental.snapshot.enabled' = '%s',"
+                                + " 'decoding.plugin.name' = 'pgoutput', "
+                                + " 'slot.name' = '%s'"
+                                + ")",
+                        POSTGIS_CONTAINER.getHost(),
+                        POSTGIS_CONTAINER.getMappedPort(POSTGRESQL_PORT),
+                        POSTGIS_CONTAINER.getUsername(),
+                        POSTGIS_CONTAINER.getPassword(),
+                        POSTGIS_CONTAINER.getDatabaseName(),
+                        "inventory",
+                        "array_types",
+                        parallelismSnapshot,
+                        getSlotName());
+
+        tEnv.executeSql(sourceDDL);
+
+        // async submit job
+        TableResult tableResult = tEnv.executeSql("SELECT * FROM array_types");
+        CloseableIterator<Row> iterator = tableResult.collect();
+
+        // wait for snapshot to complete
+        List<Row> snapshotResults = new ArrayList<>();
+        while (iterator.hasNext() && snapshotResults.size() < 1) {
+            Row row = iterator.next();
+            snapshotResults.add(row);
+        }
+
+        // verify snapshot data
+        Assertions.assertThat(snapshotResults).hasSize(1);
+        Row snapshotRow = snapshotResults.get(0);
+
+        // verify id
+        Assertions.assertThat(snapshotRow.getField(0)).isEqualTo(1);
+
+        // verify text_a1 array
+        String[] textArray = (String[]) snapshotRow.getField(1);
+        Assertions.assertThat(textArray).containsExactly("electronics", "gadget", "sale");
+
+        // verify int_a1 array
+        Integer[] intArray1 = (Integer[]) snapshotRow.getField(2);
+        Assertions.assertThat(intArray1).containsExactly(85, 90, 78);
+
+        // verify int_s1 array
+        Integer[] intArray2 = (Integer[]) snapshotRow.getField(3);
+        Assertions.assertThat(intArray2).containsExactly(42);
+
+        // verify uuid_a1 array - should be mapped to STRING array
+        String[] uuidArray = (String[]) snapshotRow.getField(4);
+        Assertions.assertThat(uuidArray).hasSize(2);
+        // UUID values should be present (format may vary)
+        Assertions.assertThat(uuidArray[0]).isNotEmpty();
+        Assertions.assertThat(uuidArray[1]).isNotEmpty();
+
+        tableResult.getJobClient().get().cancel().get();
+    }
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     void testUniqueIndexIncludingFunction(boolean parallelismSnapshot) throws Exception {

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

@@ -56,4 +56,28 @@ INSERT INTO inventory.full_types
 VALUES (1, '2', 32767, 65535, 2147483647, 5.5, 6.6, 123.12345, 404.4443, true,
         'Hello World', 'a', 'abc', 'abcd..xyz', '2020-07-17 18:00:22.123', '2020-07-17 18:00:22.123456',
         '2020-07-17', '18:00:22', 500, 'SRID=3187;POINT(174.9479 -36.7208)'::geometry,
-        'MULTILINESTRING((169.1321 -44.7032, 167.8974 -44.6414))'::geography);
+        'MULTILINESTRING((169.1321 -44.7032, 167.8974 -44.6414))'::geography);
+
+-- ----------------------------------------------------------------------------------------------------------------
+-- TABLE: array_types
+-- ----------------------------------------------------------------------------------------------------------------
+-- Test table for array types including UUID[]
+CREATE TABLE array_types (
+    id        SERIAL PRIMARY KEY,
+    text_a1   TEXT[],
+    int_a1    INTEGER[],
+    int_s1    INTEGER[],
+    uuid_a1   UUID[]
+);
+
+ALTER TABLE inventory.array_types
+    REPLICA IDENTITY FULL;
+
+INSERT INTO array_types (id, text_a1, int_a1, int_s1, uuid_a1)
+VALUES
+    (1,
+     ARRAY['electronics', 'gadget', 'sale'],
+     '{85, 90, 78}',
+     '{42}',
+     ARRAY['227496Ad-fdE9-CCFb-1F04-892fc505AFD5', '9d33f9E2-DfC7-fDef-9478-BCc5dBf7a6d7']::UUID[]
+    );