@Override
    public void deserialize(@Nullable byte[] message, Collector<RowData> out) throws IOException {
        if (message == null || message.length == 0) {
            return;
        }
        try {
            final JsonNode root = jsonDeserializer.deserializeToJsonNode(message);
            if (database != null) {
                if (!databasePattern
                        .matcher(root.get(ReadableMetadata.DATABASE.key).asText())
                        .matches()) {
                    return;
                }
            }
            if (table != null) {
                if (!tablePattern
                        .matcher(root.get(ReadableMetadata.TABLE.key).asText())
                        .matches()) {
                    return;
                }
            }
            final GenericRowData row = (GenericRowData) jsonDeserializer.convertToRowData(root);
            String type = row.getString(2).toString(); // "type" field
            if (OP_INSERT.equals(type)) {
                // "data" field is an array of row, contains inserted rows
                ArrayData data = row.getArray(0);
                for (int i = 0; i < data.size(); i++) {
                    GenericRowData insert = (GenericRowData) data.getRow(i, fieldCount);
                    insert.setRowKind(RowKind.INSERT);
                    emitRow(row, insert, out);
                }
            } else if (OP_UPDATE.equals(type)) {
                // "data" field is an array of row, contains new rows
                ArrayData data = row.getArray(0);
                // "old" field is an array of row, contains old values
                ArrayData old = row.getArray(1);
                for (int i = 0; i < data.size(); i++) {
                    // the underlying JSON deserialization schema always produce GenericRowData.
                    GenericRowData after = (GenericRowData) data.getRow(i, fieldCount);
                    GenericRowData before = (GenericRowData) old.getRow(i, fieldCount);
                    final JsonNode oldField = root.get(FIELD_OLD);
                    for (int f = 0; f < fieldCount; f++) {
                        if (before.isNullAt(f) && oldField.findValue(fieldNames.get(f)) == null) {
                            // fields in "old" (before) means the fields are changed
                            // fields not in "old" (before) means the fields are not changed
                            // so we just copy the not changed fields into before
                            before.setField(f, after.getField(f));
                        }
                    }
                    before.setRowKind(RowKind.UPDATE_BEFORE);
                    after.setRowKind(RowKind.UPDATE_AFTER);
                    emitRow(row, before, out);
                    emitRow(row, after, out);
                }
            } else if (OP_DELETE.equals(type)) {
                // "data" field is an array of row, contains deleted rows
                ArrayData data = row.getArray(0);
                for (int i = 0; i < data.size(); i++) {
                    GenericRowData insert = (GenericRowData) data.getRow(i, fieldCount);
                    insert.setRowKind(RowKind.DELETE);
                    emitRow(row, insert, out);
                }
            } else if (OP_CREATE.equals(type)) {
                // "data" field is null and "type" is "CREATE" which means
                // this is a DDL change event, and we should skip it.
                return;
            } else {
                if (!ignoreParseErrors) {
                    throw new IOException(
                            format(
                                    "Unknown \"type\" value \"%s\". The Canal JSON message is '%s'",
                                    type, new String(message)));
                }
            }
        } catch (Throwable t) {
            // a big try catch to protect the processing.
            if (!ignoreParseErrors) {
                throw new IOException(
                        format("Corrupt Canal JSON message '%s'.", new String(message)), t);
            }
        }
    }
Creates A builder for building a {@link CanalJsonEnhancedDeserializationSchema}.
inlong-sort/sort-formats/format-row/format-json-v1.18/src/main/java/org/apache/inlong/sort/formats/json/canal/CanalJsonEnhancedDeserializationSchema.java