[inlong] branch master updated: [INLONG-7470][Sort] Fix Iceberg's data were duplicated when delete records in upsert mode (#7472)

[dockerzhang]

This is an automated email from the ASF dual-hosted git repository.

dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git


The following commit(s) were added to refs/heads/master by this push:
     new bf4659612 [INLONG-7470][Sort] Fix Iceberg's data were duplicated when 
delete records in upsert mode (#7472)
bf4659612 is described below

commit bf46596123a3f1dcd7d8e3c80b87aa556f50a5ba
Author: ganfengtan <ganfeng...@users.noreply.github.com>
AuthorDate: Wed Mar 1 16:18:17 2023 +0800

    [INLONG-7470][Sort] Fix Iceberg's data were duplicated when delete records 
in upsert mode (#7472)
---
 .../org/apache/inlong/sort/iceberg/sink/BaseDeltaTaskWriter.java    | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git 
a/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/BaseDeltaTaskWriter.java
 
b/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/BaseDeltaTaskWriter.java
index 5516d650d..6b5fcb835 100644
--- 
a/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/BaseDeltaTaskWriter.java
+++ 
b/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/BaseDeltaTaskWriter.java
@@ -94,7 +94,11 @@ abstract class BaseDeltaTaskWriter extends 
BaseTaskWriter<RowData> {
                 writer.delete(row);
                 break;
             case DELETE:
-                writer.delete(row);
+                if (upsert) { // 
https://github.com/apache/iceberg/pull/6753/files
+                    writer.deleteKey(keyProjection.wrap(row));
+                } else {
+                    writer.delete(row);
+                }
                 break;
 
             default: