[Feature][Connector-V2] Support CDC mode for databend sink connector

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

@@ -12,9 +12,9 @@ import ChangeLog from '../changelog/connector-databend.md';
 
 ## Key Features
 
-- [ ] [Exactly-Once](../../concept/connector-v2-features.md)
 - [ ] [Support Multi-table Writing](../../concept/connector-v2-features.md)
-- [ ] [CDC](../../concept/connector-v2-features.md)
+- [x] [Exactly-Once](../../concept/connector-v2-features.md)
+- [x] [CDC](../../concept/connector-v2-features.md)
 - [x] [Parallelism](../../concept/connector-v2-features.md)
 
 ## Description
@@ -37,7 +37,7 @@ The Databend sink internally implements bulk data import through stage attachmen
 | Name | Type | Required | Default Value | Description                                 |
 |------|------|----------|---------------|---------------------------------------------|
 | url | String | Yes | - | Databend JDBC connection URL               |
-| username | String | Yes | - | Databend database username                    |
+| user | String | Yes | - | Databend database username                    |
 | password | String | Yes | - | Databend database password                     |
 | database | String | No | - | Databend database name, defaults to the database name specified in the connection URL |
 | table | String | No | - | Databend table name                       |
@@ -49,6 +49,8 @@ The Databend sink internally implements bulk data import through stage attachmen
 | custom_sql | String | No | - | Custom write SQL, typically used for complex write scenarios              |
 | execute_timeout_sec | Integer | No | 300 | SQL execution timeout (seconds)                      |
 | jdbc_config | Map | No | - | Additional JDBC connection configuration, such as connection timeout parameters             |
+| conflict_key | String | No | - | Conflict key for CDC mode, used to determine the primary key for conflict resolution |
+| allow_delete | Boolean | No | false | Whether to allow delete operations in CDC mode |
 
 ### schema_save_mode[Enum]
 
@@ -112,7 +114,7 @@ source {
 sink {
   Databend {
     url = "jdbc:databend://localhost:8000"
-    username = "root"
+    user = "root"
     password = ""
     database = "default"
     table = "target_table"
@@ -127,7 +129,7 @@ sink {
 sink {
   Databend {
     url = "jdbc:databend://localhost:8000"
-    username = "root"
+    user = "root"
     password = ""
     database = "default"
     table = "target_table"
@@ -142,7 +144,7 @@ sink {
 sink {
   Databend {
     url = "jdbc:databend://localhost:8000"
-    username = "root"
+    user = "root"
     password = ""
     database = "default"
     table = "target_table"
@@ -152,6 +154,26 @@ sink {
 }
 ```
 
+### CDC mode
+
+```hocon
+sink {
+  Databend {
+    url = "jdbc:databend://databend:8000/default?ssl=false"
+    user = "root"
+    password = ""
+    database = "default"
+    table = "sink_table"
+
+    # Enable CDC mode
+    batch_size = 1
+    interval = 3
+    conflict_key = "id"
+    allow_delete = true
+  }
+}
+```
+
 ## Related Links
 
 - [Databend Official Website](https://databend.rs/)

docs/zh/connector-v2/sink/Databend.md

@@ -12,9 +12,9 @@ import ChangeLog from '../changelog/connector-databend.md';
 
 ## 主要特性
 
-- [ ] [精确一次](../../concept/connector-v2-features.md)
 - [ ] [支持多表写入](../../concept/connector-v2-features.md)
-- [ ] [cdc](../../concept/connector-v2-features.md)
+- [x] [精确一次](../../concept/connector-v2-features.md)
+- [x] [cdc](../../concept/connector-v2-features.md)
 - [x] [并行度](../../concept/connector-v2-features.md)
 
 ## 描述
@@ -49,6 +49,8 @@ Databend sink 内部通过 stage attachment 实现数据的批量导入。
 | custom_sql | String | 否 | - | 自定义写入 SQL，通常用于复杂的写入场景              |
 | execute_timeout_sec | Integer | 否 | 300 | 执行SQL的超时时间（秒）                      |
 | jdbc_config | Map | 否 | - | 额外的 JDBC 连接配置，如连接超时参数等             |
+| conflict_key | String | 否 | - | cdc 模式下的冲突键，用于确定冲突解决的主键 |
+| allow_delete | Boolean | 否 | false | cdc 模式下是否允许删除操作 |
 
 ### schema_save_mode[Enum]
 
@@ -152,6 +154,26 @@ sink {
 }
 ```
 
+### CDC mode
+
+```hocon
+sink {
+  Databend {
+    url = "jdbc:databend://databend:8000/default?ssl=false"
+    user = "root"
+    password = ""
+    database = "default"
+    table = "sink_table"
+
+    # Enable CDC mode
+    batch_size = 1
+    interval = 3
+    conflict_key = "id"
+    allow_delete = true
+  }
+}
+```
+
 ## 相关链接
 
 - [Databend 官方网站](https://databend.rs/)

seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/config/DatabendOptions.java

@@ -37,7 +37,7 @@ public class DatabendOptions {
                     .withDescription("Whether to use SSL for the Databend connection");
 
     public static final Option<String> USERNAME =
-            Options.key("username")
+            Options.key("user")
                     .stringType()
                     .noDefaultValue()
                     .withDescription("The username for Databend database authentication");
@@ -97,4 +97,10 @@ public class DatabendOptions {
                     .booleanType()
                     .defaultValue(true)
                     .withDescription("Whether to auto commit for sink");
+
+    public static final Option<String> CONFLICT_KEY =
+            Options.key("conflict_key")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("The conflict key for sink, used in upsert mode");
 }