This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new d42625a5d5 [cdc] Support PostgreSQL uuid type in postgres sync (#8906)
d42625a5d5 is described below
commit d42625a5d50ea5760362e8bf71aa9e3fdec668df
Author: Arnav Balyan <[email protected]>
AuthorDate: Thu Jul 30 09:30:29 2026 +0530
[cdc] Support PostgreSQL uuid type in postgres sync (#8906)
=
---
.../paimon/flink/action/cdc/postgres/PostgresTypeUtils.java | 2 ++
.../action/cdc/postgres/PostgresSyncTableActionITCase.java | 4 ++++
.../flink/action/cdc/postgres/PostgresTypeUtilsTest.java | 10 ++++++++++
.../src/test/resources/postgres/sync_table_setup.sql | 8 +++++---
4 files changed, 21 insertions(+), 3 deletions(-)
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresTypeUtils.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresTypeUtils.java
index 29048477c1..78b80c458f 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresTypeUtils.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresTypeUtils.java
@@ -70,6 +70,7 @@ public class PostgresTypeUtils {
private static final String PG_CHARACTER_VARYING_ARRAY = "_varchar";
private static final String PG_JSON = "json";
private static final String PG_ENUM = "enum";
+ private static final String PG_UUID = "uuid";
public static DataType toDataType(
String typeName,
@@ -159,6 +160,7 @@ public class PostgresTypeUtils {
case PG_TEXT:
case PG_JSON:
case PG_ENUM:
+ case PG_UUID:
return DataTypes.STRING();
case PG_TEXT_ARRAY:
return DataTypes.ARRAY(DataTypes.STRING());
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableActionITCase.java
index 8e0efd110b..87291d2211 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableActionITCase.java
@@ -365,6 +365,7 @@ public class PostgresSyncTableActionITCase extends
PostgresActionITCaseBase {
DataTypes.STRING(), // _text
DataTypes.BYTES(), // _bin
DataTypes.STRING(), // _json
+ DataTypes.STRING(), // _uuid
DataTypes.ARRAY(DataTypes.STRING()) // _array
},
new String[] {
@@ -398,6 +399,7 @@ public class PostgresSyncTableActionITCase extends
PostgresActionITCaseBase {
"_text",
"_bin",
"_json",
+ "_uuid",
"_array",
});
FileStoreTable table = getFileStoreTable();
@@ -423,6 +425,7 @@ public class PostgresSyncTableActionITCase extends
PostgresActionITCaseBase {
+ "Paimon , Apache Paimon, Apache Paimon
PostgreSQL Test Data, "
+ "[98, 121, 116, 101, 115], "
+ "{\"a\": \"b\"}, "
+ + "123e4567-e89b-12d3-a456-426655440000, "
+ "[item1, item2]"
+ "]",
"+I["
@@ -444,6 +447,7 @@ public class PostgresSyncTableActionITCase extends
PostgresActionITCaseBase {
+ "NULL, NULL, "
+ "NULL, "
+ "NULL, "
+ + "NULL, "
+ "NULL"
+ "]");
waitForResult(expected, table, rowType, Arrays.asList("pt", "_id"));
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresTypeUtilsTest.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresTypeUtilsTest.java
index f24769e9f6..794e26df6e 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresTypeUtilsTest.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresTypeUtilsTest.java
@@ -83,4 +83,14 @@ public class PostgresTypeUtilsTest {
assertThat(PostgresTypeUtils.toDataType("_numeric", 0, 0, EMPTY))
.isEqualTo(DataTypes.ARRAY(DataTypes.DECIMAL(DecimalType.MAX_PRECISION, 18)));
}
+
+ /**
+ * Debezium emits {@code uuid} values as plain strings ({@code
io.debezium.data.Uuid}), so the
+ * type maps to {@code STRING} preserving the canonical text
representation.
+ */
+ @Test
+ public void testUuidMapsToString() {
+ assertThat(PostgresTypeUtils.toDataType("uuid", null, null, EMPTY))
+ .isEqualTo(DataTypes.STRING());
+ }
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/resources/postgres/sync_table_setup.sql
b/paimon-flink/paimon-flink-cdc/src/test/resources/postgres/sync_table_setup.sql
index 373eb3880f..ab2ec7c2f1 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/resources/postgres/sync_table_setup.sql
+++
b/paimon-flink/paimon-flink-cdc/src/test/resources/postgres/sync_table_setup.sql
@@ -105,6 +105,8 @@ CREATE TABLE all_types_table (
_bin BYTEA,
-- json
_json JSON,
+ -- UUID
+ _uuid UUID,
_array VARCHAR[],
PRIMARY KEY (_id)
);
@@ -129,7 +131,7 @@ INSERT INTO all_types_table (
_time, _time0,
_char, _varchar, _text,
_bin,
- _json,
+ _json, _uuid,
_array
) VALUES (
1, 1.1,
@@ -148,7 +150,7 @@ INSERT INTO all_types_table (
'10:13:23'::TIME, '10:13:23'::TIME,
'Paimon', 'Apache Paimon', 'Apache Paimon PostgreSQL Test Data',
'bytes',
- '{"a": "b"}'::JSON,
+ '{"a": "b"}'::JSON, '123e4567-e89b-12d3-a456-426655440000'::UUID,
ARRAY['item1', 'item2']::VARCHAR[]
), (
2, 2.2,
@@ -167,7 +169,7 @@ INSERT INTO all_types_table (
NULL, NULL,
NULL, NULL, NULL,
NULL,
- NULL,
+ NULL, NULL,
NULL
);
