private DataStream<RowData> distributeDataStream(
                DataStream<RowData> input,
                List<Integer> equalityFieldIds,
                PartitionSpec partitionSpec,
                Schema iSchema,
                RowType flinkRowType) {
            DistributionMode writeMode = flinkWriteConf.distributionMode();
            LOG.info("Write distribution mode is '{}'", writeMode.modeName());
            switch (writeMode) {
                case NONE:
                    if (equalityFieldIds.isEmpty()) {
                        return input;
                    } else {
                        LOG.info("Distribute rows by equality fields, because there are equality fields set");
                        return input.keyBy(
                                new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
                    }

                case HASH:
                    if (equalityFieldIds.isEmpty()) {
                        if (partitionSpec.isUnpartitioned()) {
                            LOG.warn(
                                    "Fallback to use 'none' distribution mode, because there are no equality fields "
                                            + "set and table is unpartitioned");
                            return input;
                        } else {
                            return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
                        }
                    } else {
                        if (partitionSpec.isUnpartitioned()) {
                            LOG.info(
                                    "Distribute rows by equality fields, because there are equality fields set "
                                            + "and table is unpartitioned");
                            return input.keyBy(
                                    new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
                        } else {
                            for (PartitionField partitionField : partitionSpec.fields()) {
                                Preconditions.checkState(
                                        equalityFieldIds.contains(partitionField.sourceId()),
                                        "In 'hash' distribution mode with equality fields set, partition field '%s' "
                                                + "should be included in equality fields: '%s'",
                                        partitionField,
                                        equalityFieldColumns);
                            }
                            return input.keyBy(new PartitionKeySelector(partitionSpec, iSchema, flinkRowType));
                        }
                    }

                case RANGE:
                    if (equalityFieldIds.isEmpty()) {
                        LOG.warn(
                                "Fallback to use 'none' distribution mode, because there are no equality fields set "
                                        + "and {}=range is not supported yet in flink",
                                WRITE_DISTRIBUTION_MODE);
                        return input;
                    } else {
                        LOG.info(
                                "Distribute rows by equality fields, because there are equality fields set "
                                        + "and{}=range is not supported yet in flink",
                                WRITE_DISTRIBUTION_MODE);
                        return input.keyBy(
                                new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));
                    }

                default:
                    throw new RuntimeException("Unrecognized " + WRITE_DISTRIBUTION_MODE + ": " + writeMode);
            }
        }
This operator is used to buffer a mini batch data group by field before writing to save memory resources.

This operator must have the same degree of parallelism as the writer operator to ensure that the data of
the same batch of equality fields in a checkpoint period will only be processed once by a certain writer
subtask (rather than processed multiple times).

@param input
@return
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkSink.java