diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/JsonSerdeUtil.java b/paimon-core/src/main/java/org/apache/paimon/utils/JsonSerdeUtil.java index c03bbcae44fbd..d2528c543aa0b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/JsonSerdeUtil.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/JsonSerdeUtil.java @@ -35,7 +35,6 @@ import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.SerializerProvider; import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.deser.std.StdDeserializer; import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.module.SimpleModule; -import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ArrayNode; import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ObjectNode; import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ser.std.StdSerializer; @@ -44,7 +43,6 @@ import java.io.IOException; import java.io.UncheckedIOException; import java.util.LinkedHashMap; -import java.util.List; import java.util.Map; import java.util.stream.Collectors; @@ -131,13 +129,10 @@ public static T fromJson(String json, TypeReference typeReference) { } } - public static ObjectNode appendArrayNode( - ObjectNode node, String fieldName, List elememts) { - ArrayNode arrayNode = OBJECT_MAPPER_INSTANCE.createArrayNode(); - for (String element : elememts) { - arrayNode.add(element); - } - node.set(fieldName, arrayNode); + public static ObjectNode setNode(ObjectNode node, String fieldName, T value) { + ObjectMapper mapper = new ObjectMapper(); + JsonNode nodeValue = OBJECT_MAPPER_INSTANCE.valueToTree(value); + node.set(fieldName, nodeValue); return node; } diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java index 89c2295bc8808..b07b1fa906aa5 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java @@ -84,7 +84,7 @@ public static String extractPrimaryKeys(Pair record) { ObjectNode valueNode = JsonSerdeUtil.fromJson(record.getValue(), ObjectNode.class); // append primary keys - JsonSerdeUtil.appendArrayNode(valueNode, FIELD_PRIMARY, primaryKeys); + JsonSerdeUtil.setNode(valueNode, FIELD_PRIMARY, primaryKeys); return JsonSerdeUtil.writeValueAsString(valueNode); } catch (JsonProcessingException e) { throw new RuntimeException(