Skip to content

[cdc] Support PostgreSQL jsonb type in postgres sync #9020

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 1 commit into from
Aug 7, 2026
Merged

[cdc] Support PostgreSQL jsonb type in postgres sync #9020

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions ...link-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresTypeUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";

Expand Down Expand Up @@ -159,6 +160,7 @@ public static DataType toDataType(
return DataTypes.ARRAY(DataTypes.VARCHAR(precision));
case PG_TEXT:
case PG_JSON:
case PG_JSONB:
case PG_ENUM:
case PG_UUID:
return DataTypes.STRING();
Expand Down
4 changes: 4 additions & 0 deletions .../test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableActionITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -365,6 +365,7 @@ private void testAllTypesImpl() throws Exception {
DataTypes.STRING(), // _text
DataTypes.BYTES(), // _bin
DataTypes.STRING(), // _json
DataTypes.STRING(), // _jsonb
DataTypes.STRING(), // _uuid
DataTypes.ARRAY(DataTypes.STRING()) // _array
},
Expand Down Expand Up @@ -399,6 +400,7 @@ private void testAllTypesImpl() throws Exception {
"_text",
"_bin",
"_json",
"_jsonb",
"_uuid",
"_array",
});
Expand All @@ -425,6 +427,7 @@ private void testAllTypesImpl() throws Exception {
+ "Paimon , Apache Paimon, Apache Paimon PostgreSQL Test Data, "
+ "[98, 121, 116, 101, 115], "
+ "{\"a\": \"b\"}, "
+ "{\"c\": \"d\"}, "
+ "123e4567-e89b-12d3-a456-426655440000, "
+ "[item1, item2]"
+ "]",
Expand All @@ -448,6 +451,7 @@ private void testAllTypesImpl() throws Exception {
+ "NULL, "
+ "NULL, "
+ "NULL, "
+ "NULL, "
+ "NULL"
+ "]");
waitForResult(expected, table, rowType, Arrays.asList("pt", "_id"));
Expand Down
6 changes: 6 additions & 0 deletions ...-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresTypeUtilsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,4 +93,10 @@ public void testUuidMapsToString() {
assertThat(PostgresTypeUtils.toDataType("uuid", null, null, EMPTY))
.isEqualTo(DataTypes.STRING());
}

@Test
public void testJsonbMapsToString() {
assertThat(PostgresTypeUtils.toDataType("jsonb", null, null, EMPTY))
.isEqualTo(DataTypes.STRING());
}
}
7 changes: 4 additions & 3 deletions paimon-flink/paimon-flink-cdc/src/test/resources/postgres/sync_table_setup.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,7 @@ CREATE TABLE all_types_table (
_bin BYTEA,
-- json
_json JSON,
_jsonb JSONB,
-- UUID
_uuid UUID,
_array VARCHAR[],
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -169,7 +170,7 @@ INSERT INTO all_types_table (
NULL, NULL,
NULL, NULL, NULL,
NULL,
NULL, NULL,
NULL, NULL, NULL,
NULL
);

Expand Down
Loading