metadataKeys) {
+ this.metadataKeys = metadataKeys;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.newBuilder()
+ .addContainedKind(RowKind.INSERT)
+ .addContainedKind(RowKind.UPDATE_BEFORE)
+ .addContainedKind(RowKind.UPDATE_AFTER)
+ .addContainedKind(RowKind.DELETE)
+ .build();
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Metadata handling
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * List of metadata that can be read with this format.
+ */
+ public enum ReadableMetadata {
+
+ SCHEMA(
+ "schema",
+ DataTypes.STRING().nullable(),
+ false,
+ DataTypes.FIELD("schema", DataTypes.STRING()),
+ new MetadataConverter() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Object convert(GenericRowData row, int pos) {
+ return row.getString(pos);
+ }
+ }),
+
+ INGESTION_TIMESTAMP(
+ "ingestion-timestamp",
+ DataTypes.BIGINT().nullable(),
+ true,
+ DataTypes.FIELD("ts_ms", DataTypes.BIGINT()),
+ new MetadataConverter() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Object convert(GenericRowData row, int pos) {
+ return row.getLong(pos);
+ }
+ }),
+
+ SOURCE_TIMESTAMP(
+ "source.timestamp",
+ DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(3).nullable(),
+ true,
+ DataTypes.FIELD("source", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING())),
+ new MetadataConverter() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Object convert(GenericRowData row, int pos) {
+ final StringData timestamp =
+ (StringData) readProperty(row, pos, KEY_SOURCE_TIMESTAMP);
+ if (timestamp == null) {
+ return null;
+ }
+ return TimestampData.fromEpochMillis(Long.parseLong(timestamp.toString()));
+ }
+ }),
+
+ SOURCE_DATABASE(
+ "source.database",
+ DataTypes.STRING().nullable(),
+ true,
+ DataTypes.FIELD("source", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING())),
+ new MetadataConverter() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Object convert(GenericRowData row, int pos) {
+ return readProperty(row, pos, KEY_SOURCE_DATABASE);
+ }
+ }),
+
+ SOURCE_SCHEMA(
+ "source.schema",
+ DataTypes.STRING().nullable(),
+ true,
+ DataTypes.FIELD("source", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING())),
+ new MetadataConverter() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Object convert(GenericRowData row, int pos) {
+ return readProperty(row, pos, KEY_SOURCE_SCHEMA);
+ }
+ }),
+
+ SOURCE_TABLE(
+ "source.table",
+ DataTypes.STRING().nullable(),
+ true,
+ DataTypes.FIELD("source", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING())),
+ new MetadataConverter() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Object convert(GenericRowData row, int pos) {
+ return readProperty(row, pos, KEY_SOURCE_TABLE);
+ }
+ }),
+
+ SOURCE_PROPERTIES(
+ "source.properties",
+ // key and value of the map are nullable to make handling easier in queries
+ DataTypes.MAP(DataTypes.STRING().nullable(), DataTypes.STRING().nullable())
+ .nullable(),
+ true,
+ DataTypes.FIELD("source", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING())),
+ new MetadataConverter() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Object convert(GenericRowData row, int pos) {
+ return row.getMap(pos);
+ }
+ });
+
+ final String key;
+
+ final DataType dataType;
+
+ final boolean isJsonPayload;
+
+ final DataTypes.Field requiredJsonField;
+
+ final MetadataConverter converter;
+
+ ReadableMetadata(
+ String key,
+ DataType dataType,
+ boolean isJsonPayload,
+ DataTypes.Field requiredJsonField,
+ MetadataConverter converter) {
+ this.key = key;
+ this.dataType = dataType;
+ this.isJsonPayload = isJsonPayload;
+ this.requiredJsonField = requiredJsonField;
+ this.converter = converter;
+ }
+
+ public String getKey() {
+ return key;
+ }
+ }
+
+ private static final StringData KEY_SOURCE_TIMESTAMP = StringData.fromString("ts_ms");
+
+ private static final StringData KEY_SOURCE_DATABASE = StringData.fromString("db");
+
+ private static final StringData KEY_SOURCE_SCHEMA = StringData.fromString("schema");
+
+ private static final StringData KEY_SOURCE_TABLE = StringData.fromString("table");
+
+ private static Object readProperty(GenericRowData row, int pos, StringData key) {
+ final GenericMapData map = (GenericMapData) row.getMap(pos);
+ if (map == null) {
+ return null;
+ }
+ return map.get(key);
+ }
+}
diff --git a/inlong-sort/sort-formats/format-row/format-json-v1.18/src/main/java/org/apache/inlong/sort/formats/json/debezium/DebeziumJsonDeserializationSchema.java b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/main/java/org/apache/inlong/sort/formats/json/debezium/DebeziumJsonDeserializationSchema.java
new file mode 100644
index 00000000000..a53d6c99654
--- /dev/null
+++ b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/main/java/org/apache/inlong/sort/formats/json/debezium/DebeziumJsonDeserializationSchema.java
@@ -0,0 +1,397 @@
+/*
+ * 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.inlong.sort.formats.json.debezium;
+
+import org.apache.inlong.sort.formats.json.MysqlBinLogData;
+import org.apache.inlong.sort.formats.json.debezium.DebeziumJsonDecodingFormat.ReadableMetadata;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.formats.json.JsonRowDataDeserializationSchema;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GenericMapData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.utils.DataTypeUtils;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.Collector;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+import static java.lang.String.format;
+import static org.apache.inlong.sort.formats.json.debezium.DebeziumUtils.getMysqlMetadataKey;
+
+/**
+ * Copied from apache flink project with a litter change.
+ *
+ * Deserialization schema from Debezium JSON to Flink Table/SQL internal data structure {@link
+ * RowData}. The deserialization schema knows Debezium's schema definition and can extract the
+ * database data and convert into {@link RowData} with {@link RowKind}.
+ *
+ * Deserializes a byte[] message as a JSON object and reads the specified fields.
+ *
+ * Failures during deserialization are forwarded as wrapped IOExceptions.
+ *
+ * @see Debezium
+ */
+@Internal
+public final class DebeziumJsonDeserializationSchema implements DeserializationSchema {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final String OP_READ = "r"; // snapshot read
+ private static final String OP_CREATE = "c"; // insert
+ private static final String OP_UPDATE = "u"; // update
+ private static final String OP_DELETE = "d"; // delete
+
+ private static final int BEFORE_POS = 0;
+ private static final int AFTER_POS = 1;
+ private static final int OP_POS = 2;
+
+ private static final String REPLICA_IDENTITY_EXCEPTION =
+ "The \"before\" field of %s message is null, "
+ + "if you are using Debezium Postgres Connector, "
+ + "please check the Postgres table has been set REPLICA IDENTITY to FULL level.";
+
+ /** The deserializer to deserialize Debezium JSON data. */
+ private final JsonRowDataDeserializationSchema jsonDeserializer;
+
+ /** Flag that indicates that an additional projection is required for metadata. */
+ private final boolean hasMetadata;
+
+ /** Metadata to be extracted for every record. */
+ private final MetadataConverter[] metadataConverters;
+
+ private final List requestedMetadata;
+
+ /** {@link TypeInformation} of the produced {@link RowData} (physical + meta data). */
+ private final TypeInformation producedTypeInfo;
+
+ /**
+ * Flag indicating whether the Debezium JSON data contains schema part or not. When Debezium
+ * Kafka Connect enables "value.converter.schemas.enable", the JSON will contain "schema"
+ * information, but we just ignore "schema" and extract data from "payload".
+ */
+ private final boolean schemaInclude;
+
+ /**
+ * Flag indicating whether to emit update before row.
+ */
+ private final boolean updateBeforeInclude;
+
+ /** Flag indicating whether to ignore invalid fields/rows (default: throw an exception). */
+ private final boolean ignoreParseErrors;
+
+ private final boolean isMigrateAll;
+
+ /**
+ * Constructor of DebeziumJsonDeserializationSchema.
+ */
+ public DebeziumJsonDeserializationSchema(
+ DataType physicalDataType,
+ List requestedMetadata,
+ TypeInformation producedTypeInfo,
+ boolean schemaInclude,
+ boolean updateBeforeInclude,
+ boolean ignoreParseErrors,
+ TimestampFormat timestampFormat,
+ boolean isMigrateAll) {
+ this.isMigrateAll = isMigrateAll;
+ final RowType jsonRowType =
+ createJsonRowType(physicalDataType, requestedMetadata, schemaInclude, isMigrateAll);
+ this.jsonDeserializer =
+ new JsonRowDataDeserializationSchema(
+ jsonRowType,
+ // the result type is never used, so it's fine to pass in the produced type
+ // info
+ producedTypeInfo,
+ false, // ignoreParseErrors already contains the functionality of
+ // failOnMissingField
+ ignoreParseErrors,
+ timestampFormat);
+ this.hasMetadata = requestedMetadata.size() > 0;
+ this.metadataConverters =
+ createMetadataConverters(jsonRowType, requestedMetadata, schemaInclude);
+ this.requestedMetadata = requestedMetadata;
+ this.producedTypeInfo = producedTypeInfo;
+ this.schemaInclude = schemaInclude;
+ this.updateBeforeInclude = updateBeforeInclude;
+ this.ignoreParseErrors = ignoreParseErrors;
+ }
+
+ @Override
+ public RowData deserialize(byte[] message) {
+ throw new RuntimeException(
+ "Please invoke DeserializationSchema#deserialize(byte[], Collector) instead.");
+ }
+
+ @Override
+ public void deserialize(byte[] message, Collector out) throws IOException {
+ if (message == null || message.length == 0) {
+ // skip tombstone messages
+ return;
+ }
+ try {
+ GenericRowData row = (GenericRowData) jsonDeserializer.deserialize(message);
+ GenericRowData payload;
+ if (schemaInclude) {
+ payload = (GenericRowData) row.getField(0);
+ } else {
+ payload = row;
+ }
+
+ GenericRowData before;
+ GenericRowData after;
+ if (isMigrateAll) {
+ before = GenericRowData.of(payload.getField(BEFORE_POS));
+ after = GenericRowData.of(payload.getField(AFTER_POS));
+ } else {
+ before = (GenericRowData) payload.getField(BEFORE_POS);
+ after = (GenericRowData) payload.getField(AFTER_POS);
+ }
+
+ String op = payload.getField(OP_POS).toString();
+ if (OP_CREATE.equals(op) || OP_READ.equals(op)) {
+ after.setRowKind(RowKind.INSERT);
+ emitRow(row, after, out);
+ } else if (OP_UPDATE.equals(op)) {
+ if (before == null) {
+ throw new IllegalStateException(
+ String.format(REPLICA_IDENTITY_EXCEPTION, "UPDATE"));
+ }
+ before.setRowKind(RowKind.UPDATE_BEFORE);
+ after.setRowKind(RowKind.UPDATE_AFTER);
+ if (updateBeforeInclude) {
+ emitRow(row, before, out);
+ }
+ emitRow(row, after, out);
+ } else if (OP_DELETE.equals(op)) {
+ if (before == null) {
+ throw new IllegalStateException(
+ String.format(REPLICA_IDENTITY_EXCEPTION, "DELETE"));
+ }
+ before.setRowKind(RowKind.DELETE);
+ emitRow(row, before, out);
+ } else {
+ if (!ignoreParseErrors) {
+ throw new IOException(
+ format(
+ "Unknown \"op\" value \"%s\". The Debezium JSON message is '%s'",
+ op, new String(message)));
+ }
+ }
+ } catch (Throwable t) {
+ // a big try catch to protect the processing.
+ if (!ignoreParseErrors) {
+ throw new IOException(
+ format("Corrupt Debezium JSON message '%s'.", new String(message)), t);
+ }
+ }
+ }
+
+ private void emitRow(GenericRowData rootRow, GenericRowData physicalRow, Collector out) {
+ int physicalArity = physicalRow.getArity();
+ if (isMigrateAll) {
+ physicalArity = 0;
+ }
+ final int metadataArity = metadataConverters.length;
+
+ final GenericRowData producedRow = new GenericRowData(physicalRow.getRowKind(), physicalArity + 1);
+
+ for (int physicalPos = 0; physicalPos < physicalArity; physicalPos++) {
+ producedRow.setField(physicalPos + 1, physicalRow.getField(physicalPos));
+ }
+
+ // Put metadata in the first field of the emitted RowData
+ Map metadataMap = new HashMap<>();
+ metadataMap.put(
+ StringData.fromString(MysqlBinLogData.MYSQL_METADATA_IS_DDL),
+ StringData.fromString("false"));
+
+ for (int metadataPos = 0; metadataPos < metadataArity; metadataPos++) {
+ metadataMap.put(
+ StringData.fromString(getMysqlMetadataKey(requestedMetadata.get(metadataPos))),
+ StringData.fromString(metadataConverters[metadataPos].convert(rootRow).toString()));
+ }
+
+ if (isMigrateAll) {
+ metadataMap.put(
+ StringData.fromString(MysqlBinLogData.MYSQL_METADATA_DATA),
+ (StringData) physicalRow.getField(0));
+ }
+
+ producedRow.setField(0, new GenericMapData(metadataMap));
+
+ out.collect(producedRow);
+ }
+
+ @Override
+ public boolean isEndOfStream(RowData nextElement) {
+ return false;
+ }
+
+ @Override
+ public TypeInformation getProducedType() {
+ return producedTypeInfo;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ DebeziumJsonDeserializationSchema that = (DebeziumJsonDeserializationSchema) o;
+ return Objects.equals(jsonDeserializer, that.jsonDeserializer)
+ && hasMetadata == that.hasMetadata
+ && Objects.equals(producedTypeInfo, that.producedTypeInfo)
+ && schemaInclude == that.schemaInclude
+ && ignoreParseErrors == that.ignoreParseErrors;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ jsonDeserializer, hasMetadata, producedTypeInfo, schemaInclude, ignoreParseErrors);
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private static RowType createJsonRowType(
+ DataType physicalDataType,
+ List readableMetadata,
+ boolean schemaInclude,
+ boolean isMigrateAll) {
+
+ DataType dataTypeForDataFields = physicalDataType;
+ if (isMigrateAll) {
+ dataTypeForDataFields = DataTypes.STRING();
+ }
+
+ DataType payload = DataTypes.ROW(
+ DataTypes.FIELD("before", dataTypeForDataFields),
+ DataTypes.FIELD("after", dataTypeForDataFields),
+ DataTypes.FIELD("op", DataTypes.STRING()));
+
+ // append fields that are required for reading metadata in the payload
+ final List payloadMetadataFields =
+ readableMetadata.stream()
+ .filter(m -> m.isJsonPayload)
+ .map(m -> m.requiredJsonField)
+ .distinct()
+ .collect(Collectors.toList());
+ payload = DataTypeUtils.appendRowFields(payload, payloadMetadataFields);
+
+ DataType root = payload;
+ if (schemaInclude) {
+ // when Debezium Kafka connect enables "value.converter.schemas.enable",
+ // the JSON will contain "schema" information and we need to extract data from
+ // "payload".
+ root = DataTypes.ROW(DataTypes.FIELD("payload", payload));
+ }
+
+ // append fields that are required for reading metadata in the root
+ final List rootMetadataFields =
+ readableMetadata.stream()
+ .filter(m -> !m.isJsonPayload)
+ .map(m -> m.requiredJsonField)
+ .distinct()
+ .collect(Collectors.toList());
+ root = DataTypeUtils.appendRowFields(root, rootMetadataFields);
+
+ return (RowType) root.getLogicalType();
+ }
+
+ private static MetadataConverter[] createMetadataConverters(
+ RowType jsonRowType, List requestedMetadata, boolean schemaInclude) {
+ return requestedMetadata.stream()
+ .map(
+ m -> {
+ if (m.isJsonPayload) {
+ return convertInPayload(jsonRowType, m, schemaInclude);
+ } else {
+ return convertInRoot(jsonRowType, m);
+ }
+ })
+ .toArray(MetadataConverter[]::new);
+ }
+
+ private static MetadataConverter convertInRoot(RowType jsonRowType, ReadableMetadata metadata) {
+ final int pos = findFieldPos(metadata, jsonRowType);
+ return new MetadataConverter() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Object convert(GenericRowData root, int unused) {
+ return metadata.converter.convert(root, pos);
+ }
+ };
+ }
+
+ private static MetadataConverter convertInPayload(
+ RowType jsonRowType, ReadableMetadata metadata, boolean schemaInclude) {
+ if (schemaInclude) {
+ final int pos = findFieldPos(metadata, (RowType) jsonRowType.getChildren().get(0));
+ return new MetadataConverter() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Object convert(GenericRowData root, int unused) {
+ final GenericRowData payload = (GenericRowData) root.getField(0);
+ return metadata.converter.convert(payload, pos);
+ }
+ };
+ }
+ return convertInRoot(jsonRowType, metadata);
+ }
+
+ private static int findFieldPos(ReadableMetadata metadata, RowType jsonRowType) {
+ return jsonRowType.getFieldNames().indexOf(metadata.requiredJsonField.getName());
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Converter that extracts a metadata field from the row (root or payload) that comes out of the
+ * JSON schema and converts it to the desired data type.
+ */
+ interface MetadataConverter extends Serializable {
+
+ // Method for top-level access.
+ default Object convert(GenericRowData row) {
+ return convert(row, -1);
+ }
+
+ Object convert(GenericRowData row, int pos);
+ }
+}
diff --git a/inlong-sort/sort-formats/format-row/format-json-v1.18/src/main/java/org/apache/inlong/sort/formats/json/debezium/DebeziumUtils.java b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/main/java/org/apache/inlong/sort/formats/json/debezium/DebeziumUtils.java
new file mode 100644
index 00000000000..f95a3bef4ac
--- /dev/null
+++ b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/main/java/org/apache/inlong/sort/formats/json/debezium/DebeziumUtils.java
@@ -0,0 +1,37 @@
+/*
+ * 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.inlong.sort.formats.json.debezium;
+
+import org.apache.inlong.sort.formats.json.MysqlBinLogData;
+import org.apache.inlong.sort.formats.json.debezium.DebeziumJsonDecodingFormat.ReadableMetadata;
+
+public class DebeziumUtils {
+
+ public static String getMysqlMetadataKey(ReadableMetadata readableMetadata) {
+ switch (readableMetadata) {
+ case SOURCE_DATABASE:
+ return MysqlBinLogData.MYSQL_METADATA_DATABASE;
+ case SOURCE_TABLE:
+ return MysqlBinLogData.MYSQL_METADATA_TABLE;
+ case INGESTION_TIMESTAMP:
+ return MysqlBinLogData.MYSQL_METADATA_EVENT_TIME;
+ default:
+ throw new IllegalArgumentException("Not supported yet");
+ }
+ }
+}
diff --git a/inlong-sort/sort-formats/format-row/format-json-v1.18/src/main/java/org/apache/inlong/sort/formats/json/utils/FormatJsonUtil.java b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/main/java/org/apache/inlong/sort/formats/json/utils/FormatJsonUtil.java
new file mode 100644
index 00000000000..8223129964b
--- /dev/null
+++ b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/main/java/org/apache/inlong/sort/formats/json/utils/FormatJsonUtil.java
@@ -0,0 +1,162 @@
+/*
+ * 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.inlong.sort.formats.json.utils;
+
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.formats.json.JsonFormatOptions.MapNullKeyMode;
+import org.apache.flink.formats.json.RowDataToJsonConverters;
+import org.apache.flink.formats.json.RowDataToJsonConverters.RowDataToJsonConverter;
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.BinaryType;
+import org.apache.flink.table.types.logical.BooleanType;
+import org.apache.flink.table.types.logical.CharType;
+import org.apache.flink.table.types.logical.DateType;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.DoubleType;
+import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LocalZonedTimestampType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.SmallIntType;
+import org.apache.flink.table.types.logical.TimeType;
+import org.apache.flink.table.types.logical.TimestampType;
+import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.VarBinaryType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+import java.util.Map;
+
+import static org.apache.inlong.sort.protocol.constant.DataTypeConstants.DEFAULT_DECIMAL_PRECISION;
+import static org.apache.inlong.sort.protocol.constant.DataTypeConstants.DEFAULT_DECIMAL_SCALE;
+import static org.apache.inlong.sort.protocol.constant.DataTypeConstants.ORACLE_TIMESTAMP_TIME_ZONE;
+
+public class FormatJsonUtil {
+
+ public static final Map SQL_TYPE_2_FLINK_TYPE_MAPPING =
+ ImmutableMap.builder()
+ .put(java.sql.Types.CHAR, new CharType())
+ .put(java.sql.Types.VARCHAR, new VarCharType())
+ .put(java.sql.Types.SMALLINT, new SmallIntType())
+ .put(java.sql.Types.INTEGER, new IntType())
+ .put(java.sql.Types.BIGINT, new BigIntType())
+ .put(java.sql.Types.REAL, new FloatType())
+ .put(java.sql.Types.DOUBLE, new DoubleType())
+ .put(java.sql.Types.FLOAT, new FloatType())
+ .put(java.sql.Types.DECIMAL, new DecimalType(DEFAULT_DECIMAL_PRECISION, DEFAULT_DECIMAL_SCALE))
+ .put(java.sql.Types.NUMERIC, new DecimalType(DEFAULT_DECIMAL_PRECISION, DEFAULT_DECIMAL_SCALE))
+ .put(java.sql.Types.BIT, new BooleanType())
+ .put(java.sql.Types.TIME, new TimeType())
+ .put(java.sql.Types.TIME_WITH_TIMEZONE, new TimeType())
+ .put(java.sql.Types.TIMESTAMP_WITH_TIMEZONE, new LocalZonedTimestampType())
+ .put(ORACLE_TIMESTAMP_TIME_ZONE, new LocalZonedTimestampType())
+ .put(java.sql.Types.TIMESTAMP, new TimestampType())
+ .put(java.sql.Types.BINARY, new BinaryType())
+ .put(java.sql.Types.VARBINARY, new VarBinaryType())
+ .put(java.sql.Types.BLOB, new VarBinaryType())
+ .put(java.sql.Types.CLOB, new VarBinaryType())
+ .put(java.sql.Types.DATE, new DateType())
+ .put(java.sql.Types.BOOLEAN, new BooleanType())
+ .put(java.sql.Types.LONGNVARCHAR, new VarCharType())
+ .put(java.sql.Types.LONGVARBINARY, new VarCharType())
+ .put(java.sql.Types.LONGVARCHAR, new VarCharType())
+ .put(java.sql.Types.ARRAY, new VarCharType())
+ .put(java.sql.Types.NCHAR, new CharType())
+ .put(java.sql.Types.NCLOB, new VarBinaryType())
+ .put(java.sql.Types.TINYINT, new TinyIntType())
+ .put(java.sql.Types.OTHER, new VarCharType())
+ .build();
+ public static final Map SQL_TYPE_2_SPARK_SUPPORTED_FLINK_TYPE_MAPPING =
+ ImmutableMap.builder()
+ .put(java.sql.Types.CHAR, new CharType())
+ .put(java.sql.Types.VARCHAR, new VarCharType())
+ .put(java.sql.Types.SMALLINT, new SmallIntType())
+ .put(java.sql.Types.INTEGER, new IntType())
+ .put(java.sql.Types.BIGINT, new BigIntType())
+ .put(java.sql.Types.REAL, new FloatType())
+ .put(java.sql.Types.DOUBLE, new DoubleType())
+ .put(java.sql.Types.FLOAT, new FloatType())
+ .put(java.sql.Types.DECIMAL, new DecimalType(DEFAULT_DECIMAL_PRECISION, DEFAULT_DECIMAL_SCALE))
+ .put(java.sql.Types.NUMERIC, new DecimalType(DEFAULT_DECIMAL_PRECISION, DEFAULT_DECIMAL_SCALE))
+ .put(java.sql.Types.BIT, new BooleanType())
+ .put(java.sql.Types.TIME, new VarCharType())
+ .put(java.sql.Types.TIMESTAMP_WITH_TIMEZONE, new LocalZonedTimestampType())
+ .put(ORACLE_TIMESTAMP_TIME_ZONE, new LocalZonedTimestampType())
+ .put(java.sql.Types.TIMESTAMP, new LocalZonedTimestampType())
+ .put(java.sql.Types.BINARY, new BinaryType())
+ .put(java.sql.Types.VARBINARY, new VarBinaryType())
+ .put(java.sql.Types.BLOB, new VarBinaryType())
+ .put(java.sql.Types.DATE, new DateType())
+ .put(java.sql.Types.BOOLEAN, new BooleanType())
+ .put(java.sql.Types.LONGNVARCHAR, new VarCharType())
+ .put(java.sql.Types.LONGVARBINARY, new VarCharType())
+ .put(java.sql.Types.LONGVARCHAR, new VarCharType())
+ .put(java.sql.Types.ARRAY, new VarCharType())
+ .put(java.sql.Types.NCHAR, new CharType())
+ .put(java.sql.Types.NCLOB, new VarBinaryType())
+ .put(java.sql.Types.TINYINT, new TinyIntType())
+ .put(java.sql.Types.OTHER, new VarCharType())
+ .build();
+ public static final Map DEBEZIUM_TYPE_2_FLINK_TYPE_MAPPING =
+ ImmutableMap.builder()
+ .put("BOOLEAN", new BooleanType())
+ .put("INT8", new TinyIntType())
+ .put("INT16", new SmallIntType())
+ .put("INT32", new IntType())
+ .put("INT64", new BigIntType())
+ .put("FLOAT32", new FloatType())
+ .put("FLOAT64", new DoubleType())
+ .put("STRING", new VarCharType())
+ .put("BYTES", new VarBinaryType())
+ .build();
+
+ public static RowDataToJsonConverter rowDataToJsonConverter(DataType physicalRowDataType) {
+ return rowDataToJsonConverter(TimestampFormat.SQL, null, physicalRowDataType);
+ }
+
+ public static RowDataToJsonConverter rowDataToJsonConverter(TimestampFormat timestampFormat,
+ String mapNullKeyLiteral,
+ DataType physicalRowDataType) {
+ return rowDataToJsonConverter(timestampFormat, MapNullKeyMode.DROP, mapNullKeyLiteral, physicalRowDataType);
+ }
+
+ public static RowDataToJsonConverter rowDataToJsonConverter(TimestampFormat timestampFormat,
+ MapNullKeyMode mapNullKeyMode,
+ String mapNullKeyLiteral, DataType physicalRowDataType) {
+ return new RowDataToJsonConverters(timestampFormat, mapNullKeyMode, mapNullKeyLiteral)
+ .createConverter(physicalRowDataType.getLogicalType());
+ }
+
+ public static RowDataToJsonConverter rowDataToJsonConverter(LogicalType rowType) {
+ return rowDataToJsonConverter(TimestampFormat.SQL, null, rowType);
+ }
+
+ public static RowDataToJsonConverter rowDataToJsonConverter(TimestampFormat timestampFormat,
+ String mapNullKeyLiteral,
+ LogicalType rowType) {
+ return rowDataToJsonConverter(timestampFormat, MapNullKeyMode.DROP, mapNullKeyLiteral, rowType);
+ }
+
+ public static RowDataToJsonConverter rowDataToJsonConverter(TimestampFormat timestampFormat,
+ MapNullKeyMode mapNullKeyMode,
+ String mapNullKeyLiteral, LogicalType rowType) {
+ return new RowDataToJsonConverters(timestampFormat, mapNullKeyMode, mapNullKeyLiteral)
+ .createConverter(rowType);
+ }
+}
diff --git a/inlong-sort/sort-formats/format-row/format-json-v1.18/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
new file mode 100644
index 00000000000..7ce31e31ed4
--- /dev/null
+++ b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.inlong.sort.formats.json.canal.CanalJsonEnhancedFormatFactory
diff --git a/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/java/org/apache/inlong/sort/formats/json/canal/CanalJsonEnhancedFormatFactoryTest.java b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/java/org/apache/inlong/sort/formats/json/canal/CanalJsonEnhancedFormatFactoryTest.java
new file mode 100644
index 00000000000..fe54e39e1ab
--- /dev/null
+++ b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/java/org/apache/inlong/sort/formats/json/canal/CanalJsonEnhancedFormatFactoryTest.java
@@ -0,0 +1,138 @@
+/*
+ * 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.inlong.sort.formats.json.canal;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.formats.json.JsonFormatOptions;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.TestDynamicTableFactory;
+import org.apache.flink.table.runtime.connector.sink.SinkRuntimeProviderContext;
+import org.apache.flink.table.runtime.connector.source.ScanRuntimeProviderContext;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import static org.apache.flink.table.factories.utils.FactoryMocks.PHYSICAL_DATA_TYPE;
+import static org.apache.flink.table.factories.utils.FactoryMocks.PHYSICAL_TYPE;
+import static org.apache.flink.table.factories.utils.FactoryMocks.SCHEMA;
+import static org.apache.flink.table.factories.utils.FactoryMocks.createTableSink;
+import static org.apache.flink.table.factories.utils.FactoryMocks.createTableSource;
+import static org.junit.Assert.assertEquals;
+
+public class CanalJsonEnhancedFormatFactoryTest {
+
+ private static final InternalTypeInfo ROW_TYPE_INFO =
+ InternalTypeInfo.of(PHYSICAL_TYPE);
+
+ @Test
+ public void testUserDefinedOptions() {
+ final Map tableOptions =
+ getModifiedOptions(opts -> {
+ opts.put("canal-json-inlong.map-null-key.mode", "LITERAL");
+ opts.put("canal-json-inlong.map-null-key.literal", "nullKey");
+ opts.put("canal-json-inlong.ignore-parse-errors", "true");
+ opts.put("canal-json-inlong.timestamp-format.standard", "ISO-8601");
+ opts.put("canal-json-inlong.database.include", "mydb");
+ opts.put("canal-json-inlong.table.include", "mytable");
+ opts.put("canal-json-inlong.map-null-key.mode", "LITERAL");
+ opts.put("canal-json-inlong.map-null-key.literal", "nullKey");
+ opts.put("canal-json-inlong.encode.decimal-as-plain-number", "true");
+ });
+
+ // test Deser
+ CanalJsonEnhancedDeserializationSchema expectedDeser =
+ CanalJsonEnhancedDeserializationSchema.builder(
+ PHYSICAL_DATA_TYPE, Collections.emptyList(), ROW_TYPE_INFO)
+ .setIgnoreParseErrors(true)
+ .setTimestampFormat(TimestampFormat.ISO_8601)
+ .setDatabase("mydb")
+ .setTable("mytable")
+ .build();
+ DeserializationSchema actualDeser = createDeserializationSchema(tableOptions);
+ assertEquals(expectedDeser, actualDeser);
+
+ // test Ser
+ CanalJsonEnhancedSerializationSchema expectedSer =
+ new CanalJsonEnhancedSerializationSchema(
+ PHYSICAL_DATA_TYPE,
+ new ArrayList<>(),
+ TimestampFormat.ISO_8601,
+ JsonFormatOptions.MapNullKeyMode.LITERAL,
+ "nullKey",
+ true);
+ SerializationSchema actualSer = createSerializationSchema(tableOptions);
+ assertEquals(expectedSer, actualSer);
+ }
+
+ // ------------------------------------------------------------------------
+ // Public Tools
+ // ------------------------------------------------------------------------
+
+ public static DeserializationSchema createDeserializationSchema(
+ Map options) {
+ DynamicTableSource source = createTableSource(SCHEMA, options);
+
+ assert source instanceof TestDynamicTableFactory.DynamicTableSourceMock;
+ TestDynamicTableFactory.DynamicTableSourceMock scanSourceMock =
+ (TestDynamicTableFactory.DynamicTableSourceMock) source;
+
+ return scanSourceMock.valueFormat.createRuntimeDecoder(
+ ScanRuntimeProviderContext.INSTANCE, PHYSICAL_DATA_TYPE);
+ }
+
+ public static SerializationSchema createSerializationSchema(
+ Map options) {
+ DynamicTableSink sink = createTableSink(SCHEMA, options);
+
+ assert sink instanceof TestDynamicTableFactory.DynamicTableSinkMock;
+ TestDynamicTableFactory.DynamicTableSinkMock sinkMock =
+ (TestDynamicTableFactory.DynamicTableSinkMock) sink;
+
+ return sinkMock.valueFormat.createRuntimeEncoder(
+ new SinkRuntimeProviderContext(false), PHYSICAL_DATA_TYPE);
+ }
+
+ /**
+ * Returns the full options modified by the given consumer {@code optionModifier}.
+ *
+ * @param optionModifier Consumer to modify the options
+ */
+ public static Map getModifiedOptions(Consumer> optionModifier) {
+ Map options = getAllOptions();
+ optionModifier.accept(options);
+ return options;
+ }
+
+ private static Map getAllOptions() {
+ final Map options = new HashMap<>();
+ options.put("connector", TestDynamicTableFactory.IDENTIFIER);
+ options.put("target", "MyTarget");
+ options.put("buffer-size", "1000");
+ options.put("format", "canal-json-inlong");
+ return options;
+ }
+}
diff --git a/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/java/org/apache/inlong/sort/formats/json/canal/CanalJsonEnhancedSerDeSchemaTest.java b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/java/org/apache/inlong/sort/formats/json/canal/CanalJsonEnhancedSerDeSchemaTest.java
new file mode 100644
index 00000000000..f730340f4cc
--- /dev/null
+++ b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/java/org/apache/inlong/sort/formats/json/canal/CanalJsonEnhancedSerDeSchemaTest.java
@@ -0,0 +1,213 @@
+/*
+ * 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.inlong.sort.formats.json.canal;
+
+import org.apache.inlong.sort.formats.json.canal.CanalJsonEnhancedDecodingFormat.ReadableMetadata;
+import org.apache.inlong.sort.formats.json.canal.CanalJsonEnhancedEncodingFormat.WriteableMetadata;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.formats.json.JsonFormatOptions;
+import org.apache.flink.formats.json.canal.CanalJsonFormatOptions;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.utils.DataTypeUtils;
+import org.apache.flink.util.Collector;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.junit.Assert.assertEquals;
+
+public class CanalJsonEnhancedSerDeSchemaTest {
+
+ public static final String DATABASE = "TEST";
+
+ public static final String TABLE = "TEST";
+
+ public static final ResolvedSchema SCHEMA =
+ ResolvedSchema.of(
+ Column.metadata("database", DataTypes.BOOLEAN(), "database", false),
+ Column.physical("id", DataTypes.BIGINT()),
+ Column.physical("name", DataTypes.STRING()),
+ Column.metadata("table", DataTypes.BOOLEAN(), "table", false),
+ Column.metadata("sql_type",
+ DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()), "sql-type", false),
+ Column.metadata("pk_names",
+ DataTypes.ARRAY(DataTypes.STRING()), "pk-names", false),
+ Column.metadata("ingestion_timestamp",
+ DataTypes.TIMESTAMP_LTZ(3), "ingestion-timestamp", false),
+ Column.metadata("event_timestamp",
+ DataTypes.TIMESTAMP_LTZ(3), "event-timestamp", false),
+ Column.metadata("op_type", DataTypes.STRING(), "op-type", false),
+ Column.metadata("is_ddl", DataTypes.BOOLEAN(), "is-ddl", false),
+ Column.metadata("mysql_type",
+ DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()), "mysql-type", false),
+ Column.metadata("batch_id", DataTypes.BIGINT(), "batch-id", false),
+ Column.metadata("update_before",
+ DataTypes.ARRAY(DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING())),
+ "update-before", false));
+
+ public static final DataType PHYSICAL_DATA_TYPE = SCHEMA.toPhysicalRowDataType();
+
+ public static final List READABLE_METADATA =
+ Stream.of(
+ ReadableMetadata.DATABASE,
+ ReadableMetadata.TABLE,
+ ReadableMetadata.SQL_TYPE,
+ ReadableMetadata.PK_NAMES,
+ ReadableMetadata.INGESTION_TIMESTAMP,
+ ReadableMetadata.EVENT_TIMESTAMP,
+ ReadableMetadata.OP_TYPE,
+ ReadableMetadata.IS_DDL,
+ ReadableMetadata.MYSQL_TYPE,
+ ReadableMetadata.BATCH_ID,
+ ReadableMetadata.UPDATE_BEFORE).collect(Collectors.toList());
+
+ public static final List WRITEABLE_METADATA =
+ Stream.of(
+ WriteableMetadata.DATABASE,
+ WriteableMetadata.TABLE,
+ WriteableMetadata.SQL_TYPE,
+ WriteableMetadata.PK_NAMES,
+ WriteableMetadata.INGESTION_TIMESTAMP,
+ WriteableMetadata.EVENT_TIMESTAMP,
+ WriteableMetadata.OP_TYPE,
+ WriteableMetadata.IS_DDL,
+ WriteableMetadata.MYSQL_TYPE,
+ WriteableMetadata.BATCH_ID,
+ WriteableMetadata.UPDATE_BEFORE).collect(Collectors.toList());
+
+ @Test
+ public void testSerDeWithMetadata() throws Exception {
+ List lines = readLines("canal-json-inlong-data.txt");
+ DeserializationSchema deserializationSchema = createCanalJsonDeserializationSchema(
+ PHYSICAL_DATA_TYPE, READABLE_METADATA);
+ // deserialize
+ SimpleCollector out = new SimpleCollector();
+ for (String line : lines) {
+ deserializationSchema.deserialize(line.getBytes(StandardCharsets.UTF_8), out);
+ }
+ List res = out.result();
+
+ // serialize
+ SerializationSchema serializationSchema = createCanalJsonSerializationSchema(
+ PHYSICAL_DATA_TYPE, WRITEABLE_METADATA);
+ serializationSchema.open(null);
+ for (int i = 0; i < lines.size(); i++) {
+ String json = new String(serializationSchema.serialize(res.get(i)), StandardCharsets.UTF_8);
+ compareJson(json, lines.get(i));
+ }
+ }
+
+ // =======================================Utils=======================================================
+
+ private CanalJsonEnhancedDeserializationSchema createCanalJsonDeserializationSchema(
+ DataType physicalDataType, List requestedMetadata) {
+ final DataType producedDataType =
+ DataTypeUtils.appendRowFields(
+ physicalDataType,
+ requestedMetadata.stream()
+ .map(m -> DataTypes.FIELD(m.key, m.dataType))
+ .collect(Collectors.toList()));
+ return CanalJsonEnhancedDeserializationSchema.builder(
+ PHYSICAL_DATA_TYPE,
+ requestedMetadata,
+ InternalTypeInfo.of(producedDataType.getLogicalType()))
+ .setDatabase(DATABASE)
+ .setTable(TABLE)
+ .setIgnoreParseErrors(JsonFormatOptions.IGNORE_PARSE_ERRORS.defaultValue())
+ .setTimestampFormat(TimestampFormat.valueOf(CanalJsonFormatOptions.TIMESTAMP_FORMAT.defaultValue()))
+ .build();
+ }
+
+ private CanalJsonEnhancedSerializationSchema createCanalJsonSerializationSchema(
+ DataType physicalDataType, List requestedMetadata) {
+ return new CanalJsonEnhancedSerializationSchema(
+ physicalDataType,
+ requestedMetadata,
+ TimestampFormat.valueOf(CanalJsonFormatOptions.TIMESTAMP_FORMAT.defaultValue()),
+ JsonFormatOptions.MapNullKeyMode.valueOf(CanalJsonFormatOptions.JSON_MAP_NULL_KEY_MODE.defaultValue()),
+ CanalJsonFormatOptions.JSON_MAP_NULL_KEY_LITERAL.defaultValue(),
+ JsonFormatOptions.ENCODE_DECIMAL_AS_PLAIN_NUMBER.defaultValue());
+ }
+
+ private static List readLines(String resource) throws IOException {
+ final URL url = CanalJsonEnhancedSerDeSchemaTest.class.getClassLoader().getResource(resource);
+ assert url != null;
+ Path path = new File(url.getFile()).toPath();
+ return Files.readAllLines(path);
+ }
+
+ private static List readRowDatas(String resource) throws IOException, ClassNotFoundException {
+ final URL url = CanalJsonEnhancedSerDeSchemaTest.class.getClassLoader().getResource(resource);
+ assert url != null;
+ Path path = new File(url.getFile()).toPath();
+ ObjectInputStream in = new ObjectInputStream(new FileInputStream(path.toString()));
+ return (List) in.readObject();
+ }
+
+ public void compareJson(String json1, String json2) throws JsonProcessingException {
+ ObjectMapper objectMapper = new ObjectMapper();
+ JsonNode node1 = objectMapper.readTree(json1);
+ JsonNode node2 = objectMapper.readTree(json2);
+ assertEquals(node1, node2);
+ }
+
+ private static class SimpleCollector implements Collector {
+
+ private List list = new ArrayList<>();
+
+ @Override
+ public void collect(RowData record) {
+ list.add(record);
+ }
+
+ @Override
+ public void close() {
+ // do nothing
+ }
+
+ public List result() {
+ List newList = new ArrayList<>();
+ list.forEach(row -> newList.add(row));
+ list.clear();
+ return newList;
+ }
+ }
+
+}
diff --git a/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/java/org/apache/inlong/sort/formats/json/canal/CanalJsonSerializationTest.java b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/java/org/apache/inlong/sort/formats/json/canal/CanalJsonSerializationTest.java
new file mode 100644
index 00000000000..a70c7bba2ab
--- /dev/null
+++ b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/java/org/apache/inlong/sort/formats/json/canal/CanalJsonSerializationTest.java
@@ -0,0 +1,87 @@
+/*
+ * 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.inlong.sort.formats.json.canal;
+
+import org.apache.inlong.sort.protocol.ddl.enums.AlterType;
+import org.apache.inlong.sort.protocol.ddl.enums.PositionType;
+import org.apache.inlong.sort.protocol.ddl.expressions.AlterColumn;
+import org.apache.inlong.sort.protocol.ddl.expressions.Column;
+import org.apache.inlong.sort.protocol.ddl.expressions.Position;
+import org.apache.inlong.sort.protocol.ddl.operations.AlterOperation;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Test for {@link CanalJson}.
+ */
+public class CanalJsonSerializationTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CanalJsonSerializationTest.class);
+
+ private final ObjectMapper objectMapper = new ObjectMapper();
+
+ @Test
+ public void testCanalJsonSerialization() {
+
+ List alterColumns = new ArrayList<>();
+
+ Column column = new Column("columnDataType.getColumnName()", new ArrayList<>(),
+ 1,
+ new Position(PositionType.FIRST, null), true, "23",
+ "23");
+
+ alterColumns.add(new AlterColumn(AlterType.ADD_COLUMN, column, null));
+
+ AlterOperation alterOperation = new AlterOperation(alterColumns);
+
+ CanalJson canalJson = CanalJson.builder()
+ .data(null)
+ .es(0)
+ .table("table")
+ .type("type")
+ .database("database")
+ .ts(0)
+ .sql("sql")
+ .mysqlType(null)
+ .sqlType(null)
+ .pkNames(null)
+ .schema("schema")
+ .oracleType(null)
+ .operation(alterOperation)
+ .incremental(false)
+ .build();
+
+ ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ try {
+ String writeValueAsString = OBJECT_MAPPER.writeValueAsString(canalJson);
+ LOG.info(writeValueAsString);
+ objectMapper.readValue(writeValueAsString, CanalJson.class);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+}
diff --git a/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/java/org/apache/inlong/sort/formats/json/canal/DebeziumJsonSerializationTest.java b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/java/org/apache/inlong/sort/formats/json/canal/DebeziumJsonSerializationTest.java
new file mode 100644
index 00000000000..0ca38431e39
--- /dev/null
+++ b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/java/org/apache/inlong/sort/formats/json/canal/DebeziumJsonSerializationTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.inlong.sort.formats.json.canal;
+
+import org.apache.inlong.sort.formats.json.debezium.DebeziumJson;
+import org.apache.inlong.sort.protocol.ddl.enums.AlterType;
+import org.apache.inlong.sort.protocol.ddl.enums.PositionType;
+import org.apache.inlong.sort.protocol.ddl.expressions.AlterColumn;
+import org.apache.inlong.sort.protocol.ddl.expressions.Column;
+import org.apache.inlong.sort.protocol.ddl.expressions.Position;
+import org.apache.inlong.sort.protocol.ddl.operations.AlterOperation;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+/**
+ * Test for {@link DebeziumJson}.
+ */
+public class DebeziumJsonSerializationTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CanalJsonSerializationTest.class);
+
+ private final ObjectMapper objectMapper = new ObjectMapper();
+
+ @Test
+ public void testDebeziumJsonSerialization() {
+
+ List alterColumns = new ArrayList<>();
+
+ Column column = new Column("columnDataType.getColumnName()", new ArrayList<>(),
+ 1,
+ new Position(PositionType.FIRST, null), true, "23",
+ "23");
+
+ alterColumns.add(new AlterColumn(AlterType.ADD_COLUMN, column, null));
+
+ AlterOperation alterOperation = new AlterOperation(alterColumns);
+
+ DebeziumJson debeziumJson = DebeziumJson.builder().source(null)
+ .dataSourceName("dataSourceName")
+ .tableChange(null).incremental(false).build();
+
+ debeziumJson.setDdl("");
+ debeziumJson.setOperation(alterOperation);
+ debeziumJson.setAfter(new HashMap<>());
+
+ ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ try {
+ String writeValueAsString = OBJECT_MAPPER.writeValueAsString(debeziumJson);
+ LOG.info(writeValueAsString);
+ objectMapper.readValue(writeValueAsString, DebeziumJson.class);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+}
diff --git a/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/resources/canal-json-inlong-data.txt b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/resources/canal-json-inlong-data.txt
new file mode 100644
index 00000000000..82a3767be03
--- /dev/null
+++ b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/resources/canal-json-inlong-data.txt
@@ -0,0 +1,3 @@
+{"data":[{"id":2,"name":"xixi"}],"type":"INSERT","pkNames":["id"],"database":"TEST","mysqlType":{"name":"VARCHAR(63)","id":"BIGINT(20)"},"opType":"INSERT","es":0,"batchId":0,"sqlType":{"name":12,"id":-5},"updateBefore":null,"ts":1651208731718,"isDdl":false,"table":"TEST"}
+{"data":[{"id":1,"name":"oooooo"}],"type":"INSERT","pkNames":["id"],"database":"TEST","mysqlType":{"name":"VARCHAR(63)","id":"BIGINT(20)"},"opType":"INSERT","es":0,"batchId":1,"sqlType":{"name":12,"id":-5},"updateBefore":null,"ts":1651208731717,"isDdl":false,"table":"TEST"}
+{"data":[{"id":3,"name":"HAHA"}],"type":"INSERT","pkNames":["id"],"database":"TEST","mysqlType":{"name":"VARCHAR(63)","id":"BIGINT(20)"},"opType":"INSERT","es":1651208797000,"batchId":2,"sqlType":{"name":12,"id":-5},"updateBefore":null,"ts":1651208799752,"isDdl":false,"table":"TEST"}
\ No newline at end of file
diff --git a/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/resources/log4j2-test.properties b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/resources/log4j2-test.properties
new file mode 100644
index 00000000000..d493c984a32
--- /dev/null
+++ b/inlong-sort/sort-formats/format-row/format-json-v1.18/src/test/resources/log4j2-test.properties
@@ -0,0 +1,32 @@
+################################################################################
+# 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.
+################################################################################
+rootLogger=INFO, testlogger
+
+appender.out.type=Console
+appender.out.name=testlogger
+appender.out.layout.type=PatternLayout
+appender.out.layout.pattern=%-4r [%t] %-5p %c %x - %m%n
+
+# Suppress the irrelevant (wrong) warnings from the Netty channel handler
+logger.flinknetty=ERROR, testlogger
+logger.flinknetty.name=org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
+logger.flinknetty.additivity=false
+# Resource leak detector only works with logging enabled at error level
+logger.leakdetector=ERROR, testlogger
+logger.leakdetector.name=org.apache.flink.shaded.netty4.io.netty.util.ResourceLeakDetector
+logger.leakdetector.additivity=false
diff --git a/inlong-sort/sort-formats/format-row/pom.xml b/inlong-sort/sort-formats/format-row/pom.xml
index cca64ed2ca6..22207262a05 100644
--- a/inlong-sort/sort-formats/format-row/pom.xml
+++ b/inlong-sort/sort-formats/format-row/pom.xml
@@ -66,5 +66,11 @@
format-json-v1.15
+
+ v1.18
+
+ format-json-v1.18
+
+
diff --git a/inlong-sort/sort-formats/pom.xml b/inlong-sort/sort-formats/pom.xml
index 4f22664ae3f..a1e69fb0c7a 100644
--- a/inlong-sort/sort-formats/pom.xml
+++ b/inlong-sort/sort-formats/pom.xml
@@ -30,12 +30,6 @@
pom
Apache InLong - Sort Formats
-
- format-common
- format-row
- format-rowdata
-
-
${project.parent.parent.basedir}
1C
@@ -103,22 +97,6 @@
test
-
-
- org.apache.flink
- flink-table-common
- ${flink.version}
- provided
-
-
-
- org.apache.flink
- flink-table-common
- ${flink.version}
- test-jar
- test
-
-
@@ -209,4 +187,87 @@
https://git.code.oa.com/flink/flink-formats
+
+
+ v1.13
+
+ true
+
+
+ format-common
+ format-row
+ format-rowdata
+
+
+
+
+
+ org.apache.flink
+ flink-table-common
+ ${flink.version}
+ provided
+
+
+
+ org.apache.flink
+ flink-table-common
+ ${flink.version}
+ test-jar
+ test
+
+
+
+
+
+ v1.15
+
+ format-common
+ format-row
+ format-rowdata
+
+
+
+
+
+ org.apache.flink
+ flink-table-common
+ ${flink.version}
+ provided
+
+
+
+ org.apache.flink
+ flink-table-common
+ ${flink.version}
+ test-jar
+ test
+
+
+
+
+
+ v1.18
+
+ format-common
+ format-row/format-json-v1.18
+
+
+
+
+ org.apache.flink
+ flink-table-common
+ ${flink.version}
+ provided
+
+
+ org.apache.flink
+ flink-table-common
+ ${flink.version}
+ test-jar
+ test
+
+
+
+
+