public static PreAggPartitionGroupBuffer preAggInstance(
            FieldGetter[] fieldsGetter,
            Schema deleteSchema,
            Schema rowSchema,
            PartitionKey partitionKey,
            BufferType type) {
        // note: here because `NestedField` does not override equals function, so can not indexOf by `NestedField`
        int[] equalityFieldIndex = deleteSchema.columns().stream()
                .map(field -> rowSchema.columns()
                        .stream()
                        .map(NestedField::fieldId)
                        .collect(Collectors.toList())
                        .indexOf(field.fieldId()))
                .sorted()
                .mapToInt(Integer::valueOf)
                .toArray();
        // do some check, check whether index is legal. can not be null and unique, and number in fields range.
        Preconditions.checkArgument(
                Arrays.stream(equalityFieldIndex)
                        .allMatch(index -> index >= 0 && index < fieldsGetter.length),
                String.format("Any equality field index (%s) should in a legal range.",
                        Arrays.toString(equalityFieldIndex)));

        KVBuffer<Tuple2<String, RowData>, RowData> buffer = null;
        if (BufferType.ROCKSDB.equals(type)) {
            TypeInformation<Tuple2<String, RowData>> keyTypeInfo = new TupleTypeInfo(
                    BasicTypeInfo.STRING_TYPE_INFO,
                    InternalTypeInfo.of(FlinkSchemaUtil.convert(deleteSchema)));
            TypeInformation<RowData> valueTypeInfo = InternalTypeInfo.of(FlinkSchemaUtil.convert(rowSchema));
            buffer = new RocksDBKVBuffer<>(
                    keyTypeInfo.createSerializer(new ExecutionConfig()),
                    valueTypeInfo.createSerializer(new ExecutionConfig()),
                    System.getProperty("java.io.tmpdir") + File.separator + "mini_batch");
        } else {
            TypeInformation<RowData> keyTypeInfo = InternalTypeInfo.of(FlinkSchemaUtil.convert(deleteSchema));
            buffer = new SortedHeapKVBuffer<>(
                    new PreAggPartitionConvertor(keyTypeInfo.createSerializer(new ExecutionConfig())));
        }
        return new PreAggPartitionGroupBuffer(buffer, partitionKey, rowSchema, fieldsGetter, equalityFieldIndex);
    }
Initialize {@link PreAggPartitionGroupBuffer}

@param fieldsGetter function to get object from {@link RowData}
@param deleteSchema equality fields schema
@param rowSchema row data schema
@param partitionKey partition key
@param type buffer type to store buffer data
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/collections/PartitionGroupBuffer.java