[feature][connector-v2][clickhouse] Support write cdc changelog event in clickhouse sink (#3653)

* [feature][connector-v2][clickhouse] Support write cdc changelog event in clickhouse sink

MergeTree Table Engine:
- Support enable `allowExperimentalLightweightDelete` setting
- CDC Events:
  - INSERT: INSERT INTO SQL
  - UPDATE_BEFORE: DELETE SQL
  - UPDATE_AFTER: INSERT INTO SQL
  - DELETE: DELETE SQL

- ReplacingMergeTree Engine CDC Events:
  - INSERT: INSERT INTO SQL
  - UPDATE_BEFORE: ignore
  - UPDATE_AFTER: INSERT INTO SQL
  - DELETE: DELETE SQL

Other Table Engine:
- CDC Events:
  - INSERT: INSERT INTO SQL
  - UPDATE_BEFORE: ignore
  - UPDATE_AFTER: ALTER TABLE UPDATE SQL
  - DELETE: ALTER TABLE DELETE SQL

* update

* fix InsertOrUpdateBatchStatementExecutor

* add check

Author: hailin0

docs/en/connector-v2/sink/Clickhouse.md

@@ -13,6 +13,7 @@ Used to write data to Clickhouse.
 The Clickhouse sink plug-in can achieve accuracy once by implementing idempotent writing, and needs to cooperate with aggregatingmergetree and other engines that support deduplication.
 
 - [ ] [schema projection](../../concept/connector-v2-features.md)
+- [x] [cdc](../../concept/connector-v2-features.md)
 
 :::tip
 
@@ -22,19 +23,22 @@ Write data to Clickhouse can also be done using JDBC
 
 ## Options
 
-| name           | type   | required | default value |
-|----------------|--------|----------|---------------|
-| host           | string | yes      | -             |
-| database       | string | yes      | -             |
-| table          | string | yes      | -             |
-| username       | string | yes      | -             |
-| password       | string | yes      | -             |
-| fields         | string | yes      | -             |
-| clickhouse.*   | string | no       |               |
-| bulk_size      | string | no       | 20000         |
-| split_mode     | string | no       | false         |
-| sharding_key   | string | no       | -             |
-| common-options |        | no       | -             |
+| name                                  | type    | required | default value |
+|---------------------------------------|---------|----------|---------------|
+| host                                  | string  | yes      | -             |
+| database                              | string  | yes      | -             |
+| table                                 | string  | yes      | -             |
+| username                              | string  | yes      | -             |
+| password                              | string  | yes      | -             |
+| fields                                | string  | yes      | -             |
+| clickhouse.*                          | string  | no       |               |
+| bulk_size                             | string  | no       | 20000         |
+| split_mode                            | string  | no       | false         |
+| sharding_key                          | string  | no       | -             |
+| primary_key                           | string  | no       | -             |
+| support_upsert                        | boolean | no       | false         |
+| allow_experimental_lightweight_delete | boolean | no       | false         |
+| common-options                        |         | no       | -             |
 
 ### host [string]
 
@@ -82,39 +86,90 @@ When use split_mode, which node to send data to is a problem, the default is ran
 'sharding_key' parameter can be used to specify the field for the sharding algorithm. This option only
 worked when 'split_mode' is true.
 
+### primary_key [string]
+
+Mark the primary key column from clickhouse table, and based on primary key execute INSERT/UPDATE/DELETE to clickhouse table
+
+### support_upsert [boolean]
+
+Support upsert row by query primary key
+
+### allow_experimental_lightweight_delete [boolean]
+
+Allow experimental lightweight delete based on `*MergeTree` table engine
+
 ### common options
 
 Sink plugin common parameters, please refer to [Sink Common Options](common-options.md) for details
 
 ## Examples
 
+Simple
+
 ```hocon
 sink {
+  Clickhouse {
+    host = "localhost:8123"
+    database = "default"
+    table = "fake_all"
+    username = "default"
+    password = ""
+  }
+}
+```
+
+Split mode
 
+```hocon
+sink {
   Clickhouse {
     host = "localhost:8123"
     database = "default"
     table = "fake_all"
     username = "default"
     password = ""
+    
+    # split mode options
     split_mode = true
     sharding_key = "age"
   }
-  
 }
 ```
 
+CDC(Change data capture)
+
 ```hocon
 sink {
+  Clickhouse {
+    host = "localhost:8123"
+    database = "default"
+    table = "fake_all"
+    username = "default"
+    password = ""
+    
+    # cdc options
+    primary_key = "id"
+    support_upsert = true
+  }
+}
+```
 
+CDC(Change data capture) for *MergeTree engine
+
+```hocon
+sink {
   Clickhouse {
     host = "localhost:8123"
     database = "default"
     table = "fake_all"
     username = "default"
     password = ""
+    
+    # cdc options
+    primary_key = "id"
+    support_upsert = true
+    allow_experimental_lightweight_delete = true
   }
-  
 }
 ```
 
@@ -132,3 +187,5 @@ sink {
 - [Improve] Clickhouse Sink support nest type and array type([3047](https://github.com/apache/incubator-seatunnel/pull/3047))
 
 - [Improve] Clickhouse Sink support geo type([3141](https://github.com/apache/incubator-seatunnel/pull/3141))
+
+- [Feature] Support CDC write DELETE/UPDATE/INSERT events ([3653](https://github.com/apache/incubator-seatunnel/pull/3653))

seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/config/ClickhouseConfig.java

@@ -81,8 +81,25 @@ public class ClickhouseConfig {
     /**
      * When split_mode is true, the sharding_key use for split
      */
-    public static final Option<String> SHARDING_KEY = Options.key("sharding_key").stringType()
-        .noDefaultValue().withDescription("When split_mode is true, the sharding_key use for split");
+    public static final Option<String> SHARDING_KEY = Options.key("sharding_key")
+        .stringType()
+        .noDefaultValue()
+        .withDescription("When split_mode is true, the sharding_key use for split");
+
+    public static final Option<String> PRIMARY_KEY = Options.key("primary_key")
+        .stringType()
+        .noDefaultValue()
+        .withDescription("Mark the primary key column from clickhouse table, and based on primary key execute INSERT/UPDATE/DELETE to clickhouse table");
+
+    public static final Option<Boolean> SUPPORT_UPSERT = Options.key("support_upsert")
+        .booleanType()
+        .defaultValue(false)
+        .withDescription("Support upsert row by query primary key");
+
+    public static final Option<Boolean> ALLOW_EXPERIMENTAL_LIGHTWEIGHT_DELETE = Options.key("allow_experimental_lightweight_delete")
+        .booleanType()
+        .defaultValue(false)
+        .withDescription("Allow experimental lightweight delete based on `*MergeTree` table engine");
 
     /**
      * ClickhouseFile sink connector used clickhouse-local program's path

seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/config/ReaderOption.java

@@ -20,75 +20,28 @@
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.connectors.seatunnel.clickhouse.shard.ShardMetadata;
 
+import lombok.Builder;
+import lombok.Getter;
+import lombok.Setter;
+
 import java.io.Serializable;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 
+@Builder
+@Getter
 public class ReaderOption implements Serializable {
 
     private ShardMetadata shardMetadata;
     private List<String> fields;
-
+    private String[] primaryKeys;
+    private boolean allowExperimentalLightweightDelete;
+    private boolean supportUpsert;
+    private String tableEngine;
     private Map<String, String> tableSchema;
+    @Setter
     private SeaTunnelRowType seaTunnelRowType;
     private Properties properties;
     private int bulkSize;
-
-    public ReaderOption(ShardMetadata shardMetadata,
-                        Properties properties, List<String> fields, Map<String, String> tableSchema, int bulkSize) {
-        this.shardMetadata = shardMetadata;
-        this.properties = properties;
-        this.fields = fields;
-        this.tableSchema = tableSchema;
-        this.bulkSize = bulkSize;
-    }
-
-    public Properties getProperties() {
-        return properties;
-    }
-
-    public void setProperties(Properties properties) {
-        this.properties = properties;
-    }
-
-    public ShardMetadata getShardMetadata() {
-        return shardMetadata;
-    }
-
-    public void setShardMetadata(ShardMetadata shardMetadata) {
-        this.shardMetadata = shardMetadata;
-    }
-
-    public SeaTunnelRowType getSeaTunnelRowType() {
-        return seaTunnelRowType;
-    }
-
-    public void setSeaTunnelRowType(SeaTunnelRowType seaTunnelRowType) {
-        this.seaTunnelRowType = seaTunnelRowType;
-    }
-
-    public Map<String, String> getTableSchema() {
-        return tableSchema;
-    }
-
-    public void setTableSchema(Map<String, String> tableSchema) {
-        this.tableSchema = tableSchema;
-    }
-
-    public List<String> getFields() {
-        return fields;
-    }
-
-    public void setFields(List<String> fields) {
-        this.fields = fields;
-    }
-
-    public int getBulkSize() {
-        return bulkSize;
-    }
-
-    public void setBulkSize(int bulkSize) {
-        this.bulkSize = bulkSize;
-    }
 }