public static NonPreAggPartitionGroupBuffer nonPreAggInstance(
            Schema rowSchema,
            PartitionKey partitionKey,
            BufferType type) {
        TypeInformation<Tuple2<String, Long>> keyTypeInfo = new TupleTypeInfo(
                BasicTypeInfo.STRING_TYPE_INFO,
                BasicTypeInfo.LONG_TYPE_INFO);
        TypeInformation<RowData> valueTypeInfo = InternalTypeInfo.of(FlinkSchemaUtil.convert(rowSchema));

        KVBuffer<Tuple2<String, Long>, RowData> buffer = null;
        if (BufferType.ROCKSDB.equals(type)) {
            buffer = new RocksDBKVBuffer<>(
                    keyTypeInfo.createSerializer(new ExecutionConfig()),
                    valueTypeInfo.createSerializer(new ExecutionConfig()),
                    System.getProperty("java.io.tmpdir") + "/mini_batch");
        } else {
            buffer = new SortedHeapKVBuffer<>(new NonPreAggPartitionConvetor());
        }
        return new NonPreAggPartitionGroupBuffer(buffer, partitionKey, rowSchema);
    }
Initialize {@link NonPreAggPartitionGroupBuffer}

@param rowSchema row data schema
@param partitionKey partition key
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/collections/PartitionGroupBuffer.java