[INLONG-9240][Manager][Sort] Add append mode for the Iceberg connector

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/iceberg/IcebergSink.java

@@ -72,6 +72,9 @@ public class IcebergSink extends StreamSink {
     @ApiModelProperty("Primary key")
     private String primaryKey;
 
+    @ApiModelProperty("append mode, UPSERT or APPEND")
+    private String appendMode;
+
     public IcebergSink() {
         this.setSinkType(SinkType.ICEBERG);
     }

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/iceberg/IcebergSinkRequest.java

@@ -64,4 +64,7 @@ public class IcebergSinkRequest extends SinkRequest {
     @ApiModelProperty("Primary key")
     private String primaryKey;
 
+    @ApiModelProperty("append mode, UPSERT or APPEND")
+    private String appendMode;
+
 }

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/node/provider/IcebergProvider.java

@@ -96,7 +96,8 @@ public LoadNode createLoadNode(StreamNode nodeInfo, Map<String, StreamField> con
                 icebergSink.getPrimaryKey(),
                 catalogType,
                 icebergSink.getCatalogUri(),
-                icebergSink.getWarehouse());
+                icebergSink.getWarehouse(),
+                icebergSink.getAppendMode());
     }
 
     @Override

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/constant/IcebergConstant.java

@@ -36,6 +36,8 @@ public class IcebergConstant {
     public static final String STREAMING = "streaming";
     public static final String STARTING_STRATEGY_KEY = "starting-strategy";
 
+    public static final String APPEND_MODE_KEY = "appendMode";
+
     /**
      * Iceberg supported catalog type
      */

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/IcebergLoadNode.java

@@ -43,6 +43,7 @@
 import java.util.EnumSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 
 @JsonTypeName("icebergLoad")
@@ -73,6 +74,9 @@ public class IcebergLoadNode extends LoadNode implements InlongMetric, Metadata,
     @JsonProperty("warehouse")
     private String warehouse;
 
+    @JsonProperty("append mode, UPSERT or APPEND")
+    private String appendMode;
+
     @JsonCreator
     public IcebergLoadNode(@JsonProperty("id") String id,
             @JsonProperty("name") String name,
@@ -87,14 +91,16 @@ public IcebergLoadNode(@JsonProperty("id") String id,
             @JsonProperty("primaryKey") String primaryKey,
             @JsonProperty("catalogType") IcebergConstant.CatalogType catalogType,
             @JsonProperty("uri") String uri,
-            @JsonProperty("warehouse") String warehouse) {
+            @JsonProperty("warehouse") String warehouse,
+            @JsonProperty("appendMode") String appendMode) {
         super(id, name, fields, fieldRelations, filters, filterStrategy, sinkParallelism, properties);
         this.tableName = Preconditions.checkNotNull(tableName, "table name is null");
         this.dbName = Preconditions.checkNotNull(dbName, "db name is null");
         this.primaryKey = primaryKey;
         this.catalogType = catalogType == null ? CatalogType.HIVE : catalogType;
         this.uri = uri;
         this.warehouse = warehouse;
+        this.appendMode = appendMode;
     }
 
     @Override
@@ -108,11 +114,12 @@ public Map<String, String> tableOptions() {
         options.put(IcebergConstant.DEFAULT_DATABASE_KEY, dbName);
         options.put(IcebergConstant.CATALOG_TYPE_KEY, catalogType.name());
         options.put(IcebergConstant.CATALOG_NAME_KEY, catalogType.name());
+        options.put(IcebergConstant.APPEND_MODE_KEY, appendMode);
         if (null != uri) {
-            options.put("uri", uri);
+            options.put(IcebergConstant.URI_KEY, uri);
         }
         if (null != warehouse) {
-            options.put("warehouse", warehouse);
+            options.put(IcebergConstant.WAREHOUSE_KEY, warehouse);
         }
         return options;
     }

inlong-sort/sort-common/src/test/java/org/apache/inlong/sort/protocol/node/load/IcebergLoadNodeTest.java

@@ -45,6 +45,7 @@ public IcebergLoadNode getTestObject() {
                 "id",
                 CatalogType.HIVE,
                 "thrift://localhost:9083",
-                "hdfs://localhost:9000/user/iceberg/warehouse");
+                "hdfs://localhost:9000/user/iceberg/warehouse",
+                null);
     }
 }

inlong-sort/sort-core/src/test/java/org/apache/inlong/sort/parser/IcebergNodeSqlParserTest.java

@@ -100,7 +100,8 @@ private IcebergLoadNode buildIcebergLoadNodeWithHadoopCatalog() {
                 null,
                 CatalogType.HADOOP,
                 null,
-                "hdfs://localhost:9000/iceberg/warehouse");
+                "hdfs://localhost:9000/iceberg/warehouse",
+                null);
     }
 
     private IcebergLoadNode buildIcebergLoadNodeWithHiveCatalog() {
@@ -139,7 +140,8 @@ private IcebergLoadNode buildIcebergLoadNodeWithHiveCatalog() {
                 null,
                 CatalogType.HIVE,
                 "thrift://localhost:9083",
-                "/hive/warehouse");
+                "/hive/warehouse",
+                null);
     }
 
     /**