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 41ba5dd982 [cdc] Support PostgreSQL jsonb type in postgres sync (#9020)
41ba5dd982 is described below
commit 41ba5dd9825cd38e8351340607fc99b447e2096d
Author: Arnav Balyan <[email protected]>
AuthorDate: Fri Aug 7 10:49:38 2026 +0530
[cdc] Support PostgreSQL jsonb type in postgres sync (#9020)
---
.../apache/paimon/flink/action/cdc/postgres/PostgresTypeUtils.java | 2 ++
.../flink/action/cdc/postgres/PostgresSyncTableActionITCase.java | 4 ++++
.../paimon/flink/action/cdc/postgres/PostgresTypeUtilsTest.java | 6 ++++++
.../src/test/resources/postgres/sync_table_setup.sql | 7 ++++---
4 files changed, 16 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 78b80c458f..7fae4c4327 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
@@ -69,6 +69,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_JSON = "json";
+ private static final String PG_JSONB = "jsonb";
private static final String PG_ENUM = "enum";
private static final String PG_UUID = "uuid";
@@ -159,6 +160,7 @@ public class PostgresTypeUtils {
return DataTypes.ARRAY(DataTypes.VARCHAR(precision));
case PG_TEXT:
case PG_JSON:
+ case PG_JSONB:
case PG_ENUM:
case PG_UUID:
return 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 87291d2211..107f3cff19 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(), // _jsonb
DataTypes.STRING(), // _uuid
DataTypes.ARRAY(DataTypes.STRING()) // _array
},
@@ -399,6 +400,7 @@ public class PostgresSyncTableActionITCase extends
PostgresActionITCaseBase {
"_text",
"_bin",
"_json",
+ "_jsonb",
"_uuid",
"_array",
});
@@ -425,6 +427,7 @@ public class PostgresSyncTableActionITCase extends
PostgresActionITCaseBase {
+ "Paimon , Apache Paimon, Apache Paimon
PostgreSQL Test Data, "
+ "[98, 121, 116, 101, 115], "
+ "{\"a\": \"b\"}, "
+ + "{\"c\": \"d\"}, "
+ "123e4567-e89b-12d3-a456-426655440000, "
+ "[item1, item2]"
+ "]",
@@ -448,6 +451,7 @@ public class PostgresSyncTableActionITCase extends
PostgresActionITCaseBase {
+ "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 794e26df6e..30b5f79803 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
@@ -93,4 +93,10 @@ public class PostgresTypeUtilsTest {
assertThat(PostgresTypeUtils.toDataType("uuid", null, null, EMPTY))
.isEqualTo(DataTypes.STRING());
}
+
+ @Test
+ public void testJsonbMapsToString() {
+ assertThat(PostgresTypeUtils.toDataType("jsonb", 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 ab2ec7c2f1..bb0e7ca30d 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,7 @@ CREATE TABLE all_types_table (
_bin BYTEA,
-- json
_json JSON,
+ _jsonb JSONB,
-- UUID
_uuid UUID,
_array VARCHAR[],
@@ -131,7 +132,7 @@ INSERT INTO all_types_table (
_time, _time0,
_char, _varchar, _text,
_bin,
- _json, _uuid,
+ _json, _jsonb, _uuid,
_array
) VALUES (
1, 1.1,
@@ -150,7 +151,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, '123e4567-e89b-12d3-a456-426655440000'::UUID,
+ '{"a": "b"}'::JSON, '{"c": "d"}'::JSONB,
'123e4567-e89b-12d3-a456-426655440000'::UUID,
ARRAY['item1', 'item2']::VARCHAR[]
), (
2, 2.2,
@@ -169,7 +170,7 @@ INSERT INTO all_types_table (
NULL, NULL,
NULL, NULL, NULL,
NULL,
- NULL, NULL,
+ NULL, NULL, NULL,
NULL
);
