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 3ff3fcf53b [Feature][Connector-V2] Extend JDBC sink table_options to
TiDB and OceanBase MySQL (#11388)
3ff3fcf53b is described below
commit 3ff3fcf53bce2db733a33e07f677788a700b1469
Author: luxiaolong <[email protected]>
AuthorDate: Sun Jul 12 19:46:13 2026 +0800
[Feature][Connector-V2] Extend JDBC sink table_options to TiDB and
OceanBase MySQL (#11388)
Co-authored-by: det101 <[email protected]>
Co-authored-by: Cursor <[email protected]>
---
docs/en/connectors/sink/Jdbc.md | 10 +++
docs/zh/connectors/sink/Jdbc.md | 10 +++
.../OceanBaseMysqlCreateTableSqlBuilder.java | 7 ++-
.../jdbc/internal/dialect/mysql/MysqlDialect.java | 25 +-------
.../dialect/mysql/MysqlFamilyTableOptions.java | 55 ++++++++++++++++
.../dialect/oceanbase/OceanBaseMysqlDialect.java | 6 ++
.../OceanBaseMysqlCreateTableSqlBuilderTest.java | 71 +++++++++++++++++++++
.../jdbc/catalog/tidb/TiDBTableOptionsTest.java | 73 ++++++++++++++++++++++
.../oceanbase/OceanBaseMysqlDialectTest.java | 55 ++++++++++++++++
.../JdbcTableOptionsConditionExtensionTest.java | 63 +++++++++++++++++++
10 files changed, 348 insertions(+), 27 deletions(-)
diff --git a/docs/en/connectors/sink/Jdbc.md b/docs/en/connectors/sink/Jdbc.md
index 5284498d1e..08637e4466 100644
--- a/docs/en/connectors/sink/Jdbc.md
+++ b/docs/en/connectors/sink/Jdbc.md
@@ -267,10 +267,20 @@ Current support:
| Dialect | Supported | Allowed keys |
|---------|-----------|--------------|
| MySQL | Yes | `engine`, `charset`, `collate` |
+| TiDB | Yes | `engine`, `charset`, `collate` (via MySQL JDBC protocol and
`jdbc:mysql://`) |
+| OceanBase (MySQL mode) | Yes | `engine`, `charset`, `collate` |
| Other JDBC dialects | No | Non-empty `table_options` fails validation at job
submission |
Invalid or unsupported keys are validated early via `JdbcSinkFactory` option
rules (`--check` and job submission), not only at runtime DDL.
+**Dialect notes:**
+
+- **MySQL**: `engine`, `charset`, and `collate` are appended to `CREATE TABLE`
and take effect.
+- **TiDB**: When connected via `jdbc:mysql://` with a MySQL JDBC driver, TiDB
shares the same key whitelist and DDL merge path as MySQL. `charset` and
`collate` take effect; `engine` is accepted for MySQL syntax compatibility but
is **ignored** by TiDB (storage engine is not configurable).
+- **OceanBase (MySQL mode)**: Supported for `jdbc:oceanbase://` when not using
Oracle-compatible mode. `charset` and `collate` must be values supported by
your OceanBase version (typically a MySQL-compatible subset; use `SHOW CHARSET`
/ `SHOW COLLATION` on the target). Unsupported values fail when `CREATE TABLE`
runs, not at job submission. OceanBase **Oracle-compatible mode** does not
support `table_options`; a non-empty map fails at job submission.
+
+SeaTunnel validates the **key whitelist** at submission time only; it does not
verify whether each value is supported by the target database (same as the
initial MySQL delivery).
+
Example (MySQL auto-create with engine and charset):
```hocon
diff --git a/docs/zh/connectors/sink/Jdbc.md b/docs/zh/connectors/sink/Jdbc.md
index e13b6e181c..a33503c583 100644
--- a/docs/zh/connectors/sink/Jdbc.md
+++ b/docs/zh/connectors/sink/Jdbc.md
@@ -256,10 +256,20 @@ Sink 在自动建表（SaveMode DDL）时附加的表级选项。仅在 `schema_
| 方言 | 是否支持 | 可用 key |
|------|----------|----------|
| MySQL | 是 | `engine`、`charset`、`collate` |
+| TiDB | 是 | `engine`、`charset`、`collate`（通过 MySQL JDBC 协议与 `jdbc:mysql://`
连接） |
+| OceanBase（MySQL 模式） | 是 | `engine`、`charset`、`collate` |
| 其他 JDBC 方言 | 否 | 配置非空 `table_options` 时任务启动即校验失败 |
非法或不支持的 key 会在 `JdbcSinkFactory` 的 option 规则阶段提前校验（`--check` 与作业提交），而非仅在运行时
DDL 阶段失败。
+**方言说明：**
+
+- **MySQL**：`engine`、`charset`、`collate` 均会写入 `CREATE TABLE` 并生效。
+- **TiDB**：通过 `jdbc:mysql://` 与 MySQL JDBC 驱动连接时，与 MySQL 使用相同的 key 白名单与 DDL
拼接方式。`charset`、`collate` 会生效；`engine` 仅为 MySQL 兼容语法，TiDB 会解析但**忽略**存储引擎设置。
+- **OceanBase（MySQL 模式）**：`jdbc:oceanbase://` 且非 Oracle 兼容模式时支持上述三个
key。`charset`、`collate` 须为 OceanBase 当前版本支持的字符集与排序规则（通常为 MySQL 兼容子集，请以目标库 `SHOW
CHARSET` / `SHOW COLLATION` 为准）；不支持的取值会在执行 `CREATE TABLE`
时报错，而非在作业提交阶段校验。OceanBase **Oracle 兼容模式**不支持 `table_options`，配置非空时任务启动即失败。
+
+SeaTunnel 在提交时仅校验 **key 白名单**，不校验具体取值是否被目标库支持（与 MySQL 首批实现一致）。
+
示例（MySQL 自动建表时指定存储引擎与字符集）：
```hocon
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMysqlCreateTableSqlBuilder.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMysqlCreateTableSqlBuilder.java
index 5053d8cc14..db80ccb09c 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMysqlCreateTableSqlBuilder.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMysqlCreateTableSqlBuilder.java
@@ -27,6 +27,7 @@ import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.converter.BasicTypeDefine;
import org.apache.seatunnel.api.table.type.SqlType;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.mysql.MySqlCatalog;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.utils.CatalogUtils;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseIdentifier;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oceanbase.OceanBaseMySqlTypeConverter;
@@ -85,9 +86,9 @@ public class OceanBaseMysqlCreateTableSqlBuilder {
return new OceanBaseMysqlCreateTableSqlBuilder(
tablePath.getTableName(), typeConverter, createIndex)
.comment(catalogTable.getComment())
- // todo: set charset and collate
- .engine(null)
- .charset(null)
+
.engine(catalogTable.getOptions().get(MySqlCatalog.TABLE_OPTION_ENGINE))
+
.charset(catalogTable.getOptions().get(MySqlCatalog.TABLE_OPTION_CHARSET))
+
.collate(catalogTable.getOptions().get(MySqlCatalog.TABLE_OPTION_COLLATE))
.primaryKey(tableSchema.getPrimaryKey())
.constraintKeys(tableSchema.getConstraintKeys())
.addColumn(tableSchema.getColumns())
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialect.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialect.java
index 09fefdd564..685b19efad 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialect.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialect.java
@@ -19,11 +19,9 @@ package
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.mysql;
import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
-import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.converter.BasicTypeDefine;
import org.apache.seatunnel.api.table.converter.TypeConverter;
-import
org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorException;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.JdbcRowConverter;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseIdentifier;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect;
@@ -43,14 +41,11 @@ import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.HashMap;
-import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
-import java.util.Set;
import java.util.stream.Collectors;
@Slf4j
@@ -58,10 +53,6 @@ public class MysqlDialect implements JdbcDialect {
private static final List NOT_SUPPORTED_DEFAULT_VALUES =
Arrays.asList(MysqlType.BLOB, MysqlType.TEXT, MysqlType.JSON,
MysqlType.GEOMETRY);
- private static final Set<String> SUPPORTED_TABLE_OPTIONS =
- Collections.unmodifiableSet(
- new LinkedHashSet<>(Arrays.asList("engine", "charset",
"collate")));
-
public String fieldIde = FieldIdeEnum.ORIGINAL.getValue();
public MysqlDialect() {}
@@ -400,20 +391,6 @@ public class MysqlDialect implements JdbcDialect {
@Override
public void validateTableOptions(Map<String, String> tableOptions) {
- if (tableOptions == null || tableOptions.isEmpty()) {
- return;
- }
-
- Set<String> unsupportedOptions = new
LinkedHashSet<>(tableOptions.keySet());
- unsupportedOptions.removeAll(SUPPORTED_TABLE_OPTIONS);
- if (!unsupportedOptions.isEmpty()) {
- throw new JdbcConnectorException(
- SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
- String.format(
- "Unsupported JDBC table_options for dialect '%s':
%s. Supported keys: %s",
- dialectName(),
- String.join(", ", unsupportedOptions),
- String.join(", ", SUPPORTED_TABLE_OPTIONS)));
- }
+ MysqlFamilyTableOptions.validate(dialectName(), tableOptions);
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlFamilyTableOptions.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlFamilyTableOptions.java
new file mode 100644
index 0000000000..3a19b0328d
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlFamilyTableOptions.java
@@ -0,0 +1,55 @@
+/*
+ * 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.jdbc.internal.dialect.mysql;
+
+import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorException;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Set;
+
+/** Shared {@code table_options} validation for MySQL-compatible JDBC
dialects. */
+public final class MysqlFamilyTableOptions {
+
+ public static final Set<String> SUPPORTED_KEYS =
+ Collections.unmodifiableSet(
+ new LinkedHashSet<>(Arrays.asList("engine", "charset",
"collate")));
+
+ private MysqlFamilyTableOptions() {}
+
+ public static void validate(String dialectName, Map<String, String>
tableOptions) {
+ if (tableOptions == null || tableOptions.isEmpty()) {
+ return;
+ }
+
+ Set<String> unsupportedOptions = new
LinkedHashSet<>(tableOptions.keySet());
+ unsupportedOptions.removeAll(SUPPORTED_KEYS);
+ if (!unsupportedOptions.isEmpty()) {
+ throw new JdbcConnectorException(
+ SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+ String.format(
+ "Unsupported JDBC table_options for dialect '%s':
%s. Supported keys: %s",
+ dialectName,
+ String.join(", ", unsupportedOptions),
+ String.join(", ", SUPPORTED_KEYS)));
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oceanbase/OceanBaseMysqlDialect.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oceanbase/OceanBaseMysqlDialect.java
index 8f36f0a6eb..f505559bad 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oceanbase/OceanBaseMysqlDialect.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oceanbase/OceanBaseMysqlDialect.java
@@ -28,6 +28,7 @@ import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDiale
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectTypeMapper;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.SQLUtils;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.dialectenum.FieldIdeEnum;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.mysql.MysqlFamilyTableOptions;
import org.apache.seatunnel.connectors.seatunnel.jdbc.source.JdbcSourceTable;
import lombok.extern.slf4j.Slf4j;
@@ -269,4 +270,9 @@ public class OceanBaseMysqlDialect implements JdbcDialect {
return false;
}
}
+
+ @Override
+ public void validateTableOptions(Map<String, String> tableOptions) {
+ MysqlFamilyTableOptions.validate(dialectName(), tableOptions);
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMysqlCreateTableSqlBuilderTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMysqlCreateTableSqlBuilderTest.java
new file mode 100644
index 0000000000..95e85035e2
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMysqlCreateTableSqlBuilderTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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.jdbc.catalog.oceanbase;
+
+import org.apache.seatunnel.shade.com.google.common.collect.Lists;
+
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.PrimaryKey;
+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.jdbc.catalog.mysql.MySqlCatalog;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseIdentifier;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oceanbase.OceanBaseMySqlTypeConverter;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class OceanBaseMysqlCreateTableSqlBuilderTest {
+
+ @Test
+ public void testBuildCreateTableSqlWithTableOptions() {
+ TablePath tablePath = TablePath.of("test_db", "test_table");
+ TableSchema tableSchema =
+ TableSchema.builder()
+ .column(PhysicalColumn.of("id", BasicType.LONG_TYPE,
0, false, null, "id"))
+ .primaryKey(PrimaryKey.of("id",
Lists.newArrayList("id")))
+ .build();
+ Map<String, String> options = new HashMap<>();
+ options.put(MySqlCatalog.TABLE_OPTION_ENGINE, "InnoDB");
+ options.put(MySqlCatalog.TABLE_OPTION_CHARSET, "utf8mb4");
+ options.put(MySqlCatalog.TABLE_OPTION_COLLATE, "utf8mb4_unicode_ci");
+ CatalogTable catalogTable =
+ CatalogTable.of(
+ TableIdentifier.of("test_catalog", "test_db",
"test_table"),
+ tableSchema,
+ options,
+ Collections.emptyList(),
+ "table with options");
+
+ String createTableSql =
+ OceanBaseMysqlCreateTableSqlBuilder.builder(
+ tablePath, catalogTable,
OceanBaseMySqlTypeConverter.INSTANCE, true)
+ .build(DatabaseIdentifier.OCEANBASE);
+
+ Assertions.assertTrue(createTableSql.contains("ENGINE = InnoDB"));
+ Assertions.assertTrue(createTableSql.contains("DEFAULT CHARSET =
utf8mb4"));
+ Assertions.assertTrue(createTableSql.contains("COLLATE =
utf8mb4_unicode_ci"));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/tidb/TiDBTableOptionsTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/tidb/TiDBTableOptionsTest.java
new file mode 100644
index 0000000000..16350db3f0
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/tidb/TiDBTableOptionsTest.java
@@ -0,0 +1,73 @@
+/*
+ * 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.jdbc.catalog.tidb;
+
+import org.apache.seatunnel.shade.com.google.common.collect.Lists;
+
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.PrimaryKey;
+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.jdbc.catalog.mysql.MySqlCatalog;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.mysql.MysqlCreateTableSqlBuilder;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseIdentifier;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.mysql.MySqlTypeConverter;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/** TiDB catalog inherits MySQL auto-create DDL; table_options use the same
keys. */
+public class TiDBTableOptionsTest {
+
+ @Test
+ public void testBuildCreateTableSqlWithTableOptions() {
+ TablePath tablePath = TablePath.of("test_db", "test_table");
+ TableSchema tableSchema =
+ TableSchema.builder()
+ .column(PhysicalColumn.of("id", BasicType.LONG_TYPE,
0, false, null, "id"))
+ .primaryKey(PrimaryKey.of("id",
Lists.newArrayList("id")))
+ .build();
+ Map<String, String> options = new HashMap<>();
+ options.put(MySqlCatalog.TABLE_OPTION_ENGINE, "InnoDB");
+ options.put(MySqlCatalog.TABLE_OPTION_CHARSET, "utf8mb4");
+ options.put(MySqlCatalog.TABLE_OPTION_COLLATE, "utf8mb4_unicode_ci");
+ CatalogTable catalogTable =
+ CatalogTable.of(
+ TableIdentifier.of("test_catalog", "test_db",
"test_table"),
+ tableSchema,
+ options,
+ Collections.emptyList(),
+ "tidb table with options");
+
+ String createTableSql =
+ MysqlCreateTableSqlBuilder.builder(
+ tablePath, catalogTable,
MySqlTypeConverter.DEFAULT_INSTANCE, true)
+ .build(DatabaseIdentifier.TIDB);
+
+ Assertions.assertTrue(createTableSql.contains("ENGINE = InnoDB"));
+ Assertions.assertTrue(createTableSql.contains("DEFAULT CHARSET =
utf8mb4"));
+ Assertions.assertTrue(createTableSql.contains("COLLATE =
utf8mb4_unicode_ci"));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oceanbase/OceanBaseMysqlDialectTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oceanbase/OceanBaseMysqlDialectTest.java
new file mode 100644
index 0000000000..f4ba5c1ec3
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oceanbase/OceanBaseMysqlDialectTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.jdbc.internal.dialect.oceanbase;
+
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorException;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class OceanBaseMysqlDialectTest {
+
+ @Test
+ public void testValidateTableOptions() {
+ OceanBaseMysqlDialect dialect = new OceanBaseMysqlDialect();
+ Map<String, String> tableOptions = new HashMap<>();
+ tableOptions.put("engine", "InnoDB");
+ tableOptions.put("charset", "utf8mb4");
+ tableOptions.put("collate", "utf8mb4_unicode_ci");
+
+ Assertions.assertDoesNotThrow(() ->
dialect.validateTableOptions(tableOptions));
+ }
+
+ @Test
+ public void testValidateTableOptionsWithUnknownKey() {
+ OceanBaseMysqlDialect dialect = new OceanBaseMysqlDialect();
+
+ JdbcConnectorException exception =
+ Assertions.assertThrows(
+ JdbcConnectorException.class,
+ () ->
+ dialect.validateTableOptions(
+ Collections.singletonMap("bucket_num",
"3")));
+ Assertions.assertTrue(exception.getMessage().contains("Unsupported
JDBC table_options"));
+ Assertions.assertTrue(exception.getMessage().contains("OceanBase"));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcTableOptionsConditionExtensionTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcTableOptionsConditionExtensionTest.java
index dd8e99697c..65d4c31e00 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcTableOptionsConditionExtensionTest.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcTableOptionsConditionExtensionTest.java
@@ -74,6 +74,44 @@ class JdbcTableOptionsConditionExtensionTest {
Assertions.assertDoesNotThrow(() -> validateSinkOptionRule(config));
}
+ @Test
+ void testTidbTableOptionsPassViaOptionRule() {
+ Map<String, Object> config = tidbSinkConfig();
+ Map<String, String> tableOptions = new HashMap<>();
+ tableOptions.put("engine", "InnoDB");
+ tableOptions.put("charset", "utf8mb4");
+ tableOptions.put("collate", "utf8mb4_unicode_ci");
+ config.put(SinkConnectorCommonOptions.TABLE_OPTIONS.key(),
tableOptions);
+
+ Assertions.assertDoesNotThrow(() -> validateSinkOptionRule(config));
+ }
+
+ @Test
+ void testOceanBaseMysqlTableOptionsPassViaOptionRule() {
+ Map<String, Object> config = oceanBaseMysqlSinkConfig();
+ Map<String, String> tableOptions = new HashMap<>();
+ tableOptions.put("engine", "InnoDB");
+ tableOptions.put("charset", "utf8mb4");
+ tableOptions.put("collate", "utf8mb4_unicode_ci");
+ config.put(SinkConnectorCommonOptions.TABLE_OPTIONS.key(),
tableOptions);
+
+ Assertions.assertDoesNotThrow(() -> validateSinkOptionRule(config));
+ }
+
+ @Test
+ void testOceanBaseOracleRejectsNonEmptyTableOptionsViaOptionRule() {
+ Map<String, Object> config = oceanBaseOracleSinkConfig();
+ Map<String, String> tableOptions = new HashMap<>();
+ tableOptions.put("engine", "InnoDB");
+ config.put(SinkConnectorCommonOptions.TABLE_OPTIONS.key(),
tableOptions);
+
+ OptionValidationException exception =
+ Assertions.assertThrows(
+ OptionValidationException.class, () ->
validateSinkOptionRule(config));
+ Assertions.assertTrue(
+ exception.getMessage().contains("not supported for dialect
'Oracle'"));
+ }
+
private static void validateSinkOptionRule(Map<String, Object> config) {
ConfigValidator.of(ReadonlyConfig.fromMap(config))
.validate(new JdbcSinkFactory().optionRule());
@@ -94,4 +132,29 @@ class JdbcTableOptionsConditionExtensionTest {
config.put("query", "INSERT INTO test_table VALUES (?)");
return config;
}
+
+ private static Map<String, Object> tidbSinkConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("url", "jdbc:mysql://127.0.0.1:4000/test");
+ config.put("driver", "com.mysql.cj.jdbc.Driver");
+ config.put("query", "INSERT INTO test_table VALUES (?)");
+ return config;
+ }
+
+ private static Map<String, Object> oceanBaseMysqlSinkConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("url", "jdbc:oceanbase://127.0.0.1:2881/test");
+ config.put("driver", "com.oceanbase.jdbc.Driver");
+ config.put("query", "INSERT INTO test_table VALUES (?)");
+ return config;
+ }
+
+ private static Map<String, Object> oceanBaseOracleSinkConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("url", "jdbc:oceanbase://127.0.0.1:2881/test");
+ config.put("driver", "com.oceanbase.jdbc.Driver");
+ config.put("compatible_mode", "oracle");
+ config.put("query", "INSERT INTO test_table VALUES (?)");
+ return config;
+ }
}
