private void extractDdlRecord(SourceRecord record, Collector<RowData> out, TableChange tableSchema,
            Struct value) {

        try {
            GenericRowData insert = (GenericRowData) physicalConverter.convert(
                    objectMapper.readTree(value.get(HISTORY_RECORD_FIELD).toString()).get(DDL_FIELD_NAME).asText(),
                    null);
            insert.setRowKind(RowKind.INSERT);
            emit(record, insert, tableSchema, out);
        } catch (Exception e) {
            LOG.error("Failed to extract DDL record {}", record, e);
            throw new RuntimeException(e);
        }

    }
extract the data with the format provided by debezium

@param fieldValue
@param schemaName
@return the extracted data with schema
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/RowDataDebeziumDeserializeSchema.java