pvary commented on code in PR #14559:
URL: https://github.com/apache/iceberg/pull/14559#discussion_r2517711092
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java:
##########
@@ -302,30 +303,58 @@ private void commitDeltaTxn(
CommitSummary summary,
String newFlinkJobId,
String operatorId) {
- for (Map.Entry<Long, List<WriteResult>> e : pendingResults.entrySet()) {
- long checkpointId = e.getKey();
- List<WriteResult> writeResults = e.getValue();
-
- RowDelta rowDelta = table.newRowDelta().scanManifestsWith(workerPool);
- for (WriteResult result : writeResults) {
- // 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.
- Arrays.stream(result.dataFiles()).forEach(rowDelta::addRows);
- Arrays.stream(result.deleteFiles()).forEach(rowDelta::addDeletes);
+ if (summary.deleteFilesCount() == 0) {
Review Comment:
Also, as I have mentioned in the comment, this could cause issues if
"replacePartition" is used
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
