private void flush() throws Exception {
        LOG.info("Flushing IcebergMiniBatchGroupOperator.");
        // Emit the rows group by partition
        // scan range key, this range key contains all one partition data
        partitionGroupBuffer.scanPartitions(tuple -> collector.collect(tuple.f1));
        partitionGroupBuffer.clear();
    }
This Operator has two functional above:
1. Aggregating calculation in advance, reduce downstream computational workload
2. Clustering data according to partition, reduce memory pressure caused by opening multiple writers downstream
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/IcebergMiniBatchGroupOperator.java