@Override
    public TaskWriter<RowData> create() {
        Preconditions.checkNotNull(outputFileFactory,
                "The outputFileFactory shouldn't be null if we have invoked the initialize().");

        if (equalityFieldIds == null || equalityFieldIds.isEmpty() || appendMode) {
            // Initialize a task writer to write INSERT only.
            if (spec.isUnpartitioned()) {
                LOGGER.info("Create an unPartitioned append writer for table {}.", table.name());
                return new UnpartitionedWriter<>(
                        spec, format, appenderFactory, outputFileFactory, io, targetFileSizeBytes);
            } else {
                LOGGER.info("Create a partitioned append writer for table {}.", table.name());
                if (miniBatchMode) {
                    return new RowDataGroupedPartitionedFanoutWriter(spec, format, appenderFactory, outputFileFactory,
                            io, targetFileSizeBytes, schema, flinkSchema);
                } else {
                    return new RowDataPartitionedFanoutWriter(spec, format, appenderFactory, outputFileFactory,
                            io, targetFileSizeBytes, schema, flinkSchema);
                }
            }
        } else {
            // Initialize a task writer to write both INSERT and equality DELETE.
            if (spec.isUnpartitioned()) {
                LOGGER.info("Create an unPartitioned upsert delta writer for table {}.", table.name());
                return new UnpartitionedDeltaWriter(spec, format, appenderFactory, outputFileFactory, io,
                        targetFileSizeBytes, schema, flinkSchema, equalityFieldIds, upsert);
            } else {
                LOGGER.info("Create a partitioned upsert delta writer for table {}.", table.name());
                if (miniBatchMode) {
                    return new GroupedPartitionedDeltaWriter(spec, format, appenderFactory, outputFileFactory, io,
                            targetFileSizeBytes, schema, flinkSchema, equalityFieldIds, upsert);
                } else {
                    return new PartitionedDeltaWriter(spec, format, appenderFactory, outputFileFactory, io,
                            targetFileSizeBytes, schema, flinkSchema, equalityFieldIds, upsert);
                }
            }
        }
    }
Copy from iceberg-flink:iceberg-flink-1.13:0.13.2
Add an option `sink.ignore.changelog` to support insert-only mode without equalityFieldIds.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/RowDataTaskWriterFactory.java