This is an automated email from the ASF dual-hosted git repository.
kerwin 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 b88938d91 [cdc] Refactor DebeziumRecordParser (#2790)
b88938d91 is described below
commit b88938d91c6f282b324f2a057d8529e035996bff
Author: yuzelin <[email protected]>
AuthorDate: Thu Jan 25 19:37:11 2024 +0800
[cdc] Refactor DebeziumRecordParser (#2790)
---
docs/content/cdc-ingestion/kafka-cdc.md | 8 +-
.../paimon/flink/action/cdc/format/DataFormat.java | 4 +-
.../cdc/format/debezium/DebeziumRecordParser.java | 100 +++++++++
.../DebeziumSchemaIncludeRecordParser.java | 100 ---------
.../cdc/format/debezium/DebeziumSchemaUtils.java | 231 +++++++++++++++++++
.../kafka/KafkaDebeziumSyncTableActionITCase.java | 9 +-
.../cdc/kafka/KafkaSyncTableActionITCase.java | 249 ++++++++++++++++++++-
.../table/schema/alltype/debezium-data-1.txt | 19 ++
8 files changed, 611 insertions(+), 109 deletions(-)
diff --git a/docs/content/cdc-ingestion/kafka-cdc.md
b/docs/content/cdc-ingestion/kafka-cdc.md
index 74e16bceb..dab2a676a 100644
--- a/docs/content/cdc-ingestion/kafka-cdc.md
+++ b/docs/content/cdc-ingestion/kafka-cdc.md
@@ -66,9 +66,11 @@ If a message in a Kafka topic is a change event captured
from another database u
The JSON sources possibly missing some information. For example, Ogg and
Maxwell format standards don't contain field
types; When you write JSON sources into Flink Kafka sink, it will only reserve
data and row type and drop other information.
The synchronization job will try best to handle the problem as follows:
-1. If missing field types, Paimon will use 'STRING' type as default.
-2. If missing database name or table name, you cannot do database
synchronization, but you can still do table synchronization.
-3. If missing primary keys, the job might create non primary key table. You
can set primary keys when submit job in table
+1. Usually, debezium-json contains 'schema' field, from which Paimon will
retrieve data types. Make sure your debezium
+json has this field, or Paimon will use 'STRING' type.
+2. If missing field types, Paimon will use 'STRING' type as default.
+3. If missing database name or table name, you cannot do database
synchronization, but you can still do table synchronization.
+4. If missing primary keys, the job might create non primary key table. You
can set primary keys when submit job in table
synchronization.
{{< /hint >}}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/DataFormat.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/DataFormat.java
index 5344190d4..28dc3e457 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/DataFormat.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/DataFormat.java
@@ -22,7 +22,6 @@ import org.apache.paimon.flink.action.cdc.ComputedColumn;
import org.apache.paimon.flink.action.cdc.TypeMapping;
import org.apache.paimon.flink.action.cdc.format.canal.CanalRecordParser;
import org.apache.paimon.flink.action.cdc.format.debezium.DebeziumRecordParser;
-import
org.apache.paimon.flink.action.cdc.format.debezium.DebeziumSchemaIncludeRecordParser;
import org.apache.paimon.flink.action.cdc.format.maxwell.MaxwellRecordParser;
import org.apache.paimon.flink.action.cdc.format.ogg.OggRecordParser;
@@ -39,8 +38,7 @@ public enum DataFormat {
CANAL_JSON(CanalRecordParser::new),
OGG_JSON(OggRecordParser::new),
MAXWELL_JSON(MaxwellRecordParser::new),
- DEBEZIUM_JSON(DebeziumRecordParser::new),
- DEBEZIUM_JSON_SCHEMA_INCLUDE(DebeziumSchemaIncludeRecordParser::new);
+ DEBEZIUM_JSON(DebeziumRecordParser::new);
// Add more data formats here if needed
private final RecordParserFactory parser;
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumRecordParser.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumRecordParser.java
index a736e5b49..d4d5b47a2 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumRecordParser.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumRecordParser.java
@@ -22,16 +22,26 @@ import org.apache.paimon.flink.action.cdc.ComputedColumn;
import org.apache.paimon.flink.action.cdc.TypeMapping;
import org.apache.paimon.flink.action.cdc.format.RecordParser;
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.utils.Preconditions;
+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 javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import static org.apache.paimon.utils.JsonSerdeUtil.getNodeAs;
import static org.apache.paimon.utils.JsonSerdeUtil.isNull;
/**
@@ -66,6 +76,11 @@ public class DebeziumRecordParser extends RecordParser {
private static final String OP_DELETE = "d";
private static final String OP_READE = "r";
+ private boolean hasSchema;
+ private final Map<String, String> debeziumTypes = new HashMap<>();
+ private final Map<String, String> classNames = new HashMap<>();
+ private final Map<String, Map<String, String>> parameters = new
HashMap<>();
+
public DebeziumRecordParser(
boolean caseSensitive, TypeMapping typeMapping,
List<ComputedColumn> computedColumns) {
super(caseSensitive, typeMapping, computedColumns);
@@ -105,13 +120,98 @@ public class DebeziumRecordParser extends RecordParser {
@Override
protected void setRoot(String record) {
JsonNode node = JsonSerdeUtil.fromJson(record, JsonNode.class);
+
+ hasSchema = false;
if (node.has(FIELD_SCHEMA)) {
root = node.get(FIELD_PAYLOAD);
+ JsonNode schema = node.get(FIELD_SCHEMA);
+ if (!isNull(schema)) {
+ parseSchema(schema);
+ hasSchema = true;
+ }
} else {
root = node;
}
}
+ private void parseSchema(JsonNode schema) {
+ debeziumTypes.clear();
+ classNames.clear();
+ parameters.clear();
+
+ ArrayNode schemaFields = getNodeAs(schema, "fields", ArrayNode.class);
+ Preconditions.checkNotNull(schemaFields);
+
+ ArrayNode fields = null;
+ for (int i = 0; i < schemaFields.size(); i++) {
+ JsonNode node = schemaFields.get(i);
+ if (getString(node, "field").equals("after")) {
+ fields = getNodeAs(node, "fields", ArrayNode.class);
+ break;
+ } else if (getString(node, "field").equals("before")) {
+ if (fields == null) {
+ fields = getNodeAs(node, "fields", ArrayNode.class);
+ }
+ }
+ }
+ Preconditions.checkNotNull(fields);
+
+ for (JsonNode node : fields) {
+ String field = getString(node, "field");
+
+ debeziumTypes.put(field, getString(node, "type"));
+ classNames.put(field, getString(node, "name"));
+
+ JsonNode parametersNode = node.get("parameters");
+ Map<String, String> parametersMap =
+ isNull(parametersNode)
+ ? Collections.emptyMap()
+ : JsonSerdeUtil.convertValue(
+ parametersNode,
+ new TypeReference<HashMap<String,
String>>() {});
+
+ parameters.put(field, parametersMap);
+ }
+ }
+
+ @Nullable
+ private String getString(JsonNode node, String fieldName) {
+ JsonNode fieldValue = node.get(fieldName);
+ return isNull(fieldValue) ? null : fieldValue.asText();
+ }
+
+ @Override
+ protected Map<String, String> extractRowData(
+ JsonNode record, LinkedHashMap<String, DataType> paimonFieldTypes)
{
+ if (!hasSchema) {
+ return super.extractRowData(record, paimonFieldTypes);
+ }
+
+ Map<String, Object> recordMap =
+ JsonSerdeUtil.convertValue(record, new
TypeReference<Map<String, Object>>() {});
+ LinkedHashMap<String, String> resultMap = new LinkedHashMap<>();
+ for (Map.Entry<String, Object> entry : recordMap.entrySet()) {
+ String fieldName = entry.getKey();
+ String rawValue = Objects.toString(entry.getValue(), null);
+ String debeziumType = debeziumTypes.get(fieldName);
+ String className = classNames.get(fieldName);
+
+ String transformed =
+ DebeziumSchemaUtils.transformRawValue(
+ rawValue, debeziumType, className, typeMapping,
record.get(fieldName));
+ resultMap.put(fieldName, transformed);
+
+ paimonFieldTypes.put(
+ fieldName,
+ DebeziumSchemaUtils.toDataType(
+ debeziumType, className,
parameters.get(fieldName)));
+ }
+
+ evalComputedColumns(resultMap, paimonFieldTypes);
+
+ return resultMap;
+ }
+
@Override
protected String primaryField() {
return FIELD_PRIMARY;
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaIncludeRecordParser.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaIncludeRecordParser.java
deleted file mode 100644
index 8c9b3402f..000000000
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaIncludeRecordParser.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.flink.action.cdc.format.debezium;
-
-import org.apache.paimon.flink.action.cdc.ComputedColumn;
-import org.apache.paimon.flink.action.cdc.TypeMapping;
-import org.apache.paimon.flink.action.cdc.format.RecordParser;
-import org.apache.paimon.types.DataType;
-import org.apache.paimon.types.DataTypes;
-import org.apache.paimon.utils.JsonSerdeUtil;
-
-import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
-
-import java.io.UncheckedIOException;
-import java.util.LinkedHashMap;
-import java.util.List;
-
-import static io.debezium.pipeline.signal.SchemaChanges.FIELD_SCHEMA;
-
-/**
- * The {@code DebeziumSchemaIncludeRecordParser} class extends the {@link
RecordParser} to parse
- * Debezium's change data capture (CDC) records, specifically focusing on
parsing records that
- * include the complete Debezium schema. This inclusion of the schema allows
for a more detailed and
- * accurate interpretation of the CDC data.
- */
-public class DebeziumSchemaIncludeRecordParser extends DebeziumRecordParser {
-
- private JsonNode schema;
-
- public DebeziumSchemaIncludeRecordParser(
- boolean caseSensitive, TypeMapping typeMapping,
List<ComputedColumn> computedColumns) {
- super(caseSensitive, typeMapping, computedColumns);
- }
-
- @Override
- protected LinkedHashMap<String, DataType> fillDefaultTypes(JsonNode
record) {
- CdcDebeziumSchema cdcDebeziumSchema;
- try {
- cdcDebeziumSchema = JsonSerdeUtil.fromJson(schema.toString(),
CdcDebeziumSchema.class);
- } catch (UncheckedIOException e) {
- throw new RuntimeException(e);
- }
- LinkedHashMap<String, DataType> fieldTypes = new LinkedHashMap<>();
- CdcDebeziumSchema field = cdcDebeziumSchema.fields().get(1);
- for (CdcDebeziumSchema f : field.fields()) {
- DataType dataType = toDataType(f.type());
- fieldTypes.put(f.field(), dataType);
- }
- return fieldTypes;
- }
-
- public static DataType toDataType(String type) {
- switch (type) {
- case "int32":
- return DataTypes.INT();
- case "int64":
- return DataTypes.BIGINT();
- case "string":
- return DataTypes.STRING();
- case "float32":
- case "float64":
- return DataTypes.FLOAT();
- case "double":
- return DataTypes.DOUBLE();
- case "bytes":
- return DataTypes.BYTES();
- default:
- throw new UnsupportedOperationException(
- String.format("Don't support type '%s' yet.", type));
- }
- }
-
- @Override
- protected void setRoot(String record) {
- JsonNode node = JsonSerdeUtil.fromJson(record, JsonNode.class);
- root = node.get(FIELD_PAYLOAD);
- schema = node.get(FIELD_SCHEMA);
- }
-
- @Override
- protected String format() {
- return "debezium_json_schema_include";
- }
-}
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
new file mode 100644
index 000000000..13874872b
--- /dev/null
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action.cdc.format.debezium;
+
+import org.apache.paimon.flink.action.cdc.TypeMapping;
+import org.apache.paimon.flink.action.cdc.mysql.MySqlTypeUtils;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.utils.DateTimeUtils;
+import org.apache.paimon.utils.StringUtils;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+import io.debezium.data.Bits;
+import io.debezium.data.geometry.Geometry;
+import io.debezium.data.geometry.Point;
+import io.debezium.time.Date;
+import io.debezium.time.MicroTime;
+import io.debezium.time.MicroTimestamp;
+import io.debezium.time.Timestamp;
+import io.debezium.time.ZonedTimestamp;
+import org.apache.kafka.connect.data.Decimal;
+import org.apache.kafka.connect.json.JsonConverterConfig;
+
+import javax.annotation.Nullable;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.util.Base64;
+import java.util.Map;
+
+import static
org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.TO_STRING;
+
+/**
+ * Utils to handle 'schema' field in debezium Json. TODO: The methods have
many duplicate codes with
+ * MySqlRecordParser. Need refactor.
+ */
+public class DebeziumSchemaUtils {
+
+ /** Transform raw string value according to schema. */
+ public static String transformRawValue(
+ @Nullable String rawValue,
+ String debeziumType,
+ @Nullable String className,
+ TypeMapping typeMapping,
+ JsonNode origin) {
+ if (rawValue == null) {
+ return null;
+ }
+
+ String transformed = rawValue;
+
+ if (Bits.LOGICAL_NAME.equals(className)) {
+ // transform little-endian form to normal order
+ //
https://debezium.io/documentation/reference/stable/connectors/mysql.html#mysql-data-types
+ byte[] littleEndian = Base64.getDecoder().decode(rawValue);
+ byte[] bigEndian = new byte[littleEndian.length];
+ for (int i = 0; i < littleEndian.length; i++) {
+ bigEndian[i] = littleEndian[littleEndian.length - 1 - i];
+ }
+ if (typeMapping.containsMode(TO_STRING)) {
+ transformed = StringUtils.bytesToBinaryString(bigEndian);
+ } else {
+ transformed = Base64.getEncoder().encodeToString(bigEndian);
+ }
+ } else if (("bytes".equals(debeziumType) && className == null)) {
+ // MySQL binary, varbinary, blob
+ transformed = new String(Base64.getDecoder().decode(rawValue));
+ } else if ("bytes".equals(debeziumType) &&
Decimal.LOGICAL_NAME.equals(className)) {
+ // MySQL numeric, fixed, decimal
+ try {
+ new BigDecimal(rawValue);
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "Invalid big decimal value "
+ + rawValue
+ + ". Make sure that in the
`customConverterConfigs` "
+ + "of the JsonDebeziumDeserializationSchema
you created, set '"
+ + JsonConverterConfig.DECIMAL_FORMAT_CONFIG
+ + "' to 'numeric'",
+ e);
+ }
+ }
+ // pay attention to the temporal types
+ //
https://debezium.io/documentation/reference/stable/connectors/mysql.html#mysql-temporal-types
+ else if (Date.SCHEMA_NAME.equals(className)) {
+ // MySQL date
+ transformed =
DateTimeUtils.toLocalDate(Integer.parseInt(rawValue)).toString();
+ } else if (Timestamp.SCHEMA_NAME.equals(className)) {
+ // MySQL datetime (precision 0-3)
+
+ // display value of datetime is not affected by timezone, see
+ // https://dev.mysql.com/doc/refman/8.0/en/datetime.html for
standard, and
+ // RowDataDebeziumDeserializeSchema#convertToTimestamp in
flink-cdc-connector
+ // for implementation
+ LocalDateTime localDateTime =
+ DateTimeUtils.toLocalDateTime(Long.parseLong(rawValue),
ZoneOffset.UTC);
+ transformed = DateTimeUtils.formatLocalDateTime(localDateTime, 3);
+ } else if (MicroTimestamp.SCHEMA_NAME.equals(className)) {
+ // MySQL datetime (precision 4-6)
+ long microseconds = Long.parseLong(rawValue);
+ long microsecondsPerSecond = 1_000_000;
+ long nanosecondsPerMicros = 1_000;
+ long seconds = microseconds / microsecondsPerSecond;
+ long nanoAdjustment = (microseconds % microsecondsPerSecond) *
nanosecondsPerMicros;
+
+ // display value of datetime is not affected by timezone, see
+ // https://dev.mysql.com/doc/refman/8.0/en/datetime.html for
standard, and
+ // RowDataDebeziumDeserializeSchema#convertToTimestamp in
flink-cdc-connector
+ // for implementation
+ LocalDateTime localDateTime =
+ Instant.ofEpochSecond(seconds, nanoAdjustment)
+ .atZone(ZoneOffset.UTC)
+ .toLocalDateTime();
+ transformed = DateTimeUtils.formatLocalDateTime(localDateTime, 6);
+ } else if (ZonedTimestamp.SCHEMA_NAME.equals(className)) {
+ // MySQL timestamp
+
+ // display value of timestamp is affected by timezone, see
+ // https://dev.mysql.com/doc/refman/8.0/en/datetime.html for
standard, and
+ // RowDataDebeziumDeserializeSchema#convertToTimestamp in
flink-cdc-connector
+ // for implementation
+ // TODO currently we cannot get zone id
+ LocalDateTime localDateTime =
+
Instant.parse(rawValue).atZone(ZoneOffset.UTC).toLocalDateTime();
+ transformed = DateTimeUtils.formatLocalDateTime(localDateTime, 6);
+ } else if (MicroTime.SCHEMA_NAME.equals(className)) {
+ long microseconds = Long.parseLong(rawValue);
+ long microsecondsPerSecond = 1_000_000;
+ long nanosecondsPerMicros = 1_000;
+ long seconds = microseconds / microsecondsPerSecond;
+ long nanoAdjustment = (microseconds % microsecondsPerSecond) *
nanosecondsPerMicros;
+
+ transformed =
+ Instant.ofEpochSecond(seconds, nanoAdjustment)
+ .atZone(ZoneOffset.UTC)
+ .toLocalTime()
+ .toString();
+ } else if (Point.LOGICAL_NAME.equals(className)
+ || Geometry.LOGICAL_NAME.equals(className)) {
+ try {
+ byte[] wkb = origin.get(Geometry.WKB_FIELD).binaryValue();
+ transformed = MySqlTypeUtils.convertWkbArray(wkb);
+ } catch (Exception e) {
+ throw new IllegalArgumentException(
+ String.format("Failed to convert %s to geometry
JSON.", rawValue), e);
+ }
+ }
+
+ return transformed;
+ }
+
+ public static DataType toDataType(
+ String debeziumType, @Nullable String className, Map<String,
String> parameters) {
+ if (className != null) {
+ switch (className) {
+ case Bits.LOGICAL_NAME:
+ int length = Integer.parseInt(parameters.get("length"));
+ return DataTypes.BINARY((length + 7) / 8);
+ case Decimal.LOGICAL_NAME:
+ String precision =
parameters.get("connect.decimal.precision");
+ if (precision == null) {
+ return DataTypes.DECIMAL(20, 0);
+ }
+
+ int p = Integer.parseInt(precision);
+ if (p > DecimalType.MAX_PRECISION) {
+ return DataTypes.STRING();
+ } else {
+ int scale = Integer.parseInt(parameters.get("scale"));
+ return DataTypes.DECIMAL(p, scale);
+ }
+ case Date.SCHEMA_NAME:
+ return DataTypes.DATE();
+ case Timestamp.SCHEMA_NAME:
+ return DataTypes.TIMESTAMP(3);
+ case MicroTimestamp.SCHEMA_NAME:
+ case ZonedTimestamp.SCHEMA_NAME:
+ return DataTypes.TIMESTAMP(6);
+ case MicroTime.SCHEMA_NAME:
+ return DataTypes.TIME();
+ }
+ }
+
+ return fromDebeziumType(debeziumType);
+ }
+
+ private static DataType fromDebeziumType(String dbzType) {
+ switch (dbzType) {
+ case "int8":
+ return DataTypes.TINYINT();
+ case "int16":
+ return DataTypes.SMALLINT();
+ case "int32":
+ return DataTypes.INT();
+ case "int64":
+ return DataTypes.BIGINT();
+ case "float32":
+ case "float64":
+ return DataTypes.FLOAT();
+ case "double":
+ return DataTypes.DOUBLE();
+ case "boolean":
+ return DataTypes.BOOLEAN();
+ case "bytes":
+ return DataTypes.BYTES();
+ case "string":
+ default:
+ return DataTypes.STRING();
+ }
+ }
+}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaDebeziumSyncTableActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaDebeziumSyncTableActionITCase.java
index e5e030a60..7aba174d3 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaDebeziumSyncTableActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaDebeziumSyncTableActionITCase.java
@@ -25,7 +25,6 @@ import org.junit.jupiter.api.Timeout;
public class KafkaDebeziumSyncTableActionITCase extends
KafkaSyncTableActionITCase {
private static final String DEBEZIUM = "debezium";
- private static final String DEBEZIUM_JSON_SCHEMA_INCLUDE =
"debezium_json_schema_include";
@Test
@Timeout(60)
@@ -96,6 +95,12 @@ public class KafkaDebeziumSyncTableActionITCase extends
KafkaSyncTableActionITCa
@Test
@Timeout(60)
public void testSchemaIncludeRecord1() throws Exception {
- testSchemaIncludeRecord(DEBEZIUM_JSON_SCHEMA_INCLUDE);
+ testSchemaIncludeRecord(DEBEZIUM);
+ }
+
+ @Test
+ @Timeout(60)
+ public void testAllTypesWithSchema() throws Exception {
+ testAllTypesWithSchemaImpl(DEBEZIUM);
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionITCase.java
index da38b2a0c..47ab426c2 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionITCase.java
@@ -675,7 +675,7 @@ public class KafkaSyncTableActionITCase extends
KafkaActionITCaseBase {
}
Map<String, String> kafkaConfig = getBasicKafkaConfig();
- kafkaConfig.put(VALUE_FORMAT.key(), format);
+ kafkaConfig.put(VALUE_FORMAT.key(), format + "-json");
kafkaConfig.put(TOPIC.key(), topic);
KafkaSyncTableAction action =
syncTableActionBuilder(kafkaConfig)
@@ -701,4 +701,251 @@ public class KafkaSyncTableActionITCase extends
KafkaActionITCaseBase {
"+I[101, scooter, Small 2-wheel scooter,
3.140000104904175]");
waitForResult(expected, table, rowType, primaryKeys);
}
+
+ // TODO some types are different from mysql cdc; maybe need to fix
+ public void testAllTypesWithSchemaImpl(String format) throws Exception {
+ String topic = "schema_include_all_type";
+ createTestTopic(topic, 1, 1);
+
+ List<String> lines =
readLines("kafka/debezium/table/schema/alltype/debezium-data-1.txt");
+ writeRecordsToKafka(topic, lines);
+
+ Map<String, String> kafkaConfig = getBasicKafkaConfig();
+ kafkaConfig.put(VALUE_FORMAT.key(), format + "-json");
+ kafkaConfig.put(TOPIC.key(), topic);
+ KafkaSyncTableAction action =
+ syncTableActionBuilder(kafkaConfig)
+ .withPartitionKeys("pt")
+ .withPrimaryKeys("pt", "_id")
+ .withTableConfig(getBasicTableConfig())
+ .build();
+ runActionWithDefaultEnv(action);
+
+ waitingTables(tableName);
+ FileStoreTable table = getFileStoreTable(tableName);
+
+ RowType rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(), // _id
+ DataTypes.DECIMAL(2, 1).notNull(), // pt
+ DataTypes.BOOLEAN(), // _bit1
+ DataTypes.BINARY(8), // _bit
+ DataTypes.SMALLINT(), // _tinyint1 different from
mysql cdc
+ DataTypes.SMALLINT(), // _boolean different from
mysql cdc
+ DataTypes.SMALLINT(), // _bool different from
mysql cdc
+ DataTypes.SMALLINT(), // _tinyint different from
mysql cdc
+ DataTypes.SMALLINT(), // _tinyint_unsigned
+ DataTypes.SMALLINT(), // _tinyint_unsigned_zerofill
+ DataTypes.SMALLINT(), // _smallint
+ DataTypes.INT(), // _smallint_unsigned
+ DataTypes.INT(), // _smallint_unsigned_zerofill
+ DataTypes.INT(), // _mediumint
+ DataTypes.INT(), // _mediumint_unsigned different
from mysql cdc
+ DataTypes
+ .INT(), // _mediumint_unsigned_zerofill
different from mysql cdc
+ DataTypes.INT(), // _int
+ DataTypes.BIGINT(), // _int_unsigned
+ DataTypes.BIGINT(), // _int_unsigned_zerofill
+ DataTypes.BIGINT(), // _bigint
+ DataTypes.DECIMAL(20, 0), // _bigint_unsigned
+ DataTypes.DECIMAL(20, 0), //
_bigint_unsigned_zerofill
+ DataTypes.DECIMAL(20, 0), // _serial different
from mysql cdc
+ DataTypes.DOUBLE(), // _float different from mysql
cdc
+ DataTypes.DOUBLE(), // _float_unsigned different
from mysql cdc
+ DataTypes.DOUBLE(), // _float_unsigned_zerofill
different from mysql cdc
+ DataTypes.DOUBLE(), // _real
+ DataTypes.DOUBLE(), // _real_unsigned
+ DataTypes.DOUBLE(), // _real_unsigned_zerofill
+ DataTypes.DOUBLE(), // _double
+ DataTypes.DOUBLE(), // _double_unsigned
+ DataTypes.DOUBLE(), // _double_unsigned_zerofill
+ DataTypes.DOUBLE(), // _double_precision
+ DataTypes.DOUBLE(), // _double_precision_unsigned
+ DataTypes.DOUBLE(), //
_double_precision_unsigned_zerofill
+ DataTypes.DECIMAL(8, 3), // _numeric
+ DataTypes.DECIMAL(8, 3), // _numeric_unsigned
+ DataTypes.DECIMAL(8, 3), //
_numeric_unsigned_zerofill
+ DataTypes.STRING(), // _fixed
+ DataTypes.STRING(), // _fixed_unsigned
+ DataTypes.STRING(), // _fixed_unsigned_zerofill
+ DataTypes.DECIMAL(8, 0), // _decimal
+ DataTypes.DECIMAL(8, 0), // _decimal_unsigned
+ DataTypes.DECIMAL(8, 0), //
_decimal_unsigned_zerofill
+ DataTypes.DECIMAL(38, 10), // _big_decimal
+ DataTypes.DATE(), // _date
+ DataTypes.TIMESTAMP(3), // _datetime different
from mysql cdc
+ DataTypes.TIMESTAMP(3), // _datetime3
+ DataTypes.TIMESTAMP(6), // _datetime6
+ DataTypes.TIMESTAMP(3), // _datetime_p different
from mysql cdc
+ DataTypes.TIMESTAMP(3), // _datetime_p2 different
from mysql cdc
+ DataTypes.TIMESTAMP(6), // _timestamp
+ DataTypes.TIMESTAMP(6), // _timestamp0 different
from mysql cdc
+ DataTypes.STRING(), // _char different from mysql
cdc
+ DataTypes.STRING(), // _varchar different from
mysql cdc
+ DataTypes.STRING(), // _tinytext
+ DataTypes.STRING(), // _text
+ DataTypes.STRING(), // _mediumtext
+ DataTypes.STRING(), // _longtext
+ DataTypes.BYTES(), // _bin different from mysql cdc
+ DataTypes.BYTES(), // _varbin different from mysql
cdc
+ DataTypes.BYTES(), // _tinyblob
+ DataTypes.BYTES(), // _blob
+ DataTypes.BYTES(), // _mediumblob
+ DataTypes.BYTES(), // _longblob
+ DataTypes.STRING(), // _json
+ DataTypes.STRING(), // _enum
+ DataTypes.INT(), // _year
+ DataTypes.TIME(), // _time
+ DataTypes.STRING(), // _point
+ DataTypes.STRING(), // _geometry
+ DataTypes.STRING(), // _linestring
+ DataTypes.STRING(), // _polygon
+ DataTypes.STRING(), // _multipoint
+ DataTypes.STRING(), // _multiline
+ DataTypes.STRING(), // _multipolygon
+ DataTypes.STRING(), // _geometrycollection
+ DataTypes.STRING() // _set different from mysql cdc
+ },
+ new String[] {
+ "_id",
+ "pt",
+ "_bit1",
+ "_bit",
+ "_tinyint1",
+ "_boolean",
+ "_bool",
+ "_tinyint",
+ "_tinyint_unsigned",
+ "_tinyint_unsigned_zerofill",
+ "_smallint",
+ "_smallint_unsigned",
+ "_smallint_unsigned_zerofill",
+ "_mediumint",
+ "_mediumint_unsigned",
+ "_mediumint_unsigned_zerofill",
+ "_int",
+ "_int_unsigned",
+ "_int_unsigned_zerofill",
+ "_bigint",
+ "_bigint_unsigned",
+ "_bigint_unsigned_zerofill",
+ "_serial",
+ "_float",
+ "_float_unsigned",
+ "_float_unsigned_zerofill",
+ "_real",
+ "_real_unsigned",
+ "_real_unsigned_zerofill",
+ "_double",
+ "_double_unsigned",
+ "_double_unsigned_zerofill",
+ "_double_precision",
+ "_double_precision_unsigned",
+ "_double_precision_unsigned_zerofill",
+ "_numeric",
+ "_numeric_unsigned",
+ "_numeric_unsigned_zerofill",
+ "_fixed",
+ "_fixed_unsigned",
+ "_fixed_unsigned_zerofill",
+ "_decimal",
+ "_decimal_unsigned",
+ "_decimal_unsigned_zerofill",
+ "_big_decimal",
+ "_date",
+ "_datetime",
+ "_datetime3",
+ "_datetime6",
+ "_datetime_p",
+ "_datetime_p2",
+ "_timestamp",
+ "_timestamp0",
+ "_char",
+ "_varchar",
+ "_tinytext",
+ "_text",
+ "_mediumtext",
+ "_longtext",
+ "_bin",
+ "_varbin",
+ "_tinyblob",
+ "_blob",
+ "_mediumblob",
+ "_longblob",
+ "_json",
+ "_enum",
+ "_year",
+ "_time",
+ "_point",
+ "_geometry",
+ "_linestring",
+ "_polygon",
+ "_multipoint",
+ "_multiline",
+ "_multipolygon",
+ "_geometrycollection",
+ "_set",
+ });
+
+ // BIT(64) data: 0B11111000111 -> 0B00000111_11000111
+ String bits =
+ Arrays.toString(
+ new byte[] {0, 0, 0, 0, 0, 0, (byte) 0B00000111,
(byte) 0B11000111});
+ List<String> expected =
+ Collections.singletonList(
+ "+I["
+ + "1, 1.1, "
+ + String.format("true, %s, ", bits)
+ + "1, 1, 0, 1, 2, 3, "
+ + "1000, 2000, 3000, "
+ + "100000, 200000, 300000, "
+ + "1000000, 2000000, 3000000, "
+ + "10000000000, 20000000000, 30000000000,
40000000000, "
+ + "1.5, 2.5, 3.5, "
+ + "1.000001, 2.000002, 3.000003, "
+ + "1.000011, 2.000022, 3.000033, "
+ + "1.000111, 2.000222, 3.000333, "
+ + "12345.110, 12345.220, 12345.330, "
+ // TODO fix FIXED
+ + "1.2345678987654322E32,
1.2345678987654322E32, 1.2345678987654322E32, "
+ // TODO fix BIG DECIMAL
+ + "11111, 22222, 33333,
2222222222222222400000000000.0000000000, "
+ + "19439, "
+ // display value of datetime is not affected
by timezone
+ + "2023-03-23T14:30:05,
2023-03-23T14:30:05.123, 2023-03-23T14:30:05.123456, "
+ + "2023-03-24T14:30, 2023-03-24T14:30:05.120, "
+ // display value of timestamp is affected by
timezone
+ // we store 2023-03-23T15:00:10.123456 in
UTC-8 system timezone
+ // and query this timestamp in UTC-5 MySQL
server timezone
+ // so the display value should increase by 3
hour
+ // TODO haven't handle zone
+ + "2023-03-23T22:00:10.123456,
2023-03-23T07:10, "
+ + "Paimon, Apache Paimon, Apache Paimon MySQL
TINYTEXT Test Data, Apache Paimon MySQL Test Data, Apache Paimon MySQL
MEDIUMTEXT Test Data, Apache Paimon MySQL Long Test Data, "
+ + "[98, 121, 116, 101, 115, 0, 0, 0, 0, 0], "
+ + "[109, 111, 114, 101, 32, 98, 121, 116, 101,
115], "
+ + "[84, 73, 78, 89, 66, 76, 79, 66, 32, 116,
121, 112, 101, 32, 116, 101, 115, 116, 32, 100, 97, 116, 97], "
+ + "[66, 76, 79, 66, 32, 116, 121, 112, 101,
32, 116, 101, 115, 116, 32, 100, 97, 116, 97], "
+ + "[77, 69, 68, 73, 85, 77, 66, 76, 79, 66,
32, 116, 121, 112, 101, 32, 116, 101, 115, 116, 32, 100, 97, 116, 97], "
+ + "[76, 79, 78, 71, 66, 76, 79, 66, 32, 32,
98, 121, 116, 101, 115, 32, 116, 101, 115, 116, 32, 100, 97, 116, 97], "
+ + "{\"a\": \"b\"}, "
+ + "value1, "
+ + "2023, "
+ + "36803000, "
+ +
"{\"coordinates\":[1,1],\"type\":\"Point\",\"srid\":0}, "
+ +
"{\"coordinates\":[[[1,1],[2,1],[2,2],[1,2],[1,1]]],\"type\":\"Polygon\",\"srid\":0},
"
+ +
"{\"coordinates\":[[3,0],[3,3],[3,5]],\"type\":\"LineString\",\"srid\":0}, "
+ +
"{\"coordinates\":[[[1,1],[2,1],[2,2],[1,2],[1,1]]],\"type\":\"Polygon\",\"srid\":0},
"
+ +
"{\"coordinates\":[[1,1],[2,2]],\"type\":\"MultiPoint\",\"srid\":0}, "
+ +
"{\"coordinates\":[[[1,1],[2,2],[3,3]],[[4,4],[5,5]]],\"type\":\"MultiLineString\",\"srid\":0},
"
+ +
"{\"coordinates\":[[[[0,0],[10,0],[10,10],[0,10],[0,0]]],[[[5,5],[7,5],[7,7],[5,7],[5,5]]]],\"type\":\"MultiPolygon\",\"srid\":0},
"
+ +
"{\"geometries\":[{\"type\":\"Point\",\"coordinates\":[10,10]},{\"type\":\"Point\",\"coordinates\":[30,30]},{\"type\":\"LineString\",\"coordinates\":[[15,15],[20,20]]}],\"type\":\"GeometryCollection\",\"srid\":0},
"
+ // TODO fix set
+ + "a,b"
+ + "]");
+
+ List<String> primaryKeys = Arrays.asList("pt", "_id");
+
+ waitForResult(expected, table, rowType, primaryKeys);
+ }
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/alltype/debezium-data-1.txt
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/alltype/debezium-data-1.txt
new file mode 100644
index 000000000..b398425ce
--- /dev/null
+++
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/alltype/debezium-data-1.txt
@@ -0,0 +1,19 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"_id"},{"type":"bytes","optional":true,"name":"org.apache.kafka.connect.data.Decimal","version":1,"parameters":{"scale":"1","connect.decimal.precision":"2"},"field":"pt"},{"type":"boolean","optional":true,"field":"_bit1"},{"type":"bytes","optional":true,"name":"io.debezium.data.Bits","version":1,"parameters":{"length":"64"},"field":"_bit"},{"type":"int16","optional":true,"field":"_tin
[...]
