MonsterChenzhuo commented on code in PR #1880:
URL: https://github.com/apache/incubator-paimon/pull/1880#discussion_r1304236017
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/canal/CanalRecordParser.java:
##########
@@ -199,69 +179,67 @@ protected String extractString(String key) {
}
private boolean isDdl() {
- return root.get("isDdl") != null && root.get("isDdl").asBoolean();
+ JsonNode isDdlNode = root.get(IS_DDL_FIELD);
+ if (isDdlNode == null) {
+ throw new IllegalArgumentException("Expected 'isDdl' field in the
JSON but not found.");
+ }
+ return isDdlNode.asBoolean();
}
private List<String> extractPrimaryKeys() {
- List<String> primaryKeys = new ArrayList<>();
- ArrayNode pkNames = (ArrayNode) root.get(FIELD_PRIMARY_KEYS);
- pkNames.iterator().forEachRemaining(pk ->
primaryKeys.add(toFieldName(pk.asText())));
- return primaryKeys;
+ ArrayNode pkNames = JsonSerdeUtil.getNodeAs(root, FIELD_PRIMARY_KEYS,
ArrayNode.class);
+ return StreamSupport.stream(pkNames.spliterator(), false)
+ .map(pk -> toFieldName(pk.asText()))
+ .collect(Collectors.toList());
}
private LinkedHashMap<String, String> extractFieldTypesFromMySqlType() {
+ JsonNode schema = JsonSerdeUtil.getNodeAs(root, FIELD_MYSQL_TYPE,
JsonNode.class);
LinkedHashMap<String, String> fieldTypes = new LinkedHashMap<>();
- JsonNode schema = root.get(FIELD_MYSQL_TYPE);
- Iterator<String> iterator = schema.fieldNames();
- while (iterator.hasNext()) {
- String fieldName = iterator.next();
- String fieldType = schema.get(fieldName).asText();
- fieldTypes.put(toFieldName(fieldName), fieldType);
- }
+ schema.fieldNames()
+ .forEachRemaining(
+ fieldName -> {
+ String fieldType = schema.get(fieldName).asText();
+ fieldTypes.put(toFieldName(fieldName), fieldType);
+ });
return fieldTypes;
}
- private Map<String, String> extractRow(
+ /**
+ * Extracts data from a given JSON node and transforms it based on
provided MySQL and Paimon
+ * field types.
+ *
+ * @param record The JSON node containing the data.
+ * @param mySqlFieldTypes A map of MySQL field types.
+ * @param paimonFieldTypes A map of Paimon field types.
+ * @return A map of extracted and transformed data.
+ */
+ private Map<String, String> extractRowFromJson(
JsonNode record,
Map<String, String> mySqlFieldTypes,
LinkedHashMap<String, DataType> paimonFieldTypes) {
+
Map<String, Object> jsonMap =
OBJECT_MAPPER.convertValue(record, new
TypeReference<Map<String, Object>>() {});
if (jsonMap == null) {
return new HashMap<>();
}
- Map<String, String> resultMap = new HashMap<>();
- for (Map.Entry<String, String> field : mySqlFieldTypes.entrySet()) {
- String fieldName = field.getKey();
- String mySqlType = field.getValue();
- Object objectValue = jsonMap.get(fieldName);
- if (objectValue == null) {
- continue;
- }
-
- String oldValue = objectValue.toString();
- String newValue = oldValue;
-
- if
(MySqlTypeUtils.isSetType(MySqlTypeUtils.getShortType(mySqlType))) {
- newValue = CanalFieldParser.convertSet(newValue, mySqlType);
- } else if
(MySqlTypeUtils.isEnumType(MySqlTypeUtils.getShortType(mySqlType))) {
- newValue = CanalFieldParser.convertEnum(newValue, mySqlType);
- } else if
(MySqlTypeUtils.isGeoType(MySqlTypeUtils.getShortType(mySqlType))) {
- try {
- byte[] wkb =
- CanalFieldParser.convertGeoType2WkbArray(
-
oldValue.getBytes(StandardCharsets.ISO_8859_1));
- newValue = MySqlTypeUtils.convertWkbArray(wkb);
- } catch (Exception e) {
- throw new IllegalArgumentException(
- String.format("Failed to convert %s to geometry
JSON.", oldValue), e);
- }
- }
- resultMap.put(fieldName, newValue);
- }
+ Map<String, String> resultMap =
+ mySqlFieldTypes.entrySet().stream()
+ .filter(
+ entry ->
+ jsonMap.containsKey(entry.getKey())
+ && jsonMap.get(entry.getKey())
!= null)
Review Comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
