[Feature][Connector-V2] StarRocks-sink support schema evolution (#8082)

Author: jw-itq

docs/en/concept/schema-evolution.md

@@ -12,6 +12,7 @@ Now we only support the operation about `add column`、`drop column`、`rename c
 ### Sink
 [Jdbc-Mysql](https://github.com/apache/seatunnel/blob/dev/docs/en/connector-v2/sink/Jdbc.md)
 [Jdbc-Oracle](https://github.com/apache/seatunnel/blob/dev/docs/en/connector-v2/sink/Jdbc.md)
+[StarRocks](https://github.com/apache/seatunnel/blob/dev/docs/en/connector-v2/sink/StarRocks.md)
 
 Note: The schema evolution is not support the transform at now. The schema evolution of different types of databases（Oracle-CDC -> Jdbc-Mysql）is currently not supported the default value of the column in ddl.
 
@@ -151,3 +152,55 @@ sink {
   }
 }
 ```
+
+### Mysql-cdc -> StarRocks
+```
+env {
+  # You can set engine configuration here
+  parallelism = 1
+  job.mode = "STREAMING"
+  checkpoint.interval = 5000
+}
+
+source {
+  MySQL-CDC {
+    username = "st_user_source"
+    password = "mysqlpw"
+    table-names = ["shop.products"]
+    base-url = "jdbc:mysql://mysql_cdc_e2e:3306/shop"
+    debezium = {
+      include.schema.changes = true
+    }
+  }
+}
+
+sink {
+  StarRocks {
+    nodeUrls = ["starrocks_cdc_e2e:8030"]
+    username = "root"
+    password = ""
+    database = "shop"
+    table = "${table_name}"
+    base-url = "jdbc:mysql://starrocks_cdc_e2e:9030/shop"
+    max_retries = 3
+    enable_upsert_delete = true
+    schema_save_mode="RECREATE_SCHEMA"
+    data_save_mode="DROP_DATA"
+    save_mode_create_template = """
+    CREATE TABLE IF NOT EXISTS shop.`${table_name}` (
+        ${rowtype_primary_key},
+        ${rowtype_fields}
+        ) ENGINE=OLAP
+        PRIMARY KEY (${rowtype_primary_key})
+        DISTRIBUTED BY HASH (${rowtype_primary_key})
+        PROPERTIES (
+                "replication_num" = "1",
+                "in_memory" = "false",
+                "enable_persistent_index" = "true",
+                "replicated_storage" = "true",
+                "compression" = "LZ4"
+          )
+    """
+  }
+}
+```

docs/zh/concept/schema-evolution.md

@@ -11,6 +11,7 @@
 ### 目标
 [Jdbc-Mysql](https://github.com/apache/seatunnel/blob/dev/docs/zh/connector-v2/sink/Jdbc.md)
 [Jdbc-Oracle](https://github.com/apache/seatunnel/blob/dev/docs/en/connector-v2/sink/Jdbc.md)
+[StarRocks](https://github.com/apache/seatunnel/blob/dev/docs/en/connector-v2/sink/StarRocks.md)
 
 注意: 目前模式演进不支持transform。不同类型数据库(Oracle-CDC -> Jdbc-Mysql)的模式演进目前不支持ddl中列的默认值。
 
@@ -150,3 +151,55 @@ sink {
   }
 }
 ```
+
+### Mysql-cdc -> StarRocks
+```
+env {
+  # You can set engine configuration here
+  parallelism = 1
+  job.mode = "STREAMING"
+  checkpoint.interval = 5000
+}
+
+source {
+  MySQL-CDC {
+    username = "st_user_source"
+    password = "mysqlpw"
+    table-names = ["shop.products"]
+    base-url = "jdbc:mysql://mysql_cdc_e2e:3306/shop"
+    debezium = {
+      include.schema.changes = true
+    }
+  }
+}
+
+sink {
+  StarRocks {
+    nodeUrls = ["starrocks_cdc_e2e:8030"]
+    username = "root"
+    password = ""
+    database = "shop"
+    table = "${table_name}"
+    base-url = "jdbc:mysql://starrocks_cdc_e2e:9030/shop"
+    max_retries = 3
+    enable_upsert_delete = true
+    schema_save_mode="RECREATE_SCHEMA"
+    data_save_mode="DROP_DATA"
+    save_mode_create_template = """
+    CREATE TABLE IF NOT EXISTS shop.`${table_name}` (
+        ${rowtype_primary_key},
+        ${rowtype_fields}
+        ) ENGINE=OLAP
+        PRIMARY KEY (${rowtype_primary_key})
+        DISTRIBUTED BY HASH (${rowtype_primary_key})
+        PROPERTIES (
+                "replication_num" = "1",
+                "in_memory" = "false",
+                "enable_persistent_index" = "true",
+                "replicated_storage" = "true",
+                "compression" = "LZ4"
+          )
+    """
+  }
+}
+```

seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksSinkManager.java

@@ -19,6 +19,7 @@
 
 import org.apache.seatunnel.shade.com.google.common.base.Strings;
 
+import org.apache.seatunnel.api.table.catalog.TableSchema;
 import org.apache.seatunnel.connectors.seatunnel.starrocks.config.SinkConfig;
 import org.apache.seatunnel.connectors.seatunnel.starrocks.exception.StarRocksConnectorErrorCode;
 import org.apache.seatunnel.connectors.seatunnel.starrocks.exception.StarRocksConnectorException;
@@ -43,10 +44,10 @@ public class StarRocksSinkManager {
     private int batchRowCount = 0;
     private long batchBytesSize = 0;
 
-    public StarRocksSinkManager(SinkConfig sinkConfig, List<String> fileNames) {
+    public StarRocksSinkManager(SinkConfig sinkConfig, TableSchema tableSchema) {
         this.sinkConfig = sinkConfig;
         this.batchList = new ArrayList<>();
-        starrocksStreamLoadVisitor = new StarRocksStreamLoadVisitor(sinkConfig, fileNames);
+        starrocksStreamLoadVisitor = new StarRocksStreamLoadVisitor(sinkConfig, tableSchema);
     }
 
     private void tryInit() throws IOException {

seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksStreamLoadVisitor.java

@@ -17,12 +17,15 @@
 
 package org.apache.seatunnel.connectors.seatunnel.starrocks.client;
 
+import org.apache.seatunnel.api.table.catalog.Column;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
 import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
 import org.apache.seatunnel.common.utils.JsonUtils;
 import org.apache.seatunnel.connectors.seatunnel.starrocks.config.SinkConfig;
 import org.apache.seatunnel.connectors.seatunnel.starrocks.exception.StarRocksConnectorErrorCode;
 import org.apache.seatunnel.connectors.seatunnel.starrocks.exception.StarRocksConnectorException;
 import org.apache.seatunnel.connectors.seatunnel.starrocks.serialize.StarRocksDelimiterParser;
+import org.apache.seatunnel.connectors.seatunnel.starrocks.serialize.StarRocksSinkOP;
 
 import org.apache.commons.codec.binary.Base64;
 
@@ -56,11 +59,11 @@ public class StarRocksStreamLoadVisitor {
     private static final String RESULT_LABEL_ABORTED = "ABORTED";
     private static final String RESULT_LABEL_UNKNOWN = "UNKNOWN";
 
-    private List<String> fieldNames;
+    private final TableSchema tableSchema;
 
-    public StarRocksStreamLoadVisitor(SinkConfig sinkConfig, List<String> fieldNames) {
+    public StarRocksStreamLoadVisitor(SinkConfig sinkConfig, TableSchema tableSchema) {
         this.sinkConfig = sinkConfig;
-        this.fieldNames = fieldNames;
+        this.tableSchema = tableSchema;
         this.httpHelper = new HttpHelper(sinkConfig);
     }
 
@@ -260,16 +263,19 @@ private String getBasicAuthHeader(String username, String password) {
 
     private Map<String, String> getStreamLoadHttpHeader(String label) {
         Map<String, String> headerMap = new HashMap<>();
-        if (null != fieldNames
-                && !fieldNames.isEmpty()
+        List<Column> columns = tableSchema.getColumns();
+        List<String> fieldNames =
+                columns.stream().map(Column::getName).collect(Collectors.toList());
+        if (sinkConfig.isEnableUpsertDelete()) {
+            fieldNames.add(StarRocksSinkOP.COLUMN_KEY);
+        }
+        if (!fieldNames.isEmpty()
                 && SinkConfig.StreamLoadFormat.CSV.equals(sinkConfig.getLoadFormat())) {
             headerMap.put(
                     "columns",
-                    String.join(
-                            ",",
-                            fieldNames.stream()
-                                    .map(f -> String.format("`%s`", f))
-                                    .collect(Collectors.toList())));
+                    fieldNames.stream()
+                            .map(f -> String.format("`%s`", f))
+                            .collect(Collectors.joining(",")));
         }
         if (null != sinkConfig.getStreamLoadProps()) {
             for (Map.Entry<String, Object> entry : sinkConfig.getStreamLoadProps().entrySet()) {

seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/datatypes/StarRocksType.java

@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.seatunnel.connectors.seatunnel.starrocks.datatypes;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+@Getter
+@AllArgsConstructor
+public class StarRocksType {
+    public static final String SR_NULL = "NULL";
+    public static final String SR_BOOLEAN = "BOOLEAN";
+    public static final String SR_TINYINT = "TINYINT";
+    public static final String SR_SMALLINT = "SMALLINT";
+    public static final String SR_INT = "INT";
+    public static final String SR_BIGINT = "BIGINT";
+    public static final String SR_LARGEINT = "LARGEINT";
+    public static final String SR_FLOAT = "FLOAT";
+    public static final String SR_DOUBLE = "DOUBLE";
+    public static final String SR_DECIMAL = "DECIMAL";
+    public static final String SR_DECIMALV3 = "DECIMALV3";
+    public static final String SR_DATE = "DATE";
+    public static final String SR_DATETIME = "DATETIME";
+    public static final String SR_CHAR = "CHAR";
+    public static final String SR_VARCHAR = "VARCHAR";
+    public static final String SR_STRING = "STRING";
+
+    public static final String SR_BOOLEAN_ARRAY = "ARRAY<boolean>";
+    public static final String SR_TINYINT_ARRAY = "ARRAY<tinyint>";
+    public static final String SR_SMALLINT_ARRAY = "ARRAY<smallint>";
+    public static final String SR_INT_ARRAY = "ARRAY<int(11)>";
+    public static final String SR_BIGINT_ARRAY = "ARRAY<bigint>";
+    public static final String SR_FLOAT_ARRAY = "ARRAY<float>";
+    public static final String SR_DOUBLE_ARRAY = "ARRAY<double>";
+    public static final String SR_DECIMALV3_ARRAY = "ARRAY<DECIMALV3>";
+    public static final String SR_DECIMALV3_ARRAY_COLUMN_TYPE_TMP = "ARRAY<DECIMALV3(%s, %s)>";
+    public static final String SR_DATEV2_ARRAY = "ARRAY<DATEV2>";
+    public static final String SR_DATETIMEV2_ARRAY = "ARRAY<DATETIMEV2>";
+    public static final String SR_STRING_ARRAY = "ARRAY<STRING>";
+
+    // Because can not get the column length from array, So the following types of arrays cannot be
+    // generated properly.
+    public static final String SR_LARGEINT_ARRAY = "ARRAY<largeint>";
+    public static final String SR_CHAR_ARRAY = "ARRAY<CHAR>";
+    public static final String SR_CHAR_ARRAY_COLUMN_TYPE_TMP = "ARRAY<CHAR(%s)>";
+    public static final String SR_VARCHAR_ARRAY = "ARRAY<VARCHAR>";
+    public static final String SR_VARCHAR_ARRAY_COLUMN_TYPE_TMP = "ARRAY<VARCHAR(%s)>";
+
+    public static final String SR_JSON = "JSON";
+    public static final String SR_JSONB = "JSONB";
+
+    public static final String SR_ARRAY = "ARRAY";
+
+    public static final String SR_ARRAY_BOOLEAN_INTER = "tinyint(1)";
+    public static final String SR_ARRAY_TINYINT_INTER = "tinyint(4)";
+    public static final String SR_ARRAY_SMALLINT_INTER = "smallint(6)";
+    public static final String SR_ARRAY_INT_INTER = "int(11)";
+    public static final String SR_ARRAY_BIGINT_INTER = "bigint(20)";
+    public static final String SR_ARRAY_DECIMAL_PRE = "DECIMAL";
+    public static final String SR_ARRAY_DATE_INTER = "date";
+    public static final String SR_ARRAY_DATEV2_INTER = "DATEV2";
+    public static final String SR_ARRAY_DATETIME_INTER = "DATETIME";
+    public static final String SR_ARRAY_DATETIMEV2_INTER = "DATETIMEV2";
+
+    public static final String SR_MAP = "MAP";
+    public static final String SR_MAP_COLUMN_TYPE = "MAP<%s, %s>";
+
+    public static final String SR_BOOLEAN_INDENTFIER = "TINYINT(1)";
+
+    private String type;
+}