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);
            }
        }
Append the iceberg sink operators to write records to iceberg table.

@return {@link DataStreamSink} for sink.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkSink.java