This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 57c1a39b5a [Feature][Connector-V2] PR2：Support table_options for
StarRocks SaveMode auto-create (#11242)
57c1a39b5a is described below
commit 57c1a39b5a60f846708840233c009c6200cfa5dd
Author: luxiaolong <[email protected]>
AuthorDate: Sun Jul 12 19:43:43 2026 +0800
[Feature][Connector-V2] PR2：Support table_options for StarRocks SaveMode
auto-create (#11242)
Co-authored-by: det101 <[email protected]>
Co-authored-by: Cursor <[email protected]>
---
docs/en/connectors/sink/StarRocks.md | 30 ++++
docs/zh/connectors/sink/StarRocks.md | 30 ++++
.../seatunnel/common/sql/ClauseMergeFormat.java | 35 ++++
.../seatunnel/common/sql/SqlTableClauseMerger.java | 196 +++++++++++++++++++++
.../common/sql/SqlTableClauseMergerTest.java | 90 ++++++++++
.../starrocks/catalog/StarRocksCatalog.java | 18 +-
.../starrocks/catalog/StarRocksCatalogFactory.java | 4 +-
.../starrocks/sink/StarRocksSaveModeUtil.java | 69 ++++++++
.../seatunnel/starrocks/sink/StarRocksSink.java | 9 +-
.../starrocks/sink/StarRocksSinkFactory.java | 15 +-
.../StarRocksTableOptionsConditionExtension.java | 58 ++++++
.../sink/StarRocksTableOptionsValidator.java | 41 +++++
.../starrocks/catalog/PreviewActionTest.java | 40 +++++
.../catalog/StarRocksCreateTableTest.java | 32 ++++
.../StarRocksSaveModeUtilTableOptionsTest.java | 74 ++++++++
...tarRocksTableOptionsConditionExtensionTest.java | 88 +++++++++
.../e2e/connector/starrocks/StarRocksIT.java | 45 +++++
.../fake-to-starrocks-with-table-options.conf | 74 ++++++++
18 files changed, 941 insertions(+), 7 deletions(-)
diff --git a/docs/en/connectors/sink/StarRocks.md
b/docs/en/connectors/sink/StarRocks.md
index 90ad3197a1..e53d77f2cf 100644
--- a/docs/en/connectors/sink/StarRocks.md
+++ b/docs/en/connectors/sink/StarRocks.md
@@ -54,6 +54,7 @@ The internal implementation of StarRocks sink connector is
cached and imported b
| http_socket_timeout_ms | int | no | 180000
| Set http socket timeout, default is 3 minutes.
|
| schema_save_mode | Enum | no |
CREATE_SCHEMA_WHEN_NOT_EXIST | Before the synchronous task is turned on,
different treatment schemes are selected for the existing surface structure of
the target side.
|
| data_save_mode | Enum | no | APPEND_DATA
| Before the synchronous task is turned on, different processing schemes are
selected for data existing data on the target side.
|
+| table_options | Map | no | -
| Sink-specific table properties merged into CREATE TABLE PROPERTIES during
SaveMode auto-create. See below.
|
| custom_sql | String | no | -
| When data_save_mode selects CUSTOM_PROCESSING, you should fill in the
CUSTOM_SQL parameter. This parameter usually fills in a SQL that can be
executed. SQL will be executed before synchronization tasks. |
### save_mode_create_template
@@ -140,6 +141,35 @@ Option introduction：
When data_save_mode selects CUSTOM_PROCESSING, you should fill in the
CUSTOM_SQL parameter. This parameter usually fills in a SQL that can be
executed. SQL will be executed before synchronization tasks.
+### table_options [Map]
+
+Sink-specific table options applied when SaveMode auto-creates the target
table (DDL phase). They take effect only when `schema_save_mode` triggers table
creation, such as `CREATE_SCHEMA_WHEN_NOT_EXIST` or `RECREATE_SCHEMA`. They do
**not** affect Stream Load writes at runtime and do **not** run `ALTER TABLE`
on existing tables.
+
+When used with the default `save_mode_create_template` (option omitted, or
configured with the same content as the built-in default), `table_options` are
merged into the template `PROPERTIES` clause. **Duplicate keys are overridden
by `table_options`.** Use property names from the [StarRocks CREATE TABLE
documentation](https://docs.starrocks.io/docs/sql-reference/sql-statements/table_bucket/partition/CREATE_TABLE/#properties);
SeaTunnel does not maintain an allowlist—invalid properties f [...]
+
+If you configure a `save_mode_create_template` that **differs from the
built-in default**, `table_options` cannot be used together (validation fails
at job submission). Put properties directly in the template instead.
+
+Invalid combinations are validated early via `StarRocksSinkFactory` option
rules (`--check` and job submission), not only when StarRocks executes CREATE
TABLE.
+
+Example:
+
+```hocon
+sink {
+ StarRocks {
+ base-url = "jdbc:mysql://127.0.0.1:9030"
+ nodeUrls = ["127.0.0.1:8030"]
+ username = "root"
+ password = ""
+ database = "test"
+ schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
+ table_options = {
+ replication_num = "3"
+ storage_format = "V2"
+ }
+ }
+}
+```
+
## Data Type Mapping
| StarRocks Data type | SeaTunnel Data type |
diff --git a/docs/zh/connectors/sink/StarRocks.md
b/docs/zh/connectors/sink/StarRocks.md
index 8fc68e0e29..36b34067f1 100644
--- a/docs/zh/connectors/sink/StarRocks.md
+++ b/docs/zh/connectors/sink/StarRocks.md
@@ -53,6 +53,7 @@ StarRocks数据接收器内部实现采用了缓存，通过stream load将数据
| http_socket_timeout_ms | int | no | 180000
| http socket超时时间，默认为3分钟
|
| schema_save_mode | Enum | no | CREATE_SCHEMA_WHEN_NOT_EXIST
| 在同步任务打开之前，针对目标端已存在的表结构选择不同的处理方法
|
| data_save_mode | Enum | no | APPEND_DATA
| 在同步任务打开之前，针对目标端已存在的数据选择不同的处理方法
|
+| table_options | Map | no | -
| SaveMode 自动建表时合并进 CREATE TABLE PROPERTIES 的表级属性，详见下文
|
| custom_sql | String | no | -
|
当data_save_mode设置为CUSTOM_PROCESSING时，必须同时设置CUSTOM_SQL参数。CUSTOM_SQL的值为可执行的SQL语句，在同步任务开启前SQL将会被执行
|
### save_mode_create_template
@@ -134,6 +135,35 @@ table选项参数可以填入一任意表名，这个名字最终会被用作目
当data_save_mode设置为CUSTOM_PROCESSING时，必须同时设置CUSTOM_SQL参数。CUSTOM_SQL的值为可执行的SQL语句，在同步任务开启前SQL将会被执行。
+### table_options [Map]
+
+Sink 在 SaveMode 自动建表（DDL）时附加的表级属性。仅在 `schema_save_mode` 触发建表时生效，例如
`CREATE_SCHEMA_WHEN_NOT_EXIST`、`RECREATE_SCHEMA`；**不影响** Stream Load
写入，也**不会**对已存在表执行 `ALTER TABLE`。
+
+在默认 `save_mode_create_template`（未配置或与内置默认值相同）下，`table_options` 会合并进模板
`PROPERTIES` 子句；**同名 key 以 `table_options` 为准**。属性名请参考 [StarRocks CREATE TABLE
文档](https://docs.starrocks.io/docs/sql-reference/sql-statements/table_bucket/partition/CREATE_TABLE/#properties)；SeaTunnel
不做白名单，非法属性由 StarRocks 执行 DDL 时报错。
+
+若配置了**与内置默认值不同**的 `save_mode_create_template`，则不能与 `table_options`
同时使用（任务提交时校验失败）；此时请将属性直接写入模板。
+
+非法组合会在 `StarRocksSinkFactory` 的 option 规则阶段提前校验（`--check` 与作业提交），而非仅在
StarRocks 执行 CREATE TABLE 时失败。
+
+示例：
+
+```hocon
+sink {
+ StarRocks {
+ base-url = "jdbc:mysql://127.0.0.1:9030"
+ nodeUrls = ["127.0.0.1:8030"]
+ username = "root"
+ password = ""
+ database = "test"
+ schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
+ table_options = {
+ replication_num = "3"
+ storage_format = "V2"
+ }
+ }
+}
+```
+
## 数据类型映射
| StarRocks数据类型 | SeaTunnel数据类型 |
diff --git
a/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/sql/ClauseMergeFormat.java
b/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/sql/ClauseMergeFormat.java
new file mode 100644
index 0000000000..a8f8f477a9
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/sql/ClauseMergeFormat.java
@@ -0,0 +1,35 @@
+/*
+ * 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;
+
+/** Format definition for SQL table property clauses merged by {@link
SqlTableClauseMerger}. */
+public enum ClauseMergeFormat {
+
+ /** StarRocks/Doris style: {@code PROPERTIES ("key" = "value")}. */
+ DOUBLE_QUOTED_PROPERTIES("PROPERTIES");
+
+ private final String keyword;
+
+ ClauseMergeFormat(String keyword) {
+ this.keyword = keyword;
+ }
+
+ public String getKeyword() {
+ return keyword;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/sql/SqlTableClauseMerger.java
b/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/sql/SqlTableClauseMerger.java
new file mode 100644
index 0000000000..532f22cc88
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/sql/SqlTableClauseMerger.java
@@ -0,0 +1,196 @@
+/*
+ * 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;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/** Utility for merging key-value table options into SQL CREATE TABLE property
clauses. */
+public final class SqlTableClauseMerger {
+
+ private static final Pattern DOUBLE_QUOTED_ENTRY_PATTERN =
+
Pattern.compile("\"((?:[^\"\\\\]|\\\\.)*)\"\\s*=\\s*\"((?:[^\"\\\\]|\\\\.)*)\"");
+
+ private SqlTableClauseMerger() {}
+
+ public static String merge(
+ String sql, ClauseMergeFormat format, Map<String, String>
tableOptions) {
+ if (tableOptions == null || tableOptions.isEmpty()) {
+ return sql;
+ }
+ switch (format) {
+ case DOUBLE_QUOTED_PROPERTIES:
+ return mergeDoubleQuotedProperties(sql, format.getKeyword(),
tableOptions);
+ default:
+ throw new IllegalArgumentException("Unsupported clause merge
format: " + format);
+ }
+ }
+
+ private static String mergeDoubleQuotedProperties(
+ String sql, String keyword, Map<String, String> tableOptions) {
+ int keywordStart = findLastKeywordPosition(sql, keyword);
+ if (keywordStart >= 0) {
+ int openParen = indexOfNonWhitespace(sql, keywordStart +
keyword.length());
+ if (openParen >= 0 && sql.charAt(openParen) == '(') {
+ int closeParen = findMatchingCloseParen(sql, openParen);
+ if (closeParen > openParen) {
+ String clauseBody = sql.substring(openParen + 1,
closeParen);
+ Map<String, String> merged =
parseDoubleQuotedEntries(clauseBody);
+ merged.putAll(tableOptions);
+ String newClause = renderDoubleQuotedProperties(keyword,
merged);
+ return sql.substring(0, keywordStart)
+ + newClause
+ + sql.substring(closeParen + 1);
+ }
+ }
+ }
+ return appendDoubleQuotedProperties(sql, keyword, tableOptions);
+ }
+
+ private static String appendDoubleQuotedProperties(
+ String sql, String keyword, Map<String, String> tableOptions) {
+ String trimmed = sql.trim();
+ boolean endsWithSemicolon = trimmed.endsWith(";");
+ if (endsWithSemicolon) {
+ trimmed = trimmed.substring(0, trimmed.length() - 1).trim();
+ }
+ StringBuilder builder = new StringBuilder(trimmed);
+ if (!trimmed.isEmpty()) {
+ builder.append('\n');
+ }
+ builder.append(renderDoubleQuotedProperties(keyword, tableOptions));
+ if (endsWithSemicolon) {
+ builder.append(';');
+ }
+ return builder.toString();
+ }
+
+ private static String renderDoubleQuotedProperties(
+ String keyword, Map<String, String> properties) {
+ StringBuilder builder = new StringBuilder(keyword).append(" (\n");
+ boolean first = true;
+ for (Map.Entry<String, String> entry : properties.entrySet()) {
+ if (!first) {
+ builder.append(",\n");
+ }
+ builder.append(" \"")
+ .append(escapeDoubleQuoted(entry.getKey()))
+ .append("\" = \"")
+ .append(escapeDoubleQuoted(entry.getValue()))
+ .append('\"');
+ first = false;
+ }
+ builder.append("\n)");
+ return builder.toString();
+ }
+
+ private static Map<String, String> parseDoubleQuotedEntries(String
clauseBody) {
+ Map<String, String> properties = new LinkedHashMap<>();
+ Matcher matcher = DOUBLE_QUOTED_ENTRY_PATTERN.matcher(clauseBody);
+ while (matcher.find()) {
+ properties.put(
+ unescapeDoubleQuoted(matcher.group(1)),
unescapeDoubleQuoted(matcher.group(2)));
+ }
+ return properties;
+ }
+
+ private static int findLastKeywordPosition(String sql, String keyword) {
+ Pattern pattern = Pattern.compile("(?i)\\b" + Pattern.quote(keyword) +
"\\b");
+ Matcher matcher = pattern.matcher(sql);
+ int lastPosition = -1;
+ while (matcher.find()) {
+ lastPosition = matcher.start();
+ }
+ return lastPosition;
+ }
+
+ private static int indexOfNonWhitespace(String sql, int fromIndex) {
+ for (int i = fromIndex; i < sql.length(); i++) {
+ if (!Character.isWhitespace(sql.charAt(i))) {
+ return i;
+ }
+ }
+ return -1;
+ }
+
+ private static int findMatchingCloseParen(String sql, int openParenIndex) {
+ int depth = 0;
+ boolean inSingleQuote = false;
+ boolean inDoubleQuote = false;
+ for (int i = openParenIndex; i < sql.length(); i++) {
+ char current = sql.charAt(i);
+ if (inSingleQuote) {
+ if (current == '\'' && !isEscaped(sql, i)) {
+ inSingleQuote = false;
+ }
+ continue;
+ }
+ if (inDoubleQuote) {
+ if (current == '"' && !isEscaped(sql, i)) {
+ inDoubleQuote = false;
+ }
+ continue;
+ }
+ if (current == '\'') {
+ inSingleQuote = true;
+ continue;
+ }
+ if (current == '"') {
+ inDoubleQuote = true;
+ continue;
+ }
+ if (current == '(') {
+ depth++;
+ } else if (current == ')') {
+ depth--;
+ if (depth == 0) {
+ return i;
+ }
+ }
+ }
+ return -1;
+ }
+
+ private static boolean isEscaped(String sql, int index) {
+ int backslashCount = 0;
+ for (int i = index - 1; i >= 0 && sql.charAt(i) == '\\'; i--) {
+ backslashCount++;
+ }
+ return backslashCount % 2 == 1;
+ }
+
+ private static String escapeDoubleQuoted(String value) {
+ return value.replace("\\", "\\\\").replace("\"", "\\\"");
+ }
+
+ private static String unescapeDoubleQuoted(String value) {
+ StringBuilder builder = new StringBuilder();
+ for (int i = 0; i < value.length(); i++) {
+ char current = value.charAt(i);
+ if (current == '\\' && i + 1 < value.length()) {
+ builder.append(value.charAt(i + 1));
+ i++;
+ } else {
+ builder.append(current);
+ }
+ }
+ return builder.toString();
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/common/sql/SqlTableClauseMergerTest.java
b/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/common/sql/SqlTableClauseMergerTest.java
new file mode 100644
index 0000000000..0640f0ab54
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/common/sql/SqlTableClauseMergerTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+class SqlTableClauseMergerTest {
+
+ @Test
+ void mergeOverridesExistingProperty() {
+ String sql =
+ "CREATE TABLE t (\n"
+ + " id INT\n"
+ + ") ENGINE=OLAP\n"
+ + "PROPERTIES (\n"
+ + " \"replication_num\" = \"1\"\n"
+ + ")";
+ Map<String, String> tableOptions = new LinkedHashMap<>();
+ tableOptions.put("replication_num", "3");
+
+ String merged =
+ SqlTableClauseMerger.merge(
+ sql, ClauseMergeFormat.DOUBLE_QUOTED_PROPERTIES,
tableOptions);
+
+ Assertions.assertTrue(merged.contains("\"replication_num\" = \"3\""));
+ Assertions.assertFalse(merged.contains("\"replication_num\" = \"1\""));
+ }
+
+ @Test
+ void mergeAddsNewProperty() {
+ String sql =
+ "CREATE TABLE t (\n"
+ + " id INT\n"
+ + ") ENGINE=OLAP\n"
+ + "PROPERTIES (\n"
+ + " \"replication_num\" = \"1\"\n"
+ + ")";
+ Map<String, String> tableOptions = new LinkedHashMap<>();
+ tableOptions.put("storage_format", "V2");
+
+ String merged =
+ SqlTableClauseMerger.merge(
+ sql, ClauseMergeFormat.DOUBLE_QUOTED_PROPERTIES,
tableOptions);
+
+ Assertions.assertTrue(merged.contains("\"replication_num\" = \"1\""));
+ Assertions.assertTrue(merged.contains("\"storage_format\" = \"V2\""));
+ }
+
+ @Test
+ void appendPropertiesWhenClauseMissing() {
+ String sql = "CREATE TABLE t (id INT) ENGINE=OLAP";
+ Map<String, String> tableOptions = new LinkedHashMap<>();
+ tableOptions.put("replication_num", "3");
+
+ String merged =
+ SqlTableClauseMerger.merge(
+ sql, ClauseMergeFormat.DOUBLE_QUOTED_PROPERTIES,
tableOptions);
+
+ Assertions.assertTrue(merged.contains("PROPERTIES ("));
+ Assertions.assertTrue(merged.contains("\"replication_num\" = \"3\""));
+ }
+
+ @Test
+ void emptyOptionsReturnsOriginalSql() {
+ String sql = "CREATE TABLE t (id INT) ENGINE=OLAP";
+ Assertions.assertEquals(
+ sql,
+ SqlTableClauseMerger.merge(
+ sql, ClauseMergeFormat.DOUBLE_QUOTED_PROPERTIES, new
LinkedHashMap<>()));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/StarRocksCatalog.java
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/StarRocksCatalog.java
index 1b35a76c3a..5699b45825 100644
---
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/StarRocksCatalog.java
+++
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/StarRocksCatalog.java
@@ -79,12 +79,23 @@ public class StarRocksCatalog implements Catalog {
protected String defaultUrl;
private final JdbcUrlUtil.UrlInfo urlInfo;
private final String template;
+ private final Map<String, String> tableOptions;
private Connection conn;
private static final Logger LOG =
LoggerFactory.getLogger(StarRocksCatalog.class);
public StarRocksCatalog(
String catalogName, String username, String pwd, String
defaultUrl, String template) {
+ this(catalogName, username, pwd, defaultUrl, template,
Collections.emptyMap());
+ }
+
+ public StarRocksCatalog(
+ String catalogName,
+ String username,
+ String pwd,
+ String defaultUrl,
+ String template,
+ Map<String, String> tableOptions) {
checkArgument(StringUtils.isNotBlank(username));
checkArgument(StringUtils.isNotBlank(defaultUrl));
@@ -98,6 +109,7 @@ public class StarRocksCatalog implements Catalog {
this.username = username;
this.pwd = pwd;
this.template = template;
+ this.tableOptions = tableOptions;
}
@Override
@@ -210,7 +222,8 @@ public class StarRocksCatalog implements Catalog {
tablePath.getTableName(),
table.getTableSchema(),
table.getComment(),
- StarRocksSinkOptions.SAVE_MODE_CREATE_TEMPLATE.key()));
+ StarRocksSinkOptions.SAVE_MODE_CREATE_TEMPLATE.key(),
+ tableOptions));
}
@Override
@@ -514,7 +527,8 @@ public class StarRocksCatalog implements Catalog {
tablePath.getTableName(),
catalogTable.get().getTableSchema(),
catalogTable.get().getComment(),
-
StarRocksSinkOptions.SAVE_MODE_CREATE_TEMPLATE.key()));
+
StarRocksSinkOptions.SAVE_MODE_CREATE_TEMPLATE.key(),
+ tableOptions));
} else if (actionType == ActionType.DROP_TABLE) {
return new SQLPreviewResult(
StarRocksSaveModeUtil.INSTANCE.getDropTableSql(tablePath,
true));
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/StarRocksCatalogFactory.java
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/StarRocksCatalogFactory.java
index 3ad81b68d1..ba06900968 100644
---
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/StarRocksCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/StarRocksCatalogFactory.java
@@ -19,6 +19,7 @@ package
org.apache.seatunnel.connectors.seatunnel.starrocks.catalog;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.options.SinkConnectorCommonOptions;
import org.apache.seatunnel.api.table.catalog.Catalog;
import org.apache.seatunnel.api.table.factory.CatalogFactory;
import org.apache.seatunnel.api.table.factory.Factory;
@@ -38,7 +39,8 @@ public class StarRocksCatalogFactory implements
CatalogFactory {
options.get(StarRocksSourceOptions.USERNAME),
options.get(StarRocksSourceOptions.PASSWORD),
options.get(StarRocksSinkOptions.BASE_URL),
- options.get(StarRocksSinkOptions.SAVE_MODE_CREATE_TEMPLATE));
+ options.get(StarRocksSinkOptions.SAVE_MODE_CREATE_TEMPLATE),
+ options.get(SinkConnectorCommonOptions.TABLE_OPTIONS));
}
@Override
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSaveModeUtil.java
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSaveModeUtil.java
index 8f160d6c84..80970a2865 100644
---
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSaveModeUtil.java
+++
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSaveModeUtil.java
@@ -19,14 +19,23 @@ package
org.apache.seatunnel.connectors.seatunnel.starrocks.sink;
import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
+import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.table.catalog.Column;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.type.ArrayType;
import org.apache.seatunnel.api.table.type.DecimalType;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+import org.apache.seatunnel.connectors.seatunnel.common.sql.ClauseMergeFormat;
+import
org.apache.seatunnel.connectors.seatunnel.common.sql.SqlTableClauseMerger;
import org.apache.seatunnel.connectors.seatunnel.common.util.CatalogUtil;
+import
org.apache.seatunnel.connectors.seatunnel.starrocks.config.StarRocksSinkOptions;
import lombok.extern.slf4j.Slf4j;
+import java.util.Map;
+
import static
org.apache.seatunnel.shade.com.google.common.base.Preconditions.checkNotNull;
@Slf4j
@@ -34,6 +43,66 @@ public class StarRocksSaveModeUtil extends CatalogUtil {
public static final StarRocksSaveModeUtil INSTANCE = new
StarRocksSaveModeUtil();
+ private StarRocksSaveModeUtil() {}
+
+ public String getCreateTableSql(
+ String template,
+ String database,
+ String table,
+ TableSchema tableSchema,
+ String comment,
+ String optionsKey,
+ Map<String, String> tableOptions) {
+ String createTableSql =
+ getCreateTableSql(template, database, table, tableSchema,
comment, optionsKey);
+ return applyTableOptionsToCreateTableSql(createTableSql, tableOptions);
+ }
+
+ public void validateTableOptions(ReadonlyConfig config, Map<String,
String> tableOptions) {
+ if (tableOptions == null || tableOptions.isEmpty()) {
+ return;
+ }
+ if (isCustomCreateTemplate(config)) {
+ throw new SeaTunnelRuntimeException(
+ SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+ "table_options cannot be used together with a custom
save_mode_create_template"
+ + " for StarRocks sink.");
+ }
+ for (Map.Entry<String, String> entry : tableOptions.entrySet()) {
+ if (StringUtils.isBlank(entry.getKey())) {
+ throw new SeaTunnelRuntimeException(
+ SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+ "table_options contains a blank property key for
StarRocks sink.");
+ }
+ if (entry.getValue() == null) {
+ throw new SeaTunnelRuntimeException(
+ SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+ String.format(
+ "table_options property '%s' has null value
for StarRocks sink.",
+ entry.getKey()));
+ }
+ }
+ }
+
+ public String applyTableOptionsToCreateTableSql(
+ String createTableSql, Map<String, String> tableOptions) {
+ if (tableOptions == null || tableOptions.isEmpty()) {
+ return createTableSql;
+ }
+ return SqlTableClauseMerger.merge(
+ createTableSql, ClauseMergeFormat.DOUBLE_QUOTED_PROPERTIES,
tableOptions);
+ }
+
+ private static boolean isCustomCreateTemplate(ReadonlyConfig config) {
+ return
config.getOptional(StarRocksSinkOptions.SAVE_MODE_CREATE_TEMPLATE)
+ .map(
+ template ->
+ !template.equals(
+
StarRocksSinkOptions.SAVE_MODE_CREATE_TEMPLATE
+ .defaultValue()))
+ .orElse(false);
+ }
+
public String columnToConnectorType(Column column) {
checkNotNull(column, "The column is required.");
String columnType;
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSink.java
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSink.java
index 1b89ef3f4e..b856ca1e64 100644
---
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSink.java
+++
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSink.java
@@ -41,6 +41,7 @@ import lombok.extern.slf4j.Slf4j;
import java.util.Arrays;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
@Slf4j
@@ -52,11 +53,14 @@ public class StarRocksSink extends
AbstractSimpleSink<SeaTunnelRow, Void>
private final DataSaveMode dataSaveMode;
private final SchemaSaveMode schemaSaveMode;
private final CatalogTable catalogTable;
+ private final Map<String, String> tableOptions;
- public StarRocksSink(SinkConfig sinkConfig, CatalogTable catalogTable) {
+ public StarRocksSink(
+ SinkConfig sinkConfig, CatalogTable catalogTable, Map<String,
String> tableOptions) {
this.sinkConfig = sinkConfig;
this.tableSchema = catalogTable.getTableSchema();
this.catalogTable = catalogTable;
+ this.tableOptions = tableOptions;
this.dataSaveMode = sinkConfig.getDataSaveMode();
this.schemaSaveMode = sinkConfig.getSchemaSaveMode();
// Load the JDBC driver in to DriverManager
@@ -103,7 +107,8 @@ public class StarRocksSink extends
AbstractSimpleSink<SeaTunnelRow, Void>
sinkConfig.getUsername(),
sinkConfig.getPassword(),
sinkConfig.getJdbcUrl(),
- sinkConfig.getSaveModeCreateTemplate());
+ sinkConfig.getSaveModeCreateTemplate(),
+ tableOptions);
return Optional.of(
new DefaultSaveModeHandler(
schemaSaveMode,
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkFactory.java
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkFactory.java
index a6efa8bfef..a2b0c4e06d 100644
---
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkFactory.java
+++
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkFactory.java
@@ -19,7 +19,10 @@ package
org.apache.seatunnel.connectors.seatunnel.starrocks.sink;
import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.Conditions;
import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.options.SinkConnectorCommonOptions;
import org.apache.seatunnel.api.sink.DataSaveMode;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.TableIdentifier;
@@ -35,6 +38,7 @@ import com.google.auto.service.AutoService;
import java.util.Arrays;
import java.util.List;
+import java.util.Map;
import static
org.apache.seatunnel.api.options.SinkConnectorCommonOptions.MULTI_TABLE_SINK_REPLICA;
import static
org.apache.seatunnel.connectors.seatunnel.starrocks.config.StarRocksSinkOptions.DATA_SAVE_MODE;
@@ -67,6 +71,11 @@ public class StarRocksSinkFactory implements
TableSinkFactory {
MULTI_TABLE_SINK_REPLICA,
StarRocksSinkOptions.SAVE_MODE_CREATE_TEMPLATE,
StarRocksSinkOptions.HTTP_SOCKET_TIMEOUT_MS)
+ .optional(
+ SinkConnectorCommonOptions.TABLE_OPTIONS,
+ Conditions.extension(
+ SinkConnectorCommonOptions.TABLE_OPTIONS,
+
StarRocksTableOptionsConditionExtension.INSTANCE))
.conditional(
DATA_SAVE_MODE,
DataSaveMode.CUSTOM_PROCESSING,
@@ -81,8 +90,10 @@ public class StarRocksSinkFactory implements
TableSinkFactory {
@Override
public TableSink createSink(TableSinkFactoryContext context) {
+ ReadonlyConfig config = context.getOptions();
+ Map<String, String> sinkTableOptions =
config.get(SinkConnectorCommonOptions.TABLE_OPTIONS);
CatalogTable catalogTable = context.getCatalogTable();
- SinkConfig sinkConfig = SinkConfig.of(context.getOptions());
+ SinkConfig sinkConfig = SinkConfig.of(config);
if (StringUtils.isBlank(sinkConfig.getTable())) {
sinkConfig.setTable(catalogTable.getTableId().getTableName());
}
@@ -101,6 +112,6 @@ public class StarRocksSinkFactory implements
TableSinkFactory {
catalogTable.getPartitionKeys(),
catalogTable.getComment());
- return () -> new StarRocksSink(sinkConfig, finalCatalogTable);
+ return () -> new StarRocksSink(sinkConfig, finalCatalogTable,
sinkTableOptions);
}
}
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksTableOptionsConditionExtension.java
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksTableOptionsConditionExtension.java
new file mode 100644
index 0000000000..7a08d2f051
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksTableOptionsConditionExtension.java
@@ -0,0 +1,58 @@
+/*
+ * 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.sink;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+
+import java.util.Map;
+
+/**
+ * Early validation for StarRocks sink {@code table_options}. Delegates to
{@link
+ * StarRocksTableOptionsValidator}.
+ */
+public class StarRocksTableOptionsConditionExtension
+ implements ConditionExtension<Map<String, String>> {
+
+ public static final StarRocksTableOptionsConditionExtension INSTANCE =
+ new StarRocksTableOptionsConditionExtension();
+
+ private StarRocksTableOptionsConditionExtension() {}
+
+ @Override
+ public String description() {
+ return "must be compatible with the default save_mode_create_template
(see StarRocks"
+ + " connector docs)";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, Map<String, String> value)
+ throws OptionValidationException {
+ if (value == null || value.isEmpty()) {
+ return true;
+ }
+ try {
+ StarRocksTableOptionsValidator.validate(config, value);
+ return true;
+ } catch (SeaTunnelRuntimeException e) {
+ throw new OptionValidationException(e.getMessage());
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksTableOptionsValidator.java
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksTableOptionsValidator.java
new file mode 100644
index 0000000000..134f4fd544
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksTableOptionsValidator.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.starrocks.sink;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.options.SinkConnectorCommonOptions;
+
+import java.util.Collections;
+import java.util.Map;
+
+/** Validates StarRocks sink {@code table_options} before job submission. */
+public final class StarRocksTableOptionsValidator {
+
+ private StarRocksTableOptionsValidator() {}
+
+ public static void validate(ReadonlyConfig config, Map<String, String>
tableOptions) {
+ StarRocksSaveModeUtil.INSTANCE.validateTableOptions(config,
tableOptions);
+ }
+
+ public static void validate(ReadonlyConfig config) {
+ validate(
+ config,
+ config.getOptional(SinkConnectorCommonOptions.TABLE_OPTIONS)
+ .orElse(Collections.emptyMap()));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/PreviewActionTest.java
b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/PreviewActionTest.java
index ba55560865..022e24f169 100644
---
a/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/PreviewActionTest.java
+++
b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/PreviewActionTest.java
@@ -28,12 +28,14 @@ import
org.apache.seatunnel.api.table.catalog.TableIdentifier;
import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.type.BasicType;
+import
org.apache.seatunnel.connectors.seatunnel.starrocks.config.StarRocksSinkOptions;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.Collections;
import java.util.HashMap;
+import java.util.Map;
import java.util.Optional;
public class PreviewActionTest {
@@ -113,6 +115,44 @@ public class PreviewActionTest {
Optional.of(CATALOG_TABLE));
}
+ @Test
+ public void testCreateTablePreviewUsesSinkTableOptionsOnly() {
+ Map<String, String> upstreamOptions = new HashMap<>();
+ upstreamOptions.put("connector", "mysql");
+ upstreamOptions.put("engine", "InnoDB");
+
+ CatalogTable tableWithUpstreamOptions =
+ CatalogTable.of(
+ CATALOG_TABLE.getTableId(),
+ CATALOG_TABLE.getTableSchema(),
+ upstreamOptions,
+ CATALOG_TABLE.getPartitionKeys(),
+ CATALOG_TABLE.getComment());
+
+ Map<String, String> sinkTableOptions = new HashMap<>();
+ sinkTableOptions.put("replication_num", "3");
+
+ StarRocksCatalog catalog =
+ new StarRocksCatalog(
+ "test",
+ "root",
+ "root",
+ "jdbc:mysql://localhost:9030",
+
StarRocksSinkOptions.SAVE_MODE_CREATE_TEMPLATE.defaultValue(),
+ sinkTableOptions);
+
+ PreviewResult previewResult =
+ catalog.previewAction(
+ Catalog.ActionType.CREATE_TABLE,
+ TablePath.of("testddatabase.testtable"),
+ Optional.of(tableWithUpstreamOptions));
+
+ String sql = ((SQLPreviewResult) previewResult).getSql();
+ Assertions.assertTrue(sql.contains("\"replication_num\" = \"3\""));
+ Assertions.assertFalse(sql.contains("InnoDB"));
+ Assertions.assertFalse(sql.contains("\"connector\""));
+ }
+
private void assertPreviewResult(
Catalog catalog,
Catalog.ActionType actionType,
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/StarRocksCreateTableTest.java
b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/StarRocksCreateTableTest.java
index ee13e2ac7d..d1cc37003c 100644
---
a/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/StarRocksCreateTableTest.java
+++
b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/StarRocksCreateTableTest.java
@@ -44,7 +44,9 @@ import lombok.extern.slf4j.Slf4j;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
@Slf4j
public class StarRocksCreateTableTest {
@@ -426,4 +428,34 @@ public class StarRocksCreateTableTest {
+ ")\n",
result);
}
+
+ @Test
+ public void testCreateTableWithTableOptions() {
+ Map<String, String> tableOptions = new HashMap<>();
+ tableOptions.put("replication_num", "3");
+ tableOptions.put("storage_format", "V2");
+
+ String result =
+ StarRocksSaveModeUtil.INSTANCE.getCreateTableSql(
+
StarRocksSinkOptions.SAVE_MODE_CREATE_TEMPLATE.defaultValue(),
+ "testdb",
+ "test_table",
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "id",
+ BasicType.INT_TYPE,
+ (Long) null,
+ false,
+ null,
+ ""))
+ .primaryKey(PrimaryKey.of("test",
Collections.singletonList("id")))
+ .build(),
+ "test table",
+ StarRocksSinkOptions.SAVE_MODE_CREATE_TEMPLATE.key(),
+ tableOptions);
+
+ Assertions.assertTrue(result.contains("\"replication_num\" = \"3\""));
+ Assertions.assertTrue(result.contains("\"storage_format\" = \"V2\""));
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSaveModeUtilTableOptionsTest.java
b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSaveModeUtilTableOptionsTest.java
new file mode 100644
index 0000000000..c64937321c
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSaveModeUtilTableOptionsTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.sink;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+import
org.apache.seatunnel.connectors.seatunnel.starrocks.config.StarRocksSinkOptions;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+class StarRocksSaveModeUtilTableOptionsTest {
+
+ @Test
+ void validateAllowsDefaultTemplateWithTableOptions() {
+ ReadonlyConfig config = ReadonlyConfig.fromMap(new HashMap<>());
+ Map<String, String> tableOptions =
Collections.singletonMap("replication_num", "3");
+
+ Assertions.assertDoesNotThrow(
+ () ->
StarRocksSaveModeUtil.INSTANCE.validateTableOptions(config, tableOptions));
+ }
+
+ @Test
+ void validateRejectsCustomTemplateWithTableOptions() {
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put(
+ StarRocksSinkOptions.SAVE_MODE_CREATE_TEMPLATE.key(),
+ "CREATE TABLE `${database}`.`${table}` (${rowtype_fields})");
+ ReadonlyConfig config = ReadonlyConfig.fromMap(configMap);
+ Map<String, String> tableOptions =
Collections.singletonMap("replication_num", "3");
+
+ SeaTunnelRuntimeException exception =
+ Assertions.assertThrows(
+ SeaTunnelRuntimeException.class,
+ () ->
+
StarRocksSaveModeUtil.INSTANCE.validateTableOptions(
+ config, tableOptions));
+
+ Assertions.assertTrue(exception.getMessage().contains("custom
save_mode_create_template"));
+ }
+
+ @Test
+ void applyMergesPropertiesIntoSql() {
+ String sql =
+ "CREATE TABLE t (id INT) ENGINE=OLAP PROPERTIES (\n"
+ + " \"replication_num\" = \"1\"\n"
+ + ")";
+ Map<String, String> tableOptions =
Collections.singletonMap("replication_num", "3");
+
+ String merged =
+
StarRocksSaveModeUtil.INSTANCE.applyTableOptionsToCreateTableSql(sql,
tableOptions);
+
+ Assertions.assertTrue(merged.contains("\"replication_num\" = \"3\""));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksTableOptionsConditionExtensionTest.java
b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksTableOptionsConditionExtensionTest.java
new file mode 100644
index 0000000000..47cade3904
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksTableOptionsConditionExtensionTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.sink;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConfigValidator;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
+import org.apache.seatunnel.api.options.SinkConnectorCommonOptions;
+import
org.apache.seatunnel.connectors.seatunnel.starrocks.config.StarRocksSinkOptions;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+class StarRocksTableOptionsConditionExtensionTest {
+
+ @Test
+ void testDefaultTemplateWithTableOptionsPassViaOptionRule() {
+ Map<String, Object> config = starRocksSinkConfig();
+ Map<String, String> tableOptions = new HashMap<>();
+ tableOptions.put("replication_num", "3");
+ config.put(SinkConnectorCommonOptions.TABLE_OPTIONS.key(),
tableOptions);
+
+ Assertions.assertDoesNotThrow(() -> validateSinkOptionRule(config));
+ }
+
+ @Test
+ void testCustomTemplateRejectsTableOptionsViaOptionRule() {
+ Map<String, Object> config = starRocksSinkConfig();
+ config.put(
+ StarRocksSinkOptions.SAVE_MODE_CREATE_TEMPLATE.key(),
+ "CREATE TABLE `${database}`.`${table}` (${rowtype_fields})");
+ Map<String, String> tableOptions = new HashMap<>();
+ tableOptions.put("replication_num", "3");
+ config.put(SinkConnectorCommonOptions.TABLE_OPTIONS.key(),
tableOptions);
+
+ OptionValidationException exception =
+ Assertions.assertThrows(
+ OptionValidationException.class, () ->
validateSinkOptionRule(config));
+ Assertions.assertTrue(exception.getMessage().contains("custom
save_mode_create_template"));
+ }
+
+ @Test
+ void testAbsentTableOptionsSkipsExtension() {
+ Assertions.assertDoesNotThrow(() ->
validateSinkOptionRule(starRocksSinkConfig()));
+ }
+
+ @Test
+ void testEmptyTableOptionsSkipsExtension() {
+ Map<String, Object> config = starRocksSinkConfig();
+ config.put(SinkConnectorCommonOptions.TABLE_OPTIONS.key(), new
HashMap<>());
+
+ Assertions.assertDoesNotThrow(() -> validateSinkOptionRule(config));
+ }
+
+ private static void validateSinkOptionRule(Map<String, Object> config) {
+ ConfigValidator.of(ReadonlyConfig.fromMap(config))
+ .validate(new StarRocksSinkFactory().optionRule());
+ }
+
+ private static Map<String, Object> starRocksSinkConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("base-url", "jdbc:mysql://127.0.0.1:9030");
+ config.put("nodeUrls", Arrays.asList("127.0.0.1:8030"));
+ config.put("username", "root");
+ config.put("password", "");
+ config.put("database", "test");
+ return config;
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksIT.java
index 872ce538c4..1af0d27988 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksIT.java
@@ -78,6 +78,9 @@ public class StarRocksIT extends TestSuiteBase implements
TestResource {
private static final String SOURCE_TABLE = "e2e_table_source";
private static final String SOURCE_TABLE_3 = "e2e_table_source_3";
private static final String SINK_TABLE = "e2e_table_sink";
+ private static final String TABLE_OPTIONS_SINK_TABLE =
"sink_table_options";
+ private static final String TABLE_OPTIONS_CONFIG_FILE =
+ "/fake-to-starrocks-with-table-options.conf";
private static final String SR_DRIVER_JAR =
"https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.16/mysql-connector-java-8.0.16.jar";;
private static final String COLUMN_STRING =
@@ -342,6 +345,19 @@ public class StarRocksIT extends TestSuiteBase implements
TestResource {
Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
}
+ @TestTemplate
+ public void testTableOptionsSink(TestContainer container)
+ throws IOException, InterruptedException, SQLException {
+ try {
+ dropTableIfExists(TABLE_OPTIONS_SINK_TABLE);
+ Container.ExecResult execResult =
container.executeJob(TABLE_OPTIONS_CONFIG_FILE);
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+ assertTableOptionsSink();
+ } finally {
+ dropTableIfExists(TABLE_OPTIONS_SINK_TABLE);
+ }
+ }
+
private void initializeJdbcConnection()
throws SQLException, ClassNotFoundException, MalformedURLException,
InstantiationException, IllegalAccessException {
@@ -409,6 +425,35 @@ public class StarRocksIT extends TestSuiteBase implements
TestResource {
}
}
+ private void assertTableOptionsSink() throws SQLException {
+ try (Statement statement = jdbcConnection.createStatement()) {
+ ResultSet createTableResult =
+ statement.executeQuery(
+ String.format(
+ "SHOW CREATE TABLE %s.%s", DATABASE,
TABLE_OPTIONS_SINK_TABLE));
+ Assertions.assertTrue(createTableResult.next());
+ String createTableSql =
createTableResult.getString(2).toLowerCase();
+ Assertions.assertTrue(
+ createTableSql.contains("\"storage_format\" = \"v2\""),
createTableSql);
+
+ ResultSet countResult =
+ statement.executeQuery(
+ String.format(
+ "SELECT COUNT(*) FROM %s.%s",
+ DATABASE, TABLE_OPTIONS_SINK_TABLE));
+ Assertions.assertTrue(countResult.next());
+ Assertions.assertEquals(100, countResult.getInt(1));
+ }
+ }
+
+ private void dropTableIfExists(String table) {
+ try (Statement statement = jdbcConnection.createStatement()) {
+ statement.execute(String.format("DROP TABLE IF EXISTS %s.%s",
DATABASE, table));
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to drop table " + table, e);
+ }
+ }
+
@Test
public void testCatalog() {
TablePath tablePathStarRocksSource = TablePath.of("test",
"e2e_table_source");
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/resources/fake-to-starrocks-with-table-options.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/resources/fake-to-starrocks-with-table-options.conf
new file mode 100644
index 0000000000..107fac6770
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/resources/fake-to-starrocks-with-table-options.conf
@@ -0,0 +1,74 @@
+#
+# 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.
+#
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ parallelism = 1
+ plugin_output = "fake"
+ row.num = 100
+ schema {
+ table = "FakeTable"
+ primaryKey {
+ name = "pk_id"
+ columnNames = [id]
+ }
+ columns = [
+ {
+ name = id
+ type = bigint
+ nullable = false
+ },
+ {
+ name = c_string
+ type = string
+ nullable = true
+ }
+ ]
+ }
+ }
+}
+
+transform {
+}
+
+sink {
+ StarRocks {
+ plugin_input = "fake"
+ nodeUrls = ["starrocks_e2e:8030"]
+ username = root
+ password = ""
+ database = "test"
+ table = "sink_table_options"
+ batch_max_rows = 100
+ max_retries = 3
+ base-url="jdbc:mysql://starrocks_e2e:9030/test"
+ starrocks.config = {
+ format = "JSON"
+ strip_outer_array = true
+ }
+ schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
+ data_save_mode = "APPEND_DATA"
+ table_options = {
+ storage_format = "V2"
+ }
+ }
+}
