@Override
        public RowData add(RowData row) {
            RowData primaryKey = GenericRowData.of(Arrays.stream(equalityFieldIndex)
                    .boxed()
                    .map(index -> fieldsGetter[index].getFieldOrNull(row))
                    .toArray(Object[]::new));
            partitionKey.partition(lazyRowDataWrapper().wrap(row));
            allPartition.add(partitionKey.toPath());

            if (RowDataUtil.isAccumulateMsg(row)) {
                return ((KVBuffer<Tuple2<String, RowData>, RowData>) buffer).put(
                        new Tuple2<>(partitionKey.toPath(), primaryKey), row);
            } else {
                return ((KVBuffer<Tuple2<String, RowData>, RowData>) buffer).remove(
                        new Tuple2<>(partitionKey.toPath(), primaryKey));
            }
        }
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