private FileAppenderFactory<RowData> createRowDataFileAppenderFactory(Table table,
            RowType flinkSchema, List<Integer> equalityFieldIds, boolean upsert, boolean appendMode) {
        if (equalityFieldIds == null || equalityFieldIds.isEmpty() || appendMode) {
            return new FlinkAppenderFactory(schema, flinkSchema, table.properties(), spec);
        } else if (upsert) {
            // In upsert mode, only the new row is emitted using INSERT row kind. Therefore, any column of the inserted
            // row may differ from the deleted row other than the primary key fields, and the delete file must contain
            // values that are correct for the deleted row. Therefore, only write the equality delete fields.
            return new FlinkAppenderFactory(schema, flinkSchema, table.properties(), spec,
                    ArrayUtil.toIntArray(equalityFieldIds),
                    TypeUtil.select(schema, Sets.newHashSet(equalityFieldIds)), null);
        } else {
            return new FlinkAppenderFactory(schema, flinkSchema, table.properties(), spec,
                    ArrayUtil.toIntArray(equalityFieldIds), schema, null);
        }
    }
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