@Override
        public Tuple2<String, RowData> upper(byte[] keyPrefix) {
            inputBuffer.setBuffer(keyPrefix);
            try {
                return new Tuple2<>(StringSerializer.INSTANCE.deserialize(inputBuffer), GenericRowData.of());
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
A {@link Convertor} that key represent <partition, primary key>
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/collections/PartitionGroupBuffer.java