@Override
    public byte[] serialize(RowData row) {
        try {
            // physical data injection
            GenericRowData physicalData = new GenericRowData(physicalFieldGetter.length);
            IntStream.range(0, physicalFieldGetter.length)
                    .forEach(targetField -> physicalData.setField(targetField,
                            physicalFieldGetter[targetField].getFieldOrNull(row)));
            ArrayData arrayData = new GenericArrayData(new RowData[]{physicalData});
            reuse.setField(0, arrayData);

            // mete data injection
            StringData opType = rowKind2String(row.getRowKind());
            reuse.setField(1, opType);
            if (typeIndex != -1) {
                IntStream.range(0, wirteableMetadataFieldGetter.length)
                        .forEach(metaIndex -> {
                            if (metaIndex < typeIndex) {
                                reuse.setField(metaIndex + 2,
                                        wirteableMetadataFieldGetter[metaIndex].getFieldOrNull(row));
                            } else if (metaIndex > typeIndex) {
                                reuse.setField(metaIndex + 1,
                                        wirteableMetadataFieldGetter[metaIndex].getFieldOrNull(row));
                            }
                        });
            } else {
                IntStream.range(0, wirteableMetadataFieldGetter.length)
                        .forEach(metaIndex -> reuse
                                .setField(metaIndex + 2, wirteableMetadataFieldGetter[metaIndex].getFieldOrNull(row)));
            }
            return jsonSerializer.serialize(reuse);
        } catch (Throwable t) {
            throw new RuntimeException("Could not serialize row '" + row + "'.", t);
        }
    }
Serialize the row with ignore the {@link WriteableMetadata#TYPE}
inlong-sort/sort-formats/format-row/format-json-v1.13/src/main/java/org/apache/inlong/sort/formats/json/canal/CanalJsonEnhancedSerializationSchema.java