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 225589839 [flink] Optimization of Canal parsing logic (#1880)
225589839 is described below
commit 22558983989c34f759a6b689c4b218c912911618
Author: monster <[email protected]>
AuthorDate: Fri Aug 25 10:11:17 2023 +0800
[flink] Optimization of Canal parsing logic (#1880)
---
.../org/apache/paimon/utils/JsonSerdeUtil.java | 27 ++
.../action/cdc/kafka/formats/RecordParser.java | 4 +
.../cdc/kafka/formats/canal/CanalRecordParser.java | 288 +++++++++++++--------
.../cdc/kafka/formats/ogg/OggRecordParser.java | 8 +-
.../cdc/kafka/KafkaCanalSyncTableActionITCase.java | 2 +-
.../canal}/table/computedcolumn/canal-data-1.txt | 0
6 files changed, 213 insertions(+), 116 deletions(-)
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 70b87f817..0ffd1fb32 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
@@ -28,6 +28,7 @@ import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonGenerator
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonParser;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.DeserializationContext;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.Module;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
@@ -37,6 +38,7 @@ import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ser.std.S
import java.io.IOException;
import java.io.UncheckedIOException;
+import java.util.Optional;
/** A utility class that provide abilities for JSON serialization and
deserialization. */
public class JsonSerdeUtil {
@@ -52,6 +54,31 @@ public class JsonSerdeUtil {
OBJECT_MAPPER_INSTANCE.registerModule(createPaimonJacksonModule());
}
+ /**
+ * Retrieves a specific node from the given root node and casts it to the
specified type.
+ *
+ * @param <T> The type of the node to be returned.
+ * @param root The root node from which the specific node is to be
retrieved.
+ * @param fieldName The name of the field to retrieve.
+ * @param clazz The class of the node to be returned.
+ * @return The node casted to the specified type.
+ * @throws IllegalArgumentException if the node is not present or if it's
not of the expected
+ * type.
+ */
+ public static <T extends JsonNode> T getNodeAs(
+ JsonNode root, String fieldName, Class<T> clazz) {
+ return Optional.ofNullable(root)
+ .map(r -> r.get(fieldName))
+ .filter(clazz::isInstance)
+ .map(clazz::cast)
+ .orElseThrow(
+ () ->
+ new IllegalArgumentException(
+ String.format(
+ "Expected node '%s' to be of
type %s but was either not found or of a different type.",
+ fieldName, clazz.getName())));
+ }
+
public static <T> T fromJson(String json, Class<T> clazz) {
try {
return OBJECT_MAPPER_INSTANCE.reader().readValue(json, clazz);
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/RecordParser.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/RecordParser.java
index cf050853b..23de278b3 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/RecordParser.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/RecordParser.java
@@ -102,4 +102,8 @@ public abstract class RecordParser implements
FlatMapFunction<String, RichCdcMul
protected String toFieldName(String rawName) {
return StringUtils.caseSensitiveConversion(rawName, caseSensitive);
}
+
+ protected Map<String, String> adjustCase(Map<String, String> map) {
+ return caseSensitive ? map : keyCaseInsensitive(map);
+ }
}
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 3b7327838..85cb07855 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
@@ -27,27 +27,40 @@ import org.apache.paimon.flink.sink.cdc.CdcRecord;
import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
import org.apache.paimon.types.DataType;
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;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ArrayNode;
-import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.NullNode;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
-import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
-/** Convert canal-json format string to list of {@link
RichCdcMultiplexRecord}s. */
+/**
+ * The {@code CanalRecordParser} class is responsible for parsing records from
the Canal-JSON
+ * format. Canal is a database binlog multi-platform consumer, which is used
to synchronize data
+ * across databases. This parser extracts relevant information from the
Canal-JSON format and
+ * transforms it into a list of {@link RichCdcMultiplexRecord} objects, which
represent the changes
+ * captured in the database.
+ *
+ * <p>The class handles different types of database operations such as INSERT,
UPDATE, and DELETE,
+ * and generates corresponding {@link RichCdcMultiplexRecord} objects for each
operation.
+ *
+ * <p>Additionally, the parser supports schema extraction, which can be used
to understand the
+ * structure of the incoming data and its corresponding field types.
+ */
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";
@@ -70,83 +83,43 @@ public class CanalRecordParser extends RecordParser {
@Override
public List<RichCdcMultiplexRecord> extractRecords() {
+ List<RichCdcMultiplexRecord> records = new ArrayList<>();
+
if (isDdl()) {
- return Collections.emptyList();
+ return records;
}
List<String> primaryKeys = extractPrimaryKeys();
-
- // extract field types
LinkedHashMap<String, String> mySqlFieldTypes =
extractFieldTypesFromMySqlType();
- LinkedHashMap<String, DataType> paimonFieldTypes = new
LinkedHashMap<>();
- mySqlFieldTypes.forEach(
- (name, type) -> paimonFieldTypes.put(name,
MySqlTypeUtils.toDataType(type)));
+ LinkedHashMap<String, DataType> paimonFieldTypes =
+ convertToPaimonFieldTypes(mySqlFieldTypes);
- // extract row kind and field values
- List<RichCdcMultiplexRecord> records = new ArrayList<>();
String type = extractString(FIELD_TYPE);
- ArrayNode data = (ArrayNode) root.get(FIELD_DATA);
+ ArrayNode data = JsonSerdeUtil.getNodeAs(root, FIELD_DATA,
ArrayNode.class);
+
switch (type) {
case OP_UPDATE:
- ArrayNode old =
- root.get(FIELD_OLD) instanceof NullNode
- ? null
- : (ArrayNode) root.get(FIELD_OLD);
- for (int i = 0; i < data.size(); i++) {
- Map<String, String> after =
- extractRow(data.get(i), mySqlFieldTypes,
paimonFieldTypes);
- if (old != null) {
- Map<String, String> before =
- extractRow(old.get(i), mySqlFieldTypes,
paimonFieldTypes);
- // fields in "old" (before) means the fields are
changed
- // fields not in "old" (before) means the fields are
not changed,
- // so we just copy the not changed fields into before
- for (Map.Entry<String, String> entry :
after.entrySet()) {
- if (!before.containsKey(entry.getKey())) {
- before.put(entry.getKey(), entry.getValue());
- }
- }
- before = caseSensitive ? before :
keyCaseInsensitive(before);
- records.add(
- new RichCdcMultiplexRecord(
- databaseName,
- tableName,
- paimonFieldTypes,
- primaryKeys,
- new CdcRecord(RowKind.DELETE,
before)));
- }
- after = caseSensitive ? after : keyCaseInsensitive(after);
- records.add(
- new RichCdcMultiplexRecord(
- databaseName,
- tableName,
- paimonFieldTypes,
- primaryKeys,
- new CdcRecord(RowKind.INSERT, after)));
- }
- break;
+ return handleUpdateOperation(
+ data, mySqlFieldTypes, paimonFieldTypes, primaryKeys,
records);
case OP_INSERT:
- // fall through
+ return handleDataOperation(
+ data,
+ mySqlFieldTypes,
+ paimonFieldTypes,
+ primaryKeys,
+ records,
+ RowKind.INSERT);
case OP_DELETE:
- for (JsonNode datum : data) {
- Map<String, String> after =
- extractRow(datum, mySqlFieldTypes,
paimonFieldTypes);
- after = caseSensitive ? after : keyCaseInsensitive(after);
- RowKind kind = type.equals(OP_INSERT) ? RowKind.INSERT :
RowKind.DELETE;
- records.add(
- new RichCdcMultiplexRecord(
- databaseName,
- tableName,
- paimonFieldTypes,
- primaryKeys,
- new CdcRecord(kind, after)));
- }
- break;
+ return handleDataOperation(
+ data,
+ mySqlFieldTypes,
+ paimonFieldTypes,
+ primaryKeys,
+ records,
+ RowKind.DELETE);
default:
throw new UnsupportedOperationException("Unknown record type:
" + type);
}
-
- return records;
}
@Override
@@ -157,11 +130,9 @@ public class CanalRecordParser extends RecordParser {
throw new RuntimeException(e);
}
validateFormat();
-
if (isDdl()) {
return null;
}
-
LinkedHashMap<String, String> mySqlFieldTypes =
extractFieldTypesFromMySqlType();
LinkedHashMap<String, DataType> paimonFieldTypes = new
LinkedHashMap<>();
mySqlFieldTypes.forEach(
@@ -184,6 +155,7 @@ public class CanalRecordParser extends RecordParser {
checkNotNull(root.get(FIELD_TABLE), errorMessageTemplate, FIELD_TABLE);
checkNotNull(root.get(FIELD_TYPE), errorMessageTemplate, FIELD_TYPE);
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);
@@ -199,31 +171,41 @@ public class CanalRecordParser extends RecordParser {
}
private boolean isDdl() {
- return root.get("isDdl") != null && root.get("isDdl").asBoolean();
+ JsonNode isDdlNode = root.get(FIELD_IS_DDL);
+ 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 = root.get(FIELD_MYSQL_TYPE);
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) {
@@ -233,35 +215,19 @@ public class CanalRecordParser extends RecordParser {
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)
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ entry ->
+ transformValue(
+
jsonMap.get(entry.getKey()).toString(),
+ entry.getValue())));
// generate values for computed columns
for (ComputedColumn computedColumn : computedColumns) {
@@ -273,4 +239,104 @@ public class CanalRecordParser extends RecordParser {
return resultMap;
}
+
+ private String transformValue(String oldValue, String mySqlType) {
+ String shortType = MySqlTypeUtils.getShortType(mySqlType);
+
+ if (MySqlTypeUtils.isSetType(shortType)) {
+ return CanalFieldParser.convertSet(oldValue, mySqlType);
+ }
+
+ if (MySqlTypeUtils.isEnumType(shortType)) {
+ return CanalFieldParser.convertEnum(oldValue, mySqlType);
+ }
+
+ if (MySqlTypeUtils.isGeoType(shortType)) {
+ try {
+ byte[] wkb =
+ CanalFieldParser.convertGeoType2WkbArray(
+
oldValue.getBytes(StandardCharsets.ISO_8859_1));
+ return MySqlTypeUtils.convertWkbArray(wkb);
+ } catch (Exception e) {
+ throw new IllegalArgumentException(
+ String.format("Failed to convert %s to geometry
JSON.", oldValue), e);
+ }
+ }
+ return oldValue;
+ }
+
+ private List<RichCdcMultiplexRecord> handleDataOperation(
+ ArrayNode data,
+ Map<String, String> mySqlFieldTypes,
+ LinkedHashMap<String, DataType> paimonFieldTypes,
+ List<String> primaryKeys,
+ List<RichCdcMultiplexRecord> records,
+ RowKind kind) {
+ for (JsonNode datum : data) {
+ Map<String, String> rowData =
+ extractRowFromJson(datum, mySqlFieldTypes,
paimonFieldTypes);
+ rowData = adjustCase(rowData);
+
+ records.add(
+ new RichCdcMultiplexRecord(
+ databaseName,
+ tableName,
+ paimonFieldTypes,
+ primaryKeys,
+ new CdcRecord(kind, rowData)));
+ }
+ return records;
+ }
+
+ private List<RichCdcMultiplexRecord> handleUpdateOperation(
+ ArrayNode data,
+ Map<String, String> mySqlFieldTypes,
+ LinkedHashMap<String, DataType> paimonFieldTypes,
+ List<String> primaryKeys,
+ List<RichCdcMultiplexRecord> records) {
+ ArrayNode old = JsonSerdeUtil.getNodeAs(root, FIELD_OLD,
ArrayNode.class);
+ for (int i = 0; i < data.size(); i++) {
+ Map<String, String> after =
+ extractRowFromJson(data.get(i), mySqlFieldTypes,
paimonFieldTypes);
+
+ if (old != null && i < old.size()) {
+ Map<String, String> before =
+ extractRowFromJson(old.get(i), mySqlFieldTypes,
paimonFieldTypes);
+
+ // Fields in "old" (before) means the fields are changed.
+ // Fields not in "old" (before) means the fields are not
changed,
+ // so we just copy the not changed fields into before.
+ for (Map.Entry<String, String> entry : after.entrySet()) {
+ before.putIfAbsent(entry.getKey(), entry.getValue());
+ }
+
+ before = adjustCase(before);
+ records.add(
+ new RichCdcMultiplexRecord(
+ databaseName,
+ tableName,
+ paimonFieldTypes,
+ primaryKeys,
+ new CdcRecord(RowKind.DELETE, before)));
+ }
+
+ after = adjustCase(after);
+ records.add(
+ new RichCdcMultiplexRecord(
+ databaseName,
+ tableName,
+ paimonFieldTypes,
+ primaryKeys,
+ new CdcRecord(RowKind.INSERT, after)));
+ }
+ return records;
+ }
+
+ private LinkedHashMap<String, DataType> convertToPaimonFieldTypes(
+ Map<String, String> mySqlFieldTypes) {
+ LinkedHashMap<String, DataType> paimonFieldTypes = new
LinkedHashMap<>();
+ mySqlFieldTypes.forEach(
+ (name, type) -> paimonFieldTypes.put(name,
MySqlTypeUtils.toDataType(type)));
+ return paimonFieldTypes;
+ }
}
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 9d529849c..19d0a1e70 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,6 +27,7 @@ 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;
@@ -93,7 +94,7 @@ public class OggRecordParser extends RecordParser {
JsonNode jsonNode, RowKind rowKind, List<RichCdcMultiplexRecord>
records) {
LinkedHashMap<String, DataType> paimonFieldTypes = new
LinkedHashMap<>();
Map<String, String> rowData = extractRow(jsonNode, paimonFieldTypes);
- rowData = caseSensitive ? rowData : keyCaseInsensitive(rowData);
+ rowData = adjustCase(rowData);
records.add(createRecord(rowKind, rowData, paimonFieldTypes));
}
@@ -163,7 +164,7 @@ public class OggRecordParser extends RecordParser {
}
private void extractDatabaseAndTableNames() {
- JsonNode tableNode = root.get(FIELD_TABLE);
+ JsonNode tableNode = JsonSerdeUtil.getNodeAs(root, FIELD_TABLE,
JsonNode.class);
if (tableNode != null) {
String[] dbt = tableNode.asText().split("\\.", 2); // Limit split
to 2 parts
if (dbt.length == 2) {
@@ -175,8 +176,7 @@ public class OggRecordParser extends RecordParser {
private LinkedHashMap<String, DataType> extractFieldTypesFromOracleType() {
LinkedHashMap<String, DataType> fieldTypes = new LinkedHashMap<>();
-
- JsonNode record = root.get(FIELD_AFTER);
+ JsonNode record = JsonSerdeUtil.getNodeAs(root, FIELD_AFTER,
JsonNode.class);
Map<String, Object> linkedHashMap =
OBJECT_MAPPER.convertValue(
record, new TypeReference<LinkedHashMap<String,
Object>>() {});
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java
index a30527261..c3b2b86dd 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java
@@ -1069,7 +1069,7 @@ public class KafkaCanalSyncTableActionITCase extends
KafkaActionITCaseBase {
String topic = "computed_column";
createTestTopic(topic, 1, 1);
- List<String> lines =
readLines("kafka.canal/table/computedcolumn/canal-data-1.txt");
+ List<String> lines =
readLines("kafka/canal/table/computedcolumn/canal-data-1.txt");
try {
writeRecordsToKafka(topic, lines);
} catch (Exception e) {
diff --git
a/paimon-flink/paimon-flink-common/src/test/resources/kafka.canal/table/computedcolumn/canal-data-1.txt
b/paimon-flink/paimon-flink-common/src/test/resources/kafka/canal/table/computedcolumn/canal-data-1.txt
similarity index 100%
rename from
paimon-flink/paimon-flink-common/src/test/resources/kafka.canal/table/computedcolumn/canal-data-1.txt
rename to
paimon-flink/paimon-flink-common/src/test/resources/kafka/canal/table/computedcolumn/canal-data-1.txt
