This is an automated email from the ASF dual-hosted git repository.
yuzelin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 59e396c43 [flink][kafka-cdc] Minor refactor kafka cdc record parser
(#1885)
59e396c43 is described below
commit 59e396c43dfa91e852d67d4103bb4c3eac1aef74
Author: yuzelin <[email protected]>
AuthorDate: Fri Aug 25 14:46:38 2023 +0800
[flink][kafka-cdc] Minor refactor kafka cdc record parser (#1885)
---
.../cdc/kafka/formats/canal/CanalRecordParser.java | 41 ++++++----------------
.../cdc/kafka/formats/ogg/OggRecordParser.java | 5 ++-
2 files changed, 13 insertions(+), 33 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/canal/CanalRecordParser.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/canal/CanalRecordParser.java
index 85cb07855..17321214e 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/canal/CanalRecordParser.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/canal/CanalRecordParser.java
@@ -36,6 +36,7 @@ import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.Arra
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
@@ -61,7 +62,6 @@ import static
org.apache.paimon.utils.Preconditions.checkNotNull;
public class CanalRecordParser extends RecordParser {
private static final String FIELD_IS_DDL = "isDdl";
- private static final String FIELD_SQL = "sql";
private static final String FIELD_MYSQL_TYPE = "mysqlType";
private static final String FIELD_PRIMARY_KEYS = "pkNames";
private static final String FIELD_TYPE = "type";
@@ -83,10 +83,8 @@ public class CanalRecordParser extends RecordParser {
@Override
public List<RichCdcMultiplexRecord> extractRecords() {
- List<RichCdcMultiplexRecord> records = new ArrayList<>();
-
if (isDdl()) {
- return records;
+ return Collections.emptyList();
}
List<String> primaryKeys = extractPrimaryKeys();
@@ -99,24 +97,13 @@ public class CanalRecordParser extends RecordParser {
switch (type) {
case OP_UPDATE:
- return handleUpdateOperation(
- data, mySqlFieldTypes, paimonFieldTypes, primaryKeys,
records);
+ return handleUpdateOperation(data, mySqlFieldTypes,
paimonFieldTypes, primaryKeys);
case OP_INSERT:
return handleDataOperation(
- data,
- mySqlFieldTypes,
- paimonFieldTypes,
- primaryKeys,
- records,
- RowKind.INSERT);
+ data, mySqlFieldTypes, paimonFieldTypes, primaryKeys,
RowKind.INSERT);
case OP_DELETE:
return handleDataOperation(
- data,
- mySqlFieldTypes,
- paimonFieldTypes,
- primaryKeys,
- records,
- RowKind.DELETE);
+ data, mySqlFieldTypes, paimonFieldTypes, primaryKeys,
RowKind.DELETE);
default:
throw new UnsupportedOperationException("Unknown record type:
" + type);
}
@@ -157,9 +144,7 @@ public class CanalRecordParser extends RecordParser {
checkNotNull(root.get(FIELD_DATA), errorMessageTemplate, FIELD_DATA);
checkNotNull(root.get(FIELD_IS_DDL), errorMessageTemplate,
FIELD_IS_DDL);
- if (isDdl()) {
- checkNotNull(root.get(FIELD_SQL), errorMessageTemplate, FIELD_SQL);
- } else {
+ if (!isDdl()) {
checkNotNull(root.get(FIELD_MYSQL_TYPE), errorMessageTemplate,
FIELD_MYSQL_TYPE);
checkNotNull(root.get(FIELD_PRIMARY_KEYS), errorMessageTemplate,
FIELD_PRIMARY_KEYS);
}
@@ -171,8 +156,7 @@ public class CanalRecordParser extends RecordParser {
}
private boolean isDdl() {
- JsonNode isDdlNode = root.get(FIELD_IS_DDL);
- return isDdlNode.asBoolean();
+ return root.get(FIELD_IS_DDL).asBoolean();
}
private List<String> extractPrimaryKeys() {
@@ -217,10 +201,7 @@ public class CanalRecordParser extends RecordParser {
Map<String, String> resultMap =
mySqlFieldTypes.entrySet().stream()
- .filter(
- entry ->
- jsonMap.containsKey(entry.getKey())
- && jsonMap.get(entry.getKey())
!= null)
+ .filter(entry -> jsonMap.get(entry.getKey()) != null)
.collect(
Collectors.toMap(
Map.Entry::getKey,
@@ -270,8 +251,8 @@ public class CanalRecordParser extends RecordParser {
Map<String, String> mySqlFieldTypes,
LinkedHashMap<String, DataType> paimonFieldTypes,
List<String> primaryKeys,
- List<RichCdcMultiplexRecord> records,
RowKind kind) {
+ List<RichCdcMultiplexRecord> records = new ArrayList<>();
for (JsonNode datum : data) {
Map<String, String> rowData =
extractRowFromJson(datum, mySqlFieldTypes,
paimonFieldTypes);
@@ -292,8 +273,8 @@ public class CanalRecordParser extends RecordParser {
ArrayNode data,
Map<String, String> mySqlFieldTypes,
LinkedHashMap<String, DataType> paimonFieldTypes,
- List<String> primaryKeys,
- List<RichCdcMultiplexRecord> records) {
+ List<String> primaryKeys) {
+ List<RichCdcMultiplexRecord> records = new ArrayList<>();
ArrayNode old = JsonSerdeUtil.getNodeAs(root, FIELD_OLD,
ArrayNode.class);
for (int i = 0; i < data.size(); i++) {
Map<String, String> after =
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/ogg/OggRecordParser.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/ogg/OggRecordParser.java
index 19d0a1e70..9e7b67631 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/ogg/OggRecordParser.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/ogg/OggRecordParser.java
@@ -27,7 +27,6 @@ import
org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowKind;
-import org.apache.paimon.utils.JsonSerdeUtil;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.type.TypeReference;
@@ -164,7 +163,7 @@ public class OggRecordParser extends RecordParser {
}
private void extractDatabaseAndTableNames() {
- JsonNode tableNode = JsonSerdeUtil.getNodeAs(root, FIELD_TABLE,
JsonNode.class);
+ JsonNode tableNode = root.get(FIELD_TABLE);
if (tableNode != null) {
String[] dbt = tableNode.asText().split("\\.", 2); // Limit split
to 2 parts
if (dbt.length == 2) {
@@ -176,7 +175,7 @@ public class OggRecordParser extends RecordParser {
private LinkedHashMap<String, DataType> extractFieldTypesFromOracleType() {
LinkedHashMap<String, DataType> fieldTypes = new LinkedHashMap<>();
- JsonNode record = JsonSerdeUtil.getNodeAs(root, FIELD_AFTER,
JsonNode.class);
+ JsonNode record = root.get(FIELD_AFTER);
Map<String, Object> linkedHashMap =
OBJECT_MAPPER.convertValue(
record, new TypeReference<LinkedHashMap<String,
Object>>() {});
