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 05c17b150 [flink] Introduce record parser for debezium with schema
(#2730)
05c17b150 is described below
commit 05c17b150ca8c2efea1d39c3d0ea0cdd026e3e82
Author: monster <[email protected]>
AuthorDate: Tue Jan 23 11:46:38 2024 +0800
[flink] Introduce record parser for debezium with schema (#2730)
---
.../paimon/flink/action/cdc/format/DataFormat.java | 4 +-
.../flink/action/cdc/format/RecordParser.java | 4 +-
.../action/cdc/format/canal/CanalRecordParser.java | 2 +-
.../cdc/format/debezium/CdcDebeziumSchema.java | 91 +++++++++++++++++++
.../cdc/format/debezium/DebeziumRecordParser.java | 2 +-
.../DebeziumSchemaIncludeRecordParser.java | 100 +++++++++++++++++++++
.../kafka/KafkaDebeziumSyncTableActionITCase.java | 7 ++
.../cdc/kafka/KafkaSyncTableActionITCase.java | 39 ++++++++
.../table/schema/include/debezium-data-1.txt | 19 ++++
9 files changed, 263 insertions(+), 5 deletions(-)
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 28dc3e457..5344190d4 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,6 +22,7 @@ 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;
@@ -38,7 +39,8 @@ public enum DataFormat {
CANAL_JSON(CanalRecordParser::new),
OGG_JSON(OggRecordParser::new),
MAXWELL_JSON(MaxwellRecordParser::new),
- DEBEZIUM_JSON(DebeziumRecordParser::new);
+ DEBEZIUM_JSON(DebeziumRecordParser::new),
+ DEBEZIUM_JSON_SCHEMA_INCLUDE(DebeziumSchemaIncludeRecordParser::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/RecordParser.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
index 66d313135..91bc946c8 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
@@ -119,7 +119,7 @@ public abstract class RecordParser implements
FlatMapFunction<String, RichCdcMul
}
// use STRING type in default when we cannot get origin data types (most
cases)
- protected LinkedHashMap<String, DataType> fillDefaultStringTypes(JsonNode
record) {
+ protected LinkedHashMap<String, DataType> fillDefaultTypes(JsonNode
record) {
LinkedHashMap<String, DataType> fieldTypes = new LinkedHashMap<>();
record.fieldNames().forEachRemaining(name -> fieldTypes.put(name,
DataTypes.STRING()));
return fieldTypes;
@@ -138,7 +138,7 @@ public abstract class RecordParser implements
FlatMapFunction<String, RichCdcMul
protected Map<String, String> extractRowData(
JsonNode record, LinkedHashMap<String, DataType> paimonFieldTypes)
{
- paimonFieldTypes.putAll(fillDefaultStringTypes(record));
+ paimonFieldTypes.putAll(fillDefaultTypes(record));
Map<String, Object> recordMap =
JsonSerdeUtil.convertValue(record, new
TypeReference<Map<String, Object>>() {});
Map<String, String> rowData =
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParser.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParser.java
index 2877b0ed3..c7cd06c18 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParser.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParser.java
@@ -177,7 +177,7 @@ public class CanalRecordParser extends RecordParser {
rowData.put(fieldName, newValue);
}
} else {
- paimonFieldTypes.putAll(fillDefaultStringTypes(record));
+ paimonFieldTypes.putAll(fillDefaultTypes(record));
for (Map.Entry<String, Object> entry : recordMap.entrySet()) {
rowData.put(entry.getKey(), Objects.toString(entry.getValue(),
null));
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/CdcDebeziumSchema.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/CdcDebeziumSchema.java
new file mode 100644
index 000000000..64f3a1267
--- /dev/null
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/CdcDebeziumSchema.java
@@ -0,0 +1,91 @@
+/*
+ * 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.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.List;
+
+/** Cdc Debezium Schema Entity. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CdcDebeziumSchema {
+
+ private static final String FIELD_FIELDS = "fields";
+ private static final String FIELD_NAME = "name";
+ private static final String FIELD_TYPE = "type";
+ private static final String FIELD_FIELD = "field";
+ private static final String FIELD_OPTIONAL = "optional";
+
+ @JsonProperty(FIELD_FIELD)
+ private final String field;
+
+ @JsonProperty(FIELD_TYPE)
+ private final String type;
+
+ @JsonProperty(FIELD_OPTIONAL)
+ private final Boolean optional;
+
+ @JsonProperty(FIELD_FIELDS)
+ private final List<CdcDebeziumSchema> fields;
+
+ @JsonProperty(FIELD_NAME)
+ private final String name;
+
+ @JsonCreator
+ public CdcDebeziumSchema(
+ @JsonProperty(FIELD_FIELD) String field,
+ @JsonProperty(FIELD_TYPE) String type,
+ @JsonProperty(FIELD_OPTIONAL) Boolean optional,
+ @JsonProperty(FIELD_FIELDS) List<CdcDebeziumSchema> fields,
+ @JsonProperty(FIELD_NAME) String name) {
+ this.field = field;
+ this.type = type;
+ this.optional = optional;
+ this.fields = fields;
+ this.name = name;
+ }
+
+ @JsonGetter(FIELD_FIELD)
+ public String field() {
+ return field;
+ }
+
+ @JsonGetter(FIELD_TYPE)
+ public String type() {
+ return type;
+ }
+
+ @JsonGetter(FIELD_OPTIONAL)
+ public Boolean optional() {
+ return optional;
+ }
+
+ @JsonGetter(FIELD_FIELDS)
+ public List<CdcDebeziumSchema> fields() {
+ return fields;
+ }
+
+ @JsonGetter(FIELD_NAME)
+ public String name() {
+ return name;
+ }
+}
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 1301bb532..a736e5b49 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
@@ -54,7 +54,7 @@ import static org.apache.paimon.utils.JsonSerdeUtil.isNull;
public class DebeziumRecordParser extends RecordParser {
private static final String FIELD_SCHEMA = "schema";
- private static final String FIELD_PAYLOAD = "payload";
+ protected static final String FIELD_PAYLOAD = "payload";
private static final String FIELD_BEFORE = "before";
private static final String FIELD_AFTER = "after";
private static final String FIELD_SOURCE = "source";
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
new file mode 100644
index 000000000..8c9b3402f
--- /dev/null
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaIncludeRecordParser.java
@@ -0,0 +1,100 @@
+/*
+ * 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/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 bb9061761..e5e030a60 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,6 +25,7 @@ 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)
@@ -91,4 +92,10 @@ public class KafkaDebeziumSyncTableActionITCase extends
KafkaSyncTableActionITCa
public void testKafkaBuildSchemaWithDelete() throws Exception {
testKafkaBuildSchemaWithDelete(DEBEZIUM);
}
+
+ @Test
+ @Timeout(60)
+ public void testSchemaIncludeRecord1() throws Exception {
+ testSchemaIncludeRecord(DEBEZIUM_JSON_SCHEMA_INCLUDE);
+ }
}
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 f025e9c96..aabb5d53c 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
@@ -663,4 +663,43 @@ public class KafkaSyncTableActionITCase extends
KafkaActionITCaseBase {
Thread.sleep(1000);
}
}
+
+ public void testSchemaIncludeRecord(String format) throws Exception {
+ String topic = "schema_include";
+ createTestTopic(topic, 1, 1);
+
+ List<String> lines =
readLines("kafka/debezium/table/schema/include/debezium-data-1.txt");
+ try {
+ writeRecordsToKafka(topic, lines);
+ } catch (Exception e) {
+ throw new Exception("Failed to write debezium data to Kafka.", e);
+ }
+
+ Map<String, String> kafkaConfig = getBasicKafkaConfig();
+ kafkaConfig.put(VALUE_FORMAT.key(), format);
+ kafkaConfig.put(TOPIC.key(), topic);
+ KafkaSyncTableAction action =
+ syncTableActionBuilder(kafkaConfig)
+ .withPrimaryKeys("id")
+ .withTableConfig(getBasicTableConfig())
+ .build();
+ runActionWithDefaultEnv(action);
+
+ FileStoreTable table = getFileStoreTable(tableName);
+
+ RowType rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.DOUBLE()
+ },
+ new String[] {"id", "name", "description", "weight"});
+ List<String> primaryKeys = Collections.singletonList("id");
+ List<String> expected =
+ Collections.singletonList(
+ "+I[101, scooter, Small 2-wheel scooter,
3.140000104904175]");
+ waitForResult(expected, table, rowType, primaryKeys);
+ }
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/include/debezium-data-1.txt
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/include/debezium-data-1.txt
new file mode 100644
index 000000000..417ba5567
--- /dev/null
+++
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/include/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":"string","optional":false,"field":"name"},{"type":"string","optional":true,"field":"description"},{"type":"double","optional":true,"field":"weight"}],"optional":true,"name":"dbserver1.inventory.products.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"name"},{"type":"string","opt
[...]
\ No newline at end of file
