private void replacePartitions(
            NavigableMap<Long, WriteResult> pendingResults,
            CommitSummary summary,
            String newFlinkJobId,
            String operatorId,
            long checkpointId) {
        Preconditions.checkState(
                summary.deleteFilesCount() == 0, "Cannot overwrite partitions with delete files.");
        // Commit the overwrite transaction.
        ReplacePartitions dynamicOverwrite = table.newReplacePartitions().scanManifestsWith(workerPool);
        for (WriteResult result : pendingResults.values()) {
            Preconditions.checkState(
                    result.referencedDataFiles().length == 0, "Should have no referenced data files.");
            Arrays.stream(result.dataFiles()).forEach(dynamicOverwrite::addFile);
        }

        commitOperation(
                dynamicOverwrite,
                summary,
                "dynamic partition overwrite",
                newFlinkJobId,
                operatorId,
                checkpointId);
    }
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