[Fix][Connector-v2] Fix the sql statement error of create table for doris and starrocks (#6679)

Author: dailai

Diff for: seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/SaveModeConstants.java

@@ -0,0 +1,66 @@
+/*
+ * 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.api.sink;
+
+import java.util.Arrays;
+import java.util.Optional;
+
+public enum SaveModePlaceHolder {
+    ROWTYPE_PRIMARY_KEY("rowtype_primary_key", "primary keys"),
+    ROWTYPE_UNIQUE_KEY("rowtype_unique_key", "unique keys"),
+    ROWTYPE_FIELDS("rowtype_fields", "fields"),
+    TABLE_NAME("table_name", "table name"),
+    DATABASE("database", "database");
+
+    private String keyValue;
+    private String display;
+
+    private static final String REPLACE_PLACE_HOLDER = "\\$\\{%s\\}";
+    private static final String PLACE_HOLDER = "${%s}";
+
+    SaveModePlaceHolder(String keyValue, String display) {
+        this.keyValue = keyValue;
+        this.display = display;
+    }
+
+    public static String getDisplay(String placeholder) {
+        Optional<SaveModePlaceHolder> saveModePlaceHolderEnumOptional =
+                Arrays.stream(SaveModePlaceHolder.values())
+                        .filter(
+                                saveModePlaceHolderEnum ->
+                                        placeholder.equals(
+                                                saveModePlaceHolderEnum.getPlaceHolder()))
+                        .findFirst();
+        if (saveModePlaceHolderEnumOptional.isPresent()) {
+            return saveModePlaceHolderEnumOptional.get().display;
+        }
+        throw new RuntimeException(String.format("Not support the placeholder: %s", placeholder));
+    }
+
+    public String getPlaceHolderKey() {
+        return this.keyValue;
+    }
+
+    public String getPlaceHolder() {
+        return String.format(PLACE_HOLDER, getPlaceHolderKey());
+    }
+
+    public String getReplacePlaceHolder() {
+        return String.format(REPLACE_PLACE_HOLDER, getPlaceHolderKey());
+    }
+}

Diff for: seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/SaveModePlaceHolder.java

@@ -36,6 +36,7 @@
 import static org.apache.seatunnel.common.exception.CommonErrorCode.GET_CATALOG_TABLES_WITH_UNSUPPORTED_TYPE_ERROR;
 import static org.apache.seatunnel.common.exception.CommonErrorCode.GET_CATALOG_TABLE_WITH_UNSUPPORTED_TYPE_ERROR;
 import static org.apache.seatunnel.common.exception.CommonErrorCode.JSON_OPERATION_FAILED;
+import static org.apache.seatunnel.common.exception.CommonErrorCode.SQL_TEMPLATE_HANDLED_ERROR;
 import static org.apache.seatunnel.common.exception.CommonErrorCode.UNSUPPORTED_DATA_TYPE;
 import static org.apache.seatunnel.common.exception.CommonErrorCode.UNSUPPORTED_ENCODING;
 import static org.apache.seatunnel.common.exception.CommonErrorCode.WRITE_SEATUNNEL_ROW_ERROR;
@@ -183,4 +184,19 @@ public static SeaTunnelRuntimeException jsonOperationError(
             return new SeaTunnelRuntimeException(code, params);
         }
     }
+
+    public static SeaTunnelRuntimeException sqlTemplateHandledError(
+            String tableName,
+            String keyName,
+            String template,
+            String placeholder,
+            String optionName) {
+        Map<String, String> params = new HashMap<>();
+        params.put("tableName", tableName);
+        params.put("keyName", keyName);
+        params.put("template", template);
+        params.put("placeholder", placeholder);
+        params.put("optionName", optionName);
+        return new SeaTunnelRuntimeException(SQL_TEMPLATE_HANDLED_ERROR, params);
+    }
 }

Diff for: seatunnel-common/src/main/java/org/apache/seatunnel/common/exception/CommonError.java

@@ -48,7 +48,10 @@ public enum CommonErrorCode implements SeaTunnelErrorCode {
             "<identifier> <operation> file '<fileName>' failed, because it not existed."),
     WRITE_SEATUNNEL_ROW_ERROR(
             "COMMON-23",
-            "<connector> write SeaTunnelRow failed, the SeaTunnelRow value is '<seaTunnelRow>'.");
+            "<connector> write SeaTunnelRow failed, the SeaTunnelRow value is '<seaTunnelRow>'."),
+    SQL_TEMPLATE_HANDLED_ERROR(
+            "COMMON-24",
+            "The table of <tableName> has no <keyName>, but the template \n <template> \n which has the place holder named <placeholder>. Please use the option named <optionName> to specify sql template");
 
     private final String code;
     private final String description;

Diff for: seatunnel-common/src/main/java/org/apache/seatunnel/common/exception/CommonErrorCode.java

@@ -0,0 +1,41 @@
+/*
+ * 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.common.sql.template;
+
+import org.apache.seatunnel.api.sink.SaveModePlaceHolder;
+import org.apache.seatunnel.common.exception.CommonError;
+
+import org.apache.commons.lang3.StringUtils;
+
+public class SqlTemplate {
+    public static void canHandledByTemplateWithPlaceholder(
+            String createTemplate,
+            String placeholder,
+            String actualPlaceHolderValue,
+            String tableName,
+            String optionsKey) {
+        if (createTemplate.contains(placeholder) && StringUtils.isBlank(actualPlaceHolderValue)) {
+            throw CommonError.sqlTemplateHandledError(
+                    tableName,
+                    SaveModePlaceHolder.getDisplay(placeholder),
+                    createTemplate,
+                    placeholder,
+                    optionsKey);
+        }
+    }
+}

Diff for: seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/sql/template/SqlTemplate.java

@@ -21,6 +21,7 @@
 import org.apache.seatunnel.api.configuration.Options;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.sink.DataSaveMode;
+import org.apache.seatunnel.api.sink.SaveModePlaceHolder;
 import org.apache.seatunnel.api.sink.SchemaSaveMode;
 
 import java.util.Map;
@@ -227,11 +228,20 @@ public interface DorisOptions {
             Options.key("save_mode_create_template")
                     .stringType()
                     .defaultValue(
-                            "CREATE TABLE IF NOT EXISTS `${database}`.`${table_name}` (\n"
-                                    + "${rowtype_fields}\n"
+                            "CREATE TABLE IF NOT EXISTS `"
+                                    + SaveModePlaceHolder.DATABASE.getPlaceHolder()
+                                    + "`.`"
+                                    + SaveModePlaceHolder.TABLE_NAME.getPlaceHolder()
+                                    + "` (\n"
+                                    + SaveModePlaceHolder.ROWTYPE_FIELDS.getPlaceHolder()
+                                    + "\n"
                                     + ") ENGINE=OLAP\n"
-                                    + " UNIQUE KEY (${rowtype_primary_key})\n"
-                                    + "DISTRIBUTED BY HASH (${rowtype_primary_key})\n "
+                                    + " UNIQUE KEY ("
+                                    + SaveModePlaceHolder.ROWTYPE_PRIMARY_KEY.getPlaceHolder()
+                                    + ")\n"
+                                    + "DISTRIBUTED BY HASH ("
+                                    + SaveModePlaceHolder.ROWTYPE_PRIMARY_KEY.getPlaceHolder()
+                                    + ")\n "
                                     + "PROPERTIES (\n"
                                     + "\"replication_allocation\" = \"tag.location.default: 1\",\n"
                                     + "\"in_memory\" = \"false\",\n"

Diff for: seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/config/DorisOptions.java

@@ -17,7 +17,7 @@
 
 package org.apache.seatunnel.connectors.doris.util;
 
-import org.apache.seatunnel.api.sink.SaveModeConstants;
+import org.apache.seatunnel.api.sink.SaveModePlaceHolder;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.catalog.Column;
 import org.apache.seatunnel.api.table.catalog.TablePath;
@@ -28,6 +28,8 @@
 import org.apache.seatunnel.api.table.type.DecimalType;
 import org.apache.seatunnel.api.table.type.LocalTimeType;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.connectors.doris.config.DorisOptions;
+import org.apache.seatunnel.connectors.seatunnel.common.sql.template.SqlTemplate;
 
 import org.apache.commons.lang3.StringUtils;
 
@@ -125,14 +127,25 @@ public static String getCreateTableStatement(
                             .map(r -> "`" + r.getColumnName() + "`")
                             .collect(Collectors.joining(","));
         }
+        SqlTemplate.canHandledByTemplateWithPlaceholder(
+                template,
+                SaveModePlaceHolder.ROWTYPE_PRIMARY_KEY.getPlaceHolder(),
+                primaryKey,
+                tablePath.getFullName(),
+                DorisOptions.SAVE_MODE_CREATE_TEMPLATE.key());
         template =
                 template.replaceAll(
-                        String.format("\\$\\{%s\\}", SaveModeConstants.ROWTYPE_PRIMARY_KEY),
+                        SaveModePlaceHolder.ROWTYPE_PRIMARY_KEY.getReplacePlaceHolder(),
                         primaryKey);
+        SqlTemplate.canHandledByTemplateWithPlaceholder(
+                template,
+                SaveModePlaceHolder.ROWTYPE_UNIQUE_KEY.getPlaceHolder(),
+                uniqueKey,
+                tablePath.getFullName(),
+                DorisOptions.SAVE_MODE_CREATE_TEMPLATE.key());
         template =
                 template.replaceAll(
-                        String.format("\\$\\{%s\\}", SaveModeConstants.ROWTYPE_UNIQUE_KEY),
-                        uniqueKey);
+                        SaveModePlaceHolder.ROWTYPE_UNIQUE_KEY.getReplacePlaceHolder(), uniqueKey);
         Map<String, CreateTableParser.ColumnInfo> columnInTemplate =
                 CreateTableParser.getColumnList(template);
         template = mergeColumnInTemplate(columnInTemplate, tableSchema, template);
@@ -143,14 +156,13 @@ public static String getCreateTableStatement(
                         .map(DorisCatalogUtil::columnToDorisType)
                         .collect(Collectors.joining(",\n"));
         return template.replaceAll(
-                        String.format("\\$\\{%s\\}", SaveModeConstants.DATABASE),
+                        SaveModePlaceHolder.DATABASE.getReplacePlaceHolder(),
                         tablePath.getDatabaseName())
                 .replaceAll(
-                        String.format("\\$\\{%s\\}", SaveModeConstants.TABLE_NAME),
+                        SaveModePlaceHolder.TABLE_NAME.getReplacePlaceHolder(),
                         tablePath.getTableName())
                 .replaceAll(
-                        String.format("\\$\\{%s\\}", SaveModeConstants.ROWTYPE_FIELDS),
-                        rowTypeFields);
+                        SaveModePlaceHolder.ROWTYPE_FIELDS.getReplacePlaceHolder(), rowTypeFields);
     }
 
     private static String mergeColumnInTemplate(

Diff for: seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/DorisCatalogUtil.java

@@ -17,6 +17,7 @@
 
 package org.apache.seatunnel.connectors.doris.catalog;
 
+import org.apache.seatunnel.api.sink.SaveModePlaceHolder;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.catalog.Column;
 import org.apache.seatunnel.api.table.catalog.ConstraintKey;
@@ -28,8 +29,13 @@
 import org.apache.seatunnel.api.table.type.BasicType;
 import org.apache.seatunnel.api.table.type.DecimalType;
 import org.apache.seatunnel.api.table.type.LocalTimeType;
+import org.apache.seatunnel.common.exception.CommonError;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+import org.apache.seatunnel.connectors.doris.config.DorisOptions;
 import org.apache.seatunnel.connectors.doris.util.DorisCatalogUtil;
 
+import org.apache.commons.lang3.StringUtils;
+
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -130,6 +136,38 @@ public void test() {
                         + "\"storage_format\" = \"V2\",\n"
                         + "\"disable_auto_compaction\" = \"false\"\n"
                         + ")");
+
+        String createTemplate = DorisOptions.SAVE_MODE_CREATE_TEMPLATE.defaultValue();
+        CatalogTable catalogTable =
+                CatalogTable.of(
+                        TableIdentifier.of("test", "test1", "test2"),
+                        TableSchema.builder()
+                                .primaryKey(
+                                        PrimaryKey.of(StringUtils.EMPTY, Collections.emptyList()))
+                                .constraintKey(Collections.emptyList())
+                                .columns(columns)
+                                .build(),
+                        Collections.emptyMap(),
+                        Collections.emptyList(),
+                        "");
+        TablePath tablePath = TablePath.of("test1.test2");
+        SeaTunnelRuntimeException actualSeaTunnelRuntimeException =
+                Assertions.assertThrows(
+                        SeaTunnelRuntimeException.class,
+                        () ->
+                                DorisCatalogUtil.getCreateTableStatement(
+                                        createTemplate, tablePath, catalogTable));
+        String primaryKeyHolder = SaveModePlaceHolder.ROWTYPE_PRIMARY_KEY.getPlaceHolder();
+        SeaTunnelRuntimeException exceptSeaTunnelRuntimeException =
+                CommonError.sqlTemplateHandledError(
+                        tablePath.getFullName(),
+                        SaveModePlaceHolder.getDisplay(primaryKeyHolder),
+                        createTemplate,
+                        primaryKeyHolder,
+                        DorisOptions.SAVE_MODE_CREATE_TEMPLATE.key());
+        Assertions.assertEquals(
+                exceptSeaTunnelRuntimeException.getMessage(),
+                actualSeaTunnelRuntimeException.getMessage());
     }
 
     @Test