MonsterChenzhuo commented on code in PR #1880:
URL: https://github.com/apache/incubator-paimon/pull/1880#discussion_r1304222486
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/canal/CanalRecordParser.java:
##########
@@ -27,27 +27,42 @@
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.Charset;
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 Charset CHARSET_ISO_8859_1 =
StandardCharsets.ISO_8859_1;
+ private static final String IS_DDL_FIELD = "isDdl";
private static final String FIELD_SQL = "sql";
Review Comment:
It is used in ValidateFormat,
if (isDdl()) { // Call the isDdl() method first, and there will be
a validity of the ddl field
// Verify the existence of sql fields
checkNotNull(root.get(FIELD_SQL), errorMessageTemplate,
FIELD_SQL);
} else {
checkNotNull(root.get(FIELD_MYSQL_TYPE), errorMessageTemplate,
FIELD_MYSQL_TYPE);
checkNotNull(root.get(FIELD_PRIMARY_KEYS), errorMessageTemplate,
FIELD_PRIMARY_KEYS);
}
--
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]
