private void commitDeltaTxn(
            NavigableMap<Long, WriteResult> pendingResults,
            CommitSummary summary,
            String newFlinkJobId,
            String operatorId,
            long checkpointId) {
        if (summary.deleteFilesCount() == 0) {
            // To be compatible with iceberg format V1.
            AppendFiles appendFiles = table.newAppend().scanManifestsWith(workerPool);
            for (WriteResult result : pendingResults.values()) {
                Preconditions.checkState(
                        result.referencedDataFiles().length == 0,
                        "Should have no referenced data files for append.");
                Arrays.stream(result.dataFiles()).forEach(appendFiles::appendFile);
            }
            commitOperation(appendFiles, summary, "append", newFlinkJobId, operatorId, checkpointId);
        } else {
            // To be compatible with iceberg format V2.
            for (Map.Entry<Long, WriteResult> e : pendingResults.entrySet()) {
                // We don't commit the merged result into a single transaction because for the sequential
                // transaction txn1 and txn2, the equality-delete files of txn2 are required to be applied
                // to data files from txn1. Committing the merged one will lead to the incorrect delete
                // semantic.
                WriteResult result = e.getValue();

                // Row delta validations are not needed for streaming changes that write equality deletes.
                // Equality deletes are applied to data in all previous sequence numbers, so retries may
                // push deletes further in the future, but do not affect correctness. Position deletes
                // committed to the table in this path are used only to delete rows from data files that are
                // being added in this commit. There is no way for data files added along with the delete
                // files to be concurrently removed, so there is no need to validate the files referenced by
                // the position delete files that are being committed.
                RowDelta rowDelta = table.newRowDelta().scanManifestsWith(workerPool);

                Arrays.stream(result.dataFiles()).forEach(rowDelta::addRows);
                Arrays.stream(result.deleteFiles()).forEach(rowDelta::addDeletes);
                commitOperation(rowDelta, summary, "rowDelta", newFlinkJobId, operatorId, e.getKey());
            }
        }
    }
Copy from iceberg-flink:iceberg-flink-1.15:1.3.1
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/IcebergFilesCommitter.java