@Override
    public void setRuntimeContext(RuntimeContext runtimeCtx) {
        final TypeSerializer<RowData> typeSerializer =
                rowDataTypeInfo.createSerializer(runtimeCtx.getExecutionConfig());
        valueTransform =
                runtimeCtx.getExecutionConfig().isObjectReuseEnabled() ? typeSerializer::copy : Function.identity();
        SerializeConfig.getGlobalInstance().put(BinaryStringData.class, new BinaryStringDataSerializer());
        SerializeConfig.getGlobalInstance().put(DecimalData.class, new DecimalDataSerializer());
        SerializeConfig.getGlobalInstance().put(TimestampData.class, new TimestampDataSerializer());
    }
copy from {@link com.starrocks.connector.flink.row.sink.StarRocksTableRowTransformer}
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/table/sink/table/StarRocksTableRowTransformer.java