This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12015-e71e5178e0d7f20ec952ff5d09e963d21659026a in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 5ba57fcbba337f153d60fd57b50bd6cf3a56f80b Author: luozihen <[email protected]> AuthorDate: Wed Oct 21 13:44:45 2026 +0000 [Feature][Connector-V2] Support per-table primary key mapping via multi_table_config.primary_keys for JDBC sink (#12015) --- docs/en/connectors/sink/Jdbc.md | 59 ++++ .../configuration/sink-options-placeholders.md | 9 + docs/zh/connectors/sink/Jdbc.md | 59 ++++ .../configuration/sink-options-placeholders.md | 8 + .../common/multitable/MultiTableFailureHelper.java | 14 +- .../multitable/MultiTableFailureHelperTest.java | 66 ++++ .../api/configuration/ReadableConfigTest.java | 11 + .../seatunnel/jdbc/config/JdbcSinkOptions.java | 13 + .../jdbc/exception/JdbcConnectorErrorCode.java | 3 +- .../JdbcMultiTableConfigConditionExtension.java | 69 ++++ .../seatunnel/jdbc/sink/JdbcSinkFactory.java | 378 ++++++++++++++++++--- .../seatunnel/jdbc/sink/JdbcSinkFactoryTest.java | 343 +++++++++++++++++++ .../core/starter/utils/ConfigBuilder.java | 22 +- .../core/starter/utils/ConfigShadeUtils.java | 2 +- .../core/starter/utils/ConfigBuilderTest.java | 77 +++++ .../core/starter/utils/ConfigShadeTest.java | 59 ++++ .../seatunnel/jdbc/JdbcMysqlMultipleTablesIT.java | 114 +++++++ ...ql_source_and_sink_with_multi_table_config.conf | 79 +++++ 18 files changed, 1340 insertions(+), 45 deletions(-) diff --git a/docs/en/connectors/sink/Jdbc.md b/docs/en/connectors/sink/Jdbc.md index 4ede4637bf..c4bd2e5639 100644 --- a/docs/en/connectors/sink/Jdbc.md +++ b/docs/en/connectors/sink/Jdbc.md @@ -176,6 +176,7 @@ Exactly-once delivery uses XA transactions and therefore requires XA support fro | tablePrefix | String | No | - | | tableSuffix | String | No | - | | primary_keys | Array | No | - | +| multi_table_config | Object | No | - | | connection_check_timeout_sec | Int | No | 30 | | connect_timeout_ms | Int | No | 86400000 | | socket_timeout_ms | Int | No | 86400000 | @@ -295,6 +296,64 @@ Deprecated. Use `table` with table placeholders instead. For example, use `table The target key columns used to generate database-native UPSERT, UPDATE, and DELETE statements. If omitted, SeaTunnel attempts to inherit a primary key or the first unique key from upstream catalog metadata. If no key is available, generated SQL uses plain INSERT. +### multi_table_config [object] + +Per-table primary key mapping for multi-table generated-SQL jobs. It has higher precedence than the top-level `primary_keys` option: when an upstream table name matches one of the patterns declared under `primary_keys`, that mapping is used; otherwise SeaTunnel falls back to the existing `primary_keys` / catalog metadata logic. + +The value must follow this shape: + +```hocon +multi_table_config { + primary_keys { + "<table-name-regex>" = ["key1", "key2"] + } +} +``` + +- `primary_keys` is required once `multi_table_config` is set. If it is missing, empty, or not a map, the job fails during option validation instead of falling back to the top-level `primary_keys` / catalog logic. +- Each key is a Java regular expression matched against the upstream table name using full match semantics (`tableName.matches(pattern)`). For example, use `"^t_nova_.*$"` or `"t_nova_.*"`, not a glob such as `"t_nova_*"`. +- Each value is a list of column names. The `${primary_key}` and `${unique_key}` placeholders are supported inside this option only, and can be mixed with static columns. `${primary_key}` expands to the upstream primary key columns, `${unique_key}` expands to the first upstream unique key columns. +- If a table matches multiple patterns, the first pattern in declaration order wins. +- If a matched table uses `${primary_key}` (or `${unique_key}`) but has no upstream primary key (or unique key), the job fails with a clear error. +- The two placeholder passes run in a fixed order: the engine-level table placeholder replacement runs first and leaves this option untouched (it only rewrites top-level `String` values and single-element `String` lists), and the JDBC sink then expands `${primary_key}` / `${unique_key}` inside `multi_table_config.primary_keys` while the sink is created. See [sink option placeholders](../../introduction/configuration/sink-options-placeholders.md) for the general rules. +- Regex keys are supported when the job is submitted from a config file or with the HOCON body of the REST submit-job API. The JSON body of that API does not support them yet and rejects such a configuration while parsing it; until the follow-up fix is available, submit the job from a config file or with the HOCON body. + +Example: use the upstream primary key plus a shared `DATA_SOURCE` column as the composite key for tables whose names start with `t_nova_`. + +```hocon +sink { + jdbc { + url = "jdbc:mysql://localhost:3306/test" + driver = "com.mysql.cj.jdbc.Driver" + generate_sink_sql = true + multi_table_config { + primary_keys { + "^t_nova_.*$" = ["${primary_key}", "DATA_SOURCE"] + } + } + } +} +``` + +Example: mixing `multi_table_config` with the top-level `primary_keys`. Tables matched by the mapping use the mapping; unmatched tables fall back to `primary_keys = ["merchant_id"]`. + +```hocon +sink { + jdbc { + url = "jdbc:mysql://localhost:3306/test" + driver = "com.mysql.cj.jdbc.Driver" + generate_sink_sql = true + primary_keys = ["merchant_id"] + multi_table_config { + primary_keys { + "t_tyuen_txn_ext.*" = ["id_txn_ctrl", "DATA_SOURCE"] + "t_nova_merge_settle_serial" = ["${primary_key}", "DATA_SOURCE"] + } + } + } +} +``` + ### connection_check_timeout_sec [int] The time in seconds to wait for the database operation used to validate the connection to complete. diff --git a/docs/en/introduction/configuration/sink-options-placeholders.md b/docs/en/introduction/configuration/sink-options-placeholders.md index 69684d4f41..7cea42e369 100644 --- a/docs/en/introduction/configuration/sink-options-placeholders.md +++ b/docs/en/introduction/configuration/sink-options-placeholders.md @@ -136,6 +136,15 @@ primary_keys = ["${primary_key}", "tenant_id"] The placeholder replacement for list values is only applied when `${primary_key}` is the only element in the list. +The restriction above applies to the top-level `primary_keys` option. The JDBC sink additionally supports +mixing `${primary_key}` (and `${unique_key}`) with static column names inside +`multi_table_config.primary_keys`; see the JDBC sink documentation for details. + +The two passes run in a fixed order. The engine-level replacement described above runs first and only +rewrites top-level `String` values and single-element `String` lists, so the nested +`multi_table_config.primary_keys` map reaches the sink untouched. The JDBC sink then expands +`${primary_key}` / `${unique_key}` inside that map once per matched table, while the sink is created. + The behavior is the same for both single-table and multi-table jobs. We will complete the placeholder replacement before the connector is started, ensuring that the sink options is ready before use. diff --git a/docs/zh/connectors/sink/Jdbc.md b/docs/zh/connectors/sink/Jdbc.md index cac905ba4b..51b08b3eb7 100644 --- a/docs/zh/connectors/sink/Jdbc.md +++ b/docs/zh/connectors/sink/Jdbc.md @@ -176,6 +176,7 @@ Exactly-once 依赖 XA 事务，因此数据库和 JDBC 驱动都必须支持 XA | tablePrefix | String | 否 | - | | tableSuffix | String | 否 | - | | primary_keys | Array | 否 | - | +| multi_table_config | Object | 否 | - | | connection_check_timeout_sec | Int | 否 | 30 | | connect_timeout_ms | Int | 否 | 86400000 | | socket_timeout_ms | Int | 否 | 86400000 | @@ -296,6 +297,64 @@ Tip: 如果目标数据库有 SCHEMA 的概念，则表参数必须写成 `xxx.x 生成数据库原生 UPSERT、UPDATE 和 DELETE 语句时使用的目标键列。未配置时，SeaTunnel 会尝试从上游 catalog 元数据继承主键或第一组唯一键；仍无可用键时，自动生成的 SQL 退回普通 INSERT。 +### multi_table_config [object] + +用于多表自动生成 SQL 场景的按表主键映射，优先级高于顶层 `primary_keys` 选项：当上游表名匹配到 `primary_keys` 下声明的某个模式时，使用该映射；否则回退到现有 `primary_keys` / catalog 元数据逻辑。 + +配置结构如下： + +```hocon +multi_table_config { + primary_keys { + "<表名正则>" = ["key1", "key2"] + } +} +``` + +- 一旦配置了 `multi_table_config`，就必须配置 `primary_keys`。若缺失、为空或不是 map，任务会在选项校验阶段直接失败，而不会回退到顶层 `primary_keys` / catalog 逻辑。 +- 每个 key 是匹配上游表名的 Java 正则表达式，使用全匹配语义（`tableName.matches(pattern)`）。例如使用 `"^t_nova_.*$"` 或 `"t_nova_.*"`，不要使用 `"t_nova_*"` 这类 glob 写法。 +- 每个 value 是列名列表。仅在该选项内支持 `${primary_key}` 和 `${unique_key}` 占位符，且可与静态列混用。`${primary_key}` 展开为上游主键列，`${unique_key}` 展开为上游第一组唯一键列。 +- 若一张表同时匹配多个模式，按声明顺序取第一个匹配。 +- 若命中的表使用了 `${primary_key}`（或 `${unique_key}`）但上游没有主键（或唯一键），任务会以清晰错误信息失败。 +- 两个占位符替换阶段的顺序是固定的：引擎层的表占位符替换先执行，且不会修改该选项（它只处理顶层的 `String` 值和单元素 `String` 列表）；随后由 JDBC Sink 在创建每个表时展开 `multi_table_config.primary_keys` 中的 `${primary_key}` / `${unique_key}`。通用规则见 [Sink 占位符说明](../../introduction/configuration/sink-options-placeholders.md)。 +- 通过配置文件提交，或通过 REST submit-job 接口的 HOCON body 提交时，支持正则 key；该接口的 JSON body 暂不支持，会在解析阶段直接拒绝。在该问题修复前，请改用配置文件或 HOCON body 提交。 + +示例：为表名以 `t_nova_` 开头的表，使用上游主键加上共享的 `DATA_SOURCE` 列作为复合主键。 + +```hocon +sink { + jdbc { + url = "jdbc:mysql://localhost:3306/test" + driver = "com.mysql.cj.jdbc.Driver" + generate_sink_sql = true + multi_table_config { + primary_keys { + "^t_nova_.*$" = ["${primary_key}", "DATA_SOURCE"] + } + } + } +} +``` + +示例：`multi_table_config` 与顶层 `primary_keys` 混用。被映射命中的表使用映射值，未命中的表回退到 `primary_keys = ["merchant_id"]`。 + +```hocon +sink { + jdbc { + url = "jdbc:mysql://localhost:3306/test" + driver = "com.mysql.cj.jdbc.Driver" + generate_sink_sql = true + primary_keys = ["merchant_id"] + multi_table_config { + primary_keys { + "t_tyuen_txn_ext.*" = ["id_txn_ctrl", "DATA_SOURCE"] + "t_nova_merge_settle_serial" = ["${primary_key}", "DATA_SOURCE"] + } + } + } +} +``` + ### connection_check_timeout_sec [int] 用于验证数据库连接的有效性时等待数据库操作完成所需的时间，单位是秒 diff --git a/docs/zh/introduction/configuration/sink-options-placeholders.md b/docs/zh/introduction/configuration/sink-options-placeholders.md index 624f0928bc..41f5898d8c 100644 --- a/docs/zh/introduction/configuration/sink-options-placeholders.md +++ b/docs/zh/introduction/configuration/sink-options-placeholders.md @@ -137,6 +137,14 @@ primary_keys = ["${primary_key}", "tenant_id"] 只有当 `${primary_key}` 是列表中的唯一元素时，才会执行列表占位符替换。 +上述限制仅针对顶层 `primary_keys` 选项。JDBC Sink 额外支持在 +`multi_table_config.primary_keys` 中将 `${primary_key}`（以及 `${unique_key}`）与静态列名混用， +详见 JDBC Sink 文档。 + +两个阶段的执行顺序是固定的：上文的引擎层替换先执行，且只处理顶层的 `String` 值和单元素 `String` 列表， +因此嵌套的 `multi_table_config.primary_keys` 映射会原样传给 Sink；随后由 JDBC Sink 在创建每个表时 +展开其中的 `${primary_key}` / `${unique_key}`。 + 单表任务和多表任务中的行为保持一致。 占位符的替换将在连接器启动之前完成，确保 Sink 参数在使用前已准备就绪。 diff --git a/seatunnel-api/src/main/java/org/apache/seatunnel/api/common/multitable/MultiTableFailureHelper.java b/seatunnel-api/src/main/java/org/apache/seatunnel/api/common/multitable/MultiTableFailureHelper.java index ca87980ef0..ba1d25dd2d 100644 --- a/seatunnel-api/src/main/java/org/apache/seatunnel/api/common/multitable/MultiTableFailureHelper.java +++ b/seatunnel-api/src/main/java/org/apache/seatunnel/api/common/multitable/MultiTableFailureHelper.java @@ -52,6 +52,15 @@ public final class MultiTableFailureHelper { .continueOtherTables(); } + /** + * Merges two option sets with primary precedence. The merge is shallow/top-level: on a key + * collision, {@code primary}'s value replaces {@code fallback}'s value wholesale, and nested + * objects are not merged recursively. + * + * @param primary the options that take precedence + * @param fallback the options used to fill in keys absent from {@code primary} + * @return the merged options + */ public static ReadonlyConfig mergeOptions(ReadonlyConfig primary, ReadonlyConfig fallback) { if (primary == null) { return fallback; @@ -59,7 +68,10 @@ public final class MultiTableFailureHelper { if (fallback == null) { return primary; } - return ReadonlyConfig.fromConfig(primary.toConfig().withFallback(fallback.toConfig())); + Map<String, Object> merged = new HashMap<>(); + merged.putAll(fallback.getSourceMap()); + merged.putAll(primary.getSourceMap()); + return ReadonlyConfig.fromMap(merged); } public static ReadonlyConfig withMultiTableFailurePolicy( diff --git a/seatunnel-api/src/test/java/org/apache/seatunnel/api/common/multitable/MultiTableFailureHelperTest.java b/seatunnel-api/src/test/java/org/apache/seatunnel/api/common/multitable/MultiTableFailureHelperTest.java new file mode 100644 index 0000000000..228f5edde7 --- /dev/null +++ b/seatunnel-api/src/test/java/org/apache/seatunnel/api/common/multitable/MultiTableFailureHelperTest.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.common.multitable; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; + +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 MultiTableFailureHelperTest { + + @Test + void testMergeOptionsPreservesSpecialCharacterKeys() { + Map<String, Object> primaryKeys = new HashMap<>(); + primaryKeys.put("^t_nova_.*$", Arrays.asList("${primary_key}", "DATA_SOURCE")); + + Map<String, Object> multiTableConfig = new HashMap<>(); + multiTableConfig.put("primary_keys", primaryKeys); + + Map<String, Object> primaryMap = new HashMap<>(); + primaryMap.put("multi_table_config", multiTableConfig); + primaryMap.put("url", "jdbc:mysql://localhost:3306/primary"); + + Map<String, Object> fallbackMap = new HashMap<>(); + fallbackMap.put("url", "jdbc:mysql://localhost:3306/fallback"); + fallbackMap.put("username", "u"); + + ReadonlyConfig merged = + MultiTableFailureHelper.mergeOptions( + ReadonlyConfig.fromMap(primaryMap), ReadonlyConfig.fromMap(fallbackMap)); + + Map<String, Object> mergedMap = merged.getSourceMap(); + Assertions.assertEquals("jdbc:mysql://localhost:3306/primary", mergedMap.get("url")); + Assertions.assertEquals("u", mergedMap.get("username")); + + Map<String, Object> mergedMultiTableConfig = + (Map<String, Object>) mergedMap.get("multi_table_config"); + Assertions.assertNotNull(mergedMultiTableConfig); + Map<String, Object> mergedPrimaryKeys = + (Map<String, Object>) mergedMultiTableConfig.get("primary_keys"); + Assertions.assertNotNull(mergedPrimaryKeys); + Assertions.assertTrue(mergedPrimaryKeys.containsKey("^t_nova_.*$")); + Assertions.assertEquals( + Arrays.asList("${primary_key}", "DATA_SOURCE"), + mergedPrimaryKeys.get("^t_nova_.*$")); + } +} diff --git a/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/ReadableConfigTest.java b/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/ReadableConfigTest.java index 24bd843ee7..d10450feea 100644 --- a/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/ReadableConfigTest.java +++ b/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/ReadableConfigTest.java @@ -378,4 +378,15 @@ public class ReadableConfigTest { ReadonlyConfig readonlyConfig = ReadonlyConfig.fromMap(map); Assertions.assertNull(readonlyConfig.toMap().get("user")); } + + @Test + public void testToConfigPreservesDottedKeyExpansion() { + Map<String, Object> map = new HashMap<>(); + map.put("a.b", 1); + + ReadonlyConfig readonlyConfig = ReadonlyConfig.fromMap(map); + Config config = readonlyConfig.toConfig(); + + Assertions.assertEquals(1, config.getInt("a.b")); + } } diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSinkOptions.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSinkOptions.java index 22c9c4dfc4..d96fc9f98d 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSinkOptions.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSinkOptions.java @@ -24,6 +24,7 @@ import org.apache.seatunnel.api.sink.SchemaSaveMode; import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.dialectenum.FieldIdeEnum; import java.util.List; +import java.util.Map; public class JdbcSinkOptions extends JdbcCommonOptions { @@ -108,6 +109,18 @@ public class JdbcSinkOptions extends JdbcCommonOptions { public static final Option<List<String>> PRIMARY_KEYS = Options.key("primary_keys").listType().noDefaultValue().withDescription("primary keys"); + public static final Option<Map<String, Object>> MULTI_TABLE_CONFIG = + Options.key("multi_table_config") + .mapObjectType() + .noDefaultValue() + .withDescription( + "Per-table primary key mapping for multi-table generated-SQL sinks. " + + "When a table matches one of the patterns under primary_keys, " + + "that mapping takes precedence over the top-level `primary_keys` " + + "option. Value shape: { primary_keys = { \"<regex>\" = " + + "[\"col1\", \"${primary_key}\"] } }. `primary_keys` is required " + + "once this option is set."); + public static final Option<Boolean> IS_PRIMARY_KEY_UPDATED = Options.key("is_primary_key_updated") .booleanType() diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/exception/JdbcConnectorErrorCode.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/exception/JdbcConnectorErrorCode.java index 2c2cac27b4..7a7c63f944 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/exception/JdbcConnectorErrorCode.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/exception/JdbcConnectorErrorCode.java @@ -33,7 +33,8 @@ public enum JdbcConnectorErrorCode implements SeaTunnelErrorCode { NO_SUPPORT_OPERATION_FAILED("JDBC-09", "The jdbc driver not support operation."), DATA_TYPE_CAST_FAILED("JDBC-10", "Data type cast failed"), REFRESH_PHYSICAL_TABLESCHEMA_BY_SCHEMA_CHANGE_EVENT( - "JDBC-11", "Refresh the table with schema change failed"); + "JDBC-11", "Refresh the table with schema change failed"), + INVALID_MULTI_TABLE_CONFIG("JDBC-12", "Invalid multi_table_config primary key mapping"); private final String code; diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcMultiTableConfigConditionExtension.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcMultiTableConfigConditionExtension.java new file mode 100644 index 0000000000..f9569b0e14 --- /dev/null +++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcMultiTableConfigConditionExtension.java @@ -0,0 +1,69 @@ +/* + * 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.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 java.util.Map; + +/** + * Early validation for the JDBC sink {@code multi_table_config} option. + * + * <p>Configuring {@code multi_table_config} is an explicit customization request, so it is only + * meaningful together with a {@code primary_keys} mapping. A missing, empty or non-map {@code + * primary_keys} is rejected during option-rule validation instead of being silently ignored while + * the sink is created, which would fall back to the top-level {@code primary_keys} / catalog + * primary key and write with an unexpected key. + */ +public class JdbcMultiTableConfigConditionExtension + implements ConditionExtension<Map<String, Object>> { + + public static final JdbcMultiTableConfigConditionExtension INSTANCE = + new JdbcMultiTableConfigConditionExtension(); + + private static final String PRIMARY_KEYS = "primary_keys"; + + private JdbcMultiTableConfigConditionExtension() {} + + @Override + public String description() { + return "must contain a non-empty primary_keys map when it is configured, for example " + + "primary_keys { \"^t_.*$\" = [\"key1\"] }"; + } + + @Override + public boolean evaluate(ReadonlyConfig config, Map<String, Object> value) + throws OptionValidationException { + // An absent or empty option means the user did not ask for per-table keys; keep the + // existing fallback behavior in that case. + if (value == null || value.isEmpty()) { + return true; + } + Object primaryKeys = value.get(PRIMARY_KEYS); + if (!(primaryKeys instanceof Map) || ((Map<?, ?>) primaryKeys).isEmpty()) { + throw new OptionValidationException( + "multi_table_config is configured but multi_table_config.primary_keys is missing " + + "or empty. Configure at least one pattern-to-columns mapping under " + + "multi_table_config.primary_keys, or remove multi_table_config to keep " + + "the fallback behavior."); + } + return true; + } +} diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkFactory.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkFactory.java index b77843556e..c6ba1a65f7 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkFactory.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkFactory.java @@ -28,6 +28,7 @@ import org.apache.seatunnel.api.options.ConnectorCommonOptions; import org.apache.seatunnel.api.options.SinkConnectorCommonOptions; import org.apache.seatunnel.api.sink.DataSaveMode; import org.apache.seatunnel.api.sink.SchemaSaveMode; +import org.apache.seatunnel.api.sink.TablePlaceholder; import org.apache.seatunnel.api.table.catalog.Catalog; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.catalog.Column; @@ -55,16 +56,35 @@ import org.apache.commons.collections4.CollectionUtils; import com.google.auto.service.AutoService; import java.sql.Connection; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.regex.Pattern; +import java.util.regex.PatternSyntaxException; import java.util.stream.Collectors; @AutoService(Factory.class) public class JdbcSinkFactory implements TableSinkFactory, SupportSinkDryRunValidation { + private static final int MAX_CACHED_PATTERNS = 256; + + // Bounded LRU so dynamically-generated regex patterns cannot grow without limit. Patterns are + // immutable, so a benign duplicate compile under concurrent access has no correctness impact. + private static final Map<String, Pattern> COMPILED_PATTERN_CACHE = + Collections.synchronizedMap( + new LinkedHashMap<String, Pattern>(16, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry<String, Pattern> eldest) { + return size() > MAX_CACHED_PATTERNS; + } + }); + @Override public String factoryIdentifier() { return "Jdbc"; @@ -87,6 +107,9 @@ public class JdbcSinkFactory implements TableSinkFactory, SupportSinkDryRunValid Map<String, String> sinkTableOptions = config.get(SinkConnectorCommonOptions.TABLE_OPTIONS); CatalogTable catalogTable = context.getCatalogTable(); ReadonlyConfig catalogOptions = getCatalogOptions(context); + // Keep the upstream table reference for pattern matching: catalogTable is rebuilt below + // with the resolved sink table path, whose table name no longer represents the source. + CatalogTable upstreamTable = catalogTable; // source table info TableIdentifier tableId = catalogTable.getTableId(); // sink table info @@ -118,48 +141,12 @@ public class JdbcSinkFactory implements TableSinkFactory, SupportSinkDryRunValid map.put(JdbcSinkOptions.TABLE.key(), catalogTable.getTableId().getTableName()); } map.put(JdbcSinkOptions.DATABASE.key(), catalogTable.getTableId().getDatabaseName()); - PrimaryKey primaryKey = catalogTable.getTableSchema().getPrimaryKey(); - if (CollectionUtils.isEmpty(config.get(JdbcSinkOptions.PRIMARY_KEYS))) { - if (primaryKey != null && !CollectionUtils.isEmpty(primaryKey.getColumnNames())) { - map.put( - JdbcSinkOptions.PRIMARY_KEYS.key(), - String.join(",", primaryKey.getColumnNames())); - } else { - Optional<ConstraintKey> keyOptional = - catalogTable.getTableSchema().getConstraintKeys().stream() - .filter( - key -> - ConstraintKey.ConstraintType.UNIQUE_KEY.equals( - key.getConstraintType())) - .findFirst(); - keyOptional.ifPresent( - constraintKey -> - map.put( - JdbcSinkOptions.PRIMARY_KEYS.key(), - constraintKey.getColumnNames().stream() - .map( - ConstraintKey.ConstraintKeyColumn - ::getColumnName) - .collect(Collectors.joining(",")))); - } + Optional<List<String>> multiTablePrimaryKeys = + resolveMultiTablePrimaryKeys(config, upstreamTable); + if (multiTablePrimaryKeys.isPresent()) { + catalogTable = applyPrimaryKeys(map, catalogTable, multiTablePrimaryKeys.get()); } else { - PrimaryKey configPk = - PrimaryKey.of( - catalogTable.getTablePath().getTableName() + "_config_pk", - config.get(JdbcSinkOptions.PRIMARY_KEYS)); - TableSchema tableSchema = catalogTable.getTableSchema(); - catalogTable = - CatalogTable.of( - catalogTable.getTableId(), - TableSchema.builder() - .primaryKey(configPk) - .constraintKey(tableSchema.getConstraintKeys()) - .columns(tableSchema.getColumns()) - .build(), - catalogTable.getOptions(), - catalogTable.getPartitionKeys(), - catalogTable.getComment(), - catalogTable.getCatalogName()); + catalogTable = applyFallbackPrimaryKeys(config, map, catalogTable); } config = ReadonlyConfig.fromMap(new HashMap<>(map)); final ReadonlyConfig options = config; @@ -192,6 +179,310 @@ public class JdbcSinkFactory implements TableSinkFactory, SupportSinkDryRunValid finalCatalogTable); } + /** + * Writes the resolved primary key columns into the sink config map and rebuilds the catalog + * table so that auto-created tables and generated upsert/update/delete statements use the given + * key columns. + * + * @param map the sink config map that is later turned back into a {@link ReadonlyConfig} + * @param catalogTable the table being processed + * @param primaryKeys the resolved key columns + * @return a new catalog table whose primary key is replaced with the resolved columns + */ + private CatalogTable applyPrimaryKeys( + Map<String, String> map, CatalogTable catalogTable, List<String> primaryKeys) { + validatePrimaryKeyColumns(primaryKeys, catalogTable.getTablePath().getTableName()); + map.put(JdbcSinkOptions.PRIMARY_KEYS.key(), String.join(",", primaryKeys)); + return buildCatalogTableWithPrimaryKey(catalogTable, primaryKeys); + } + + /** + * Rebuilds the catalog table so that its primary key is the given columns. The option map is + * left untouched, so a caller that resolved the columns from the existing configuration keeps + * the value the user wrote. + * + * @param catalogTable the table being processed + * @param primaryKeys the resolved key columns + * @return a new catalog table whose primary key is replaced with the resolved columns + */ + private CatalogTable buildCatalogTableWithPrimaryKey( + CatalogTable catalogTable, List<String> primaryKeys) { + PrimaryKey configPk = + PrimaryKey.of( + catalogTable.getTablePath().getTableName() + "_config_pk", primaryKeys); + TableSchema tableSchema = catalogTable.getTableSchema(); + return CatalogTable.of( + catalogTable.getTableId(), + TableSchema.builder() + .primaryKey(configPk) + .constraintKey(tableSchema.getConstraintKeys()) + .columns(tableSchema.getColumns()) + .build(), + catalogTable.getOptions(), + catalogTable.getPartitionKeys(), + catalogTable.getComment(), + catalogTable.getCatalogName()); + } + + /** + * Resolves the primary key columns using the pre-existing fallback logic when no multi-table + * mapping matches: explicit top-level {@code primary_keys}, otherwise the catalog primary key, + * otherwise the first unique key. When no key can be determined, the config map is left + * unchanged and the sink falls back to plain INSERT. + * + * @param config the sink config + * @param map the sink config map that is later turned back into a {@link ReadonlyConfig} + * @param catalogTable the table being processed + * @return the (possibly rebuilt) catalog table + */ + private CatalogTable applyFallbackPrimaryKeys( + ReadonlyConfig config, Map<String, String> map, CatalogTable catalogTable) { + PrimaryKey primaryKey = catalogTable.getTableSchema().getPrimaryKey(); + if (CollectionUtils.isEmpty(config.get(JdbcSinkOptions.PRIMARY_KEYS))) { + if (primaryKey != null && !CollectionUtils.isEmpty(primaryKey.getColumnNames())) { + map.put( + JdbcSinkOptions.PRIMARY_KEYS.key(), + String.join(",", primaryKey.getColumnNames())); + } else { + Optional<ConstraintKey> keyOptional = + catalogTable.getTableSchema().getConstraintKeys().stream() + .filter( + key -> + ConstraintKey.ConstraintType.UNIQUE_KEY.equals( + key.getConstraintType())) + .findFirst(); + keyOptional.ifPresent( + constraintKey -> + map.put( + JdbcSinkOptions.PRIMARY_KEYS.key(), + constraintKey.getColumnNames().stream() + .map( + ConstraintKey.ConstraintKeyColumn + ::getColumnName) + .collect(Collectors.joining(",")))); + } + return catalogTable; + } + // The legacy explicit primary_keys branch keeps the pre-existing behaviour: only the + // catalog primary key is rebuilt, the option value is neither rewritten nor re-validated. + return buildCatalogTableWithPrimaryKey( + catalogTable, config.get(JdbcSinkOptions.PRIMARY_KEYS)); + } + + /** + * Validates that each resolved primary key column is a non-empty plain identifier that does not + * contain a comma, so it can be safely comma-joined into {@code PRIMARY_KEYS} and used in + * generated SQL. Generated statements quote identifiers through the dialect, so a blank name + * and a comma (the delimiter of the comma-joined {@code PRIMARY_KEYS} value) are the only + * inputs that cannot be represented safely. + * + * @param primaryKeys the resolved key columns + * @param tableName the table being processed, used in the error message + * @throws JdbcConnectorException when a column name is blank or contains a comma + */ + private void validatePrimaryKeyColumns(List<String> primaryKeys, String tableName) { + for (String key : primaryKeys) { + if (StringUtils.isBlank(key)) { + throw new JdbcConnectorException( + JdbcConnectorErrorCode.INVALID_MULTI_TABLE_CONFIG, + String.format( + "Resolved primary key column for table '%s' is empty.", tableName)); + } + if (key.contains(",")) { + throw new JdbcConnectorException( + JdbcConnectorErrorCode.INVALID_MULTI_TABLE_CONFIG, + String.format( + "Resolved primary key column '%s' for table '%s' must not contain a comma.", + key, tableName)); + } + } + } + + /** + * Resolves the per-table primary key mapping from {@code multi_table_config.primary_keys}. + * + * <p>Each key is a Java regular expression matched against the upstream table name using full + * match semantics; the first matching pattern in declaration order wins. Every pattern is + * compiled eagerly so an invalid expression fails fast with {@code JDBC-12} instead of + * surfacing lazily once a matching table is processed. An unmatched table returns {@link + * Optional#empty()}, leaving the fallback logic to run. + * + * @param config the sink config + * @param catalogTable the upstream table; its name is matched against the patterns and its + * schema is used to expand {@code ${primary_key}} / {@code ${unique_key}} + * @return the resolved key columns, or {@link Optional#empty()} when no pattern matches + */ + Optional<List<String>> resolveMultiTablePrimaryKeys( + ReadonlyConfig config, CatalogTable catalogTable) { + Map<String, Object> multiTableConfig = config.get(JdbcSinkOptions.MULTI_TABLE_CONFIG); + if (multiTableConfig == null || multiTableConfig.isEmpty()) { + return Optional.empty(); + } + Object primaryKeysObj = multiTableConfig.get("primary_keys"); + if (!(primaryKeysObj instanceof Map)) { + return Optional.empty(); + } + LinkedHashMap<Pattern, List<String>> primaryKeyMap = + toCompiledPatternMap((Map<?, ?>) primaryKeysObj); + String tableName = catalogTable.getTableId().getTableName(); + for (Map.Entry<Pattern, List<String>> entry : primaryKeyMap.entrySet()) { + if (entry.getKey().matcher(tableName).matches()) { + return Optional.of( + expandPrimaryKeyPlaceholder( + entry.getValue(), catalogTable, entry.getKey().pattern())); + } + } + return Optional.empty(); + } + + /** + * Converts the configured {@code primary_keys} map into an ordered map of compiled patterns to + * key-column lists. A {@link LinkedHashMap} is used so the first matching pattern in + * declaration order wins. + * + * @throws JdbcConnectorException when a pattern is not a valid regular expression + */ + private LinkedHashMap<Pattern, List<String>> toCompiledPatternMap(Map<?, ?> primaryKeyMap) { + LinkedHashMap<Pattern, List<String>> ordered = new LinkedHashMap<>(); + for (Map.Entry<?, ?> entry : primaryKeyMap.entrySet()) { + ordered.put( + compilePattern(String.valueOf(entry.getKey())), + toPrimaryKeyList(entry.getValue())); + } + return ordered; + } + + /** + * Compiles a user-supplied regular expression, memoizing the result so each distinct pattern is + * compiled only once and reused across tables. Raises {@code JDBC-12} up front when the pattern + * is invalid. + * + * @throws JdbcConnectorException when the pattern is not a valid regular expression + */ + Pattern compilePattern(String pattern) { + Pattern compiled = COMPILED_PATTERN_CACHE.get(pattern); + if (compiled != null) { + return compiled; + } + try { + compiled = Pattern.compile(pattern); + } catch (PatternSyntaxException e) { + throw new JdbcConnectorException( + JdbcConnectorErrorCode.INVALID_MULTI_TABLE_CONFIG, + String.format( + "Invalid regular expression '%s' in multi_table_config.primary_keys.", + pattern), + e); + } + COMPILED_PATTERN_CACHE.put(pattern, compiled); + return compiled; + } + + /** + * Converts a configured primary key value into a list of column names. A list value is used + * as-is; a string value is split by comma. + * + * @throws JdbcConnectorException when the value is neither a list nor a string + */ + private List<String> toPrimaryKeyList(Object value) { + if (value instanceof List) { + List<String> keys = new ArrayList<>(); + for (Object element : (List<?>) value) { + keys.add(String.valueOf(element)); + } + return keys; + } + if (value instanceof String) { + String stringValue = (String) value; + if (StringUtils.isBlank(stringValue)) { + return Collections.emptyList(); + } + return Arrays.stream(stringValue.split(",")) + .map(String::trim) + .collect(Collectors.toList()); + } + throw new JdbcConnectorException( + JdbcConnectorErrorCode.INVALID_MULTI_TABLE_CONFIG, + "multi_table_config.primary_keys values must be a string or a list of strings."); + } + + /** + * Expands {@code ${primary_key}} and {@code ${unique_key}} placeholders in a key-column list. + * Each placeholder must be a whole element and is replaced by the corresponding upstream key + * columns. + * + * @throws JdbcConnectorException when a placeholder is used but the upstream table has no + * matching key + */ + private List<String> expandPrimaryKeyPlaceholder( + List<String> keys, CatalogTable catalogTable, String pattern) { + String primaryKeyPlaceholder = + "${" + TablePlaceholder.REPLACE_PRIMARY_KEY.getPlaceholder() + "}"; + String uniqueKeyPlaceholder = + "${" + TablePlaceholder.REPLACE_UNIQUE_KEY.getPlaceholder() + "}"; + List<String> primaryKeyColumns = getPrimaryKeyColumns(catalogTable); + List<String> uniqueKeyColumns = getUniqueKeyColumns(catalogTable); + List<String> resolved = new ArrayList<>(); + for (String key : keys) { + if (primaryKeyPlaceholder.equals(key)) { + if (primaryKeyColumns.isEmpty()) { + throw new JdbcConnectorException( + JdbcConnectorErrorCode.INVALID_MULTI_TABLE_CONFIG, + String.format( + "Table '%s' matched pattern '%s' in multi_table_config.primary_keys " + + "which uses '${primary_key}', but the upstream table has no primary key.", + catalogTable.getTableId().getTableName(), pattern)); + } + resolved.addAll(primaryKeyColumns); + } else if (uniqueKeyPlaceholder.equals(key)) { + if (uniqueKeyColumns.isEmpty()) { + throw new JdbcConnectorException( + JdbcConnectorErrorCode.INVALID_MULTI_TABLE_CONFIG, + String.format( + "Table '%s' matched pattern '%s' in multi_table_config.primary_keys " + + "which uses '${unique_key}', but the upstream table has no unique key.", + catalogTable.getTableId().getTableName(), pattern)); + } + resolved.addAll(uniqueKeyColumns); + } else { + resolved.add(key); + } + } + return resolved; + } + + /** + * Returns the upstream primary key column names, or an empty list when there is no primary key. + */ + private List<String> getPrimaryKeyColumns(CatalogTable catalogTable) { + PrimaryKey primaryKey = catalogTable.getTableSchema().getPrimaryKey(); + if (primaryKey == null || CollectionUtils.isEmpty(primaryKey.getColumnNames())) { + return Collections.emptyList(); + } + return new ArrayList<>(primaryKey.getColumnNames()); + } + + /** + * Returns the first upstream unique key column names, or an empty list when there is no unique + * key. + */ + private List<String> getUniqueKeyColumns(CatalogTable catalogTable) { + Optional<ConstraintKey> keyOptional = + catalogTable.getTableSchema().getConstraintKeys().stream() + .filter( + key -> + ConstraintKey.ConstraintType.UNIQUE_KEY.equals( + key.getConstraintType())) + .findFirst(); + return keyOptional + .map( + constraintKey -> + constraintKey.getColumnNames().stream() + .map(ConstraintKey.ConstraintKeyColumn::getColumnName) + .collect(Collectors.toList())) + .orElseGet(Collections::emptyList); + } + @Override public OptionRule optionRule() { return OptionRule.builder() @@ -245,6 +536,11 @@ public class JdbcSinkFactory implements TableSinkFactory, SupportSinkDryRunValid JdbcSinkOptions.TABLE_SUFFIX, SinkConnectorCommonOptions.MULTI_TABLE_SINK_REPLICA, JdbcSinkOptions.DIALECT) + .optional( + JdbcSinkOptions.MULTI_TABLE_CONFIG, + Conditions.extension( + JdbcSinkOptions.MULTI_TABLE_CONFIG, + JdbcMultiTableConfigConditionExtension.INSTANCE)) .optional( SinkConnectorCommonOptions.TABLE_OPTIONS, Conditions.extension( diff --git a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkFactoryTest.java b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkFactoryTest.java index bdcd346fa4..3f2260322d 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkFactoryTest.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkFactoryTest.java @@ -17,10 +17,15 @@ package org.apache.seatunnel.connectors.seatunnel.jdbc.sink; +import org.apache.seatunnel.shade.com.typesafe.config.Config; +import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory; + import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.api.configuration.util.ConfigValidator; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.configuration.util.OptionValidationException; +import org.apache.seatunnel.api.sink.SeaTunnelSink; +import org.apache.seatunnel.api.sink.TablePlaceholderProcessor; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.catalog.PhysicalColumn; import org.apache.seatunnel.api.table.catalog.PrimaryKey; @@ -29,14 +34,21 @@ import org.apache.seatunnel.api.table.catalog.TableSchema; import org.apache.seatunnel.api.table.connector.TableSink; import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext; import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSinkOptions; +import org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorException; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; +import java.util.Optional; +import java.util.regex.Pattern; class JdbcSinkFactoryTest { @@ -77,6 +89,41 @@ class JdbcSinkFactoryTest { "catalog"); } + private CatalogTable createCatalogTable(String tableName, List<String> primaryKeyColumns) { + TableSchema.Builder schemaBuilder = + TableSchema.builder() + .column(PhysicalColumn.of("id", BasicType.LONG_TYPE, 22, false, null, "id")) + .column( + PhysicalColumn.of( + "name", BasicType.STRING_TYPE, 128, false, null, "name")) + .column( + PhysicalColumn.of( + "DATA_SOURCE", + BasicType.STRING_TYPE, + 32, + false, + null, + "DATA_SOURCE")); + if (primaryKeyColumns != null && !primaryKeyColumns.isEmpty()) { + schemaBuilder.primaryKey(PrimaryKey.of("pk_id", primaryKeyColumns)); + } + return CatalogTable.of( + TableIdentifier.of("catalog", "ORCL", null, tableName), + schemaBuilder.build(), + new HashMap<>(), + new ArrayList<>(), + null, + "catalog"); + } + + private ReadonlyConfig multiTableReadonlyConfig(Map<String, Object> primaryKeys) { + Map<String, Object> cfg = baseConfig(); + Map<String, Object> multiTableConfig = new LinkedHashMap<>(); + multiTableConfig.put("primary_keys", primaryKeys); + cfg.put("multi_table_config", multiTableConfig); + return ReadonlyConfig.fromMap(cfg); + } + /** * Simulates the FactoryUtil.createAndPrepareSink entry path: OptionRule validation followed by * factory.createSink(context). This covers the real submission-time path end-to-end. @@ -285,4 +332,300 @@ class JdbcSinkFactoryTest { + "CatalogTable has primary keys; the PK + APPEND_VALUES conflict " + "is guarded at runtime by JdbcOutputFormatBuilder.validateOracleInsertMode"); } + + @Test + void testMultiTableConfigWithoutPrimaryKeysFailsOptionValidation() { + Map<String, Object> cfg = baseConfig(); + Map<String, Object> multiTableConfig = new LinkedHashMap<>(); + multiTableConfig.put("primary_key", Collections.singletonMap("^TEST_.*$", "id")); + cfg.put("multi_table_config", multiTableConfig); + + OptionValidationException exception = + Assertions.assertThrows( + OptionValidationException.class, + () -> createSinkViaFactoryContext(cfg, true)); + Assertions.assertTrue( + exception.getMessage().contains("multi_table_config.primary_keys"), + "The error must name the missing option, but was: " + exception.getMessage()); + } + + @Test + void testMultiTableConfigWithEmptyPrimaryKeysFailsOptionValidation() { + Map<String, Object> cfg = baseConfig(); + Map<String, Object> multiTableConfig = new LinkedHashMap<>(); + multiTableConfig.put("primary_keys", Collections.emptyMap()); + cfg.put("multi_table_config", multiTableConfig); + + Assertions.assertThrows(OptionValidationException.class, () -> validate(cfg)); + } + + @Test + void testMultiTableConfigWithNonMapPrimaryKeysFailsOptionValidation() { + Map<String, Object> cfg = baseConfig(); + Map<String, Object> multiTableConfig = new LinkedHashMap<>(); + multiTableConfig.put("primary_keys", Collections.singletonList("id")); + cfg.put("multi_table_config", multiTableConfig); + + Assertions.assertThrows(OptionValidationException.class, () -> validate(cfg)); + } + + @Test + void testMultiTableConfigWithPrimaryKeysPassesOptionValidation() { + Map<String, Object> cfg = baseConfig(); + Map<String, Object> multiTableConfig = new LinkedHashMap<>(); + multiTableConfig.put( + "primary_keys", + Collections.singletonMap("^TEST_.*$", Collections.singletonList("id"))); + cfg.put("multi_table_config", multiTableConfig); + + Assertions.assertDoesNotThrow(() -> validate(cfg)); + } + + @Test + void testEmptyMultiTableConfigSkipsOptionValidation() { + Map<String, Object> cfg = baseConfig(); + cfg.put("multi_table_config", Collections.emptyMap()); + + Assertions.assertDoesNotThrow( + () -> validate(cfg), "An empty multi_table_config is not a customization request"); + } + + @Test + void testResolveMultiTablePrimaryKeysMapped() { + Map<String, Object> primaryKeys = new LinkedHashMap<>(); + primaryKeys.put("^TEST_.*$", Arrays.asList("${primary_key}", "DATA_SOURCE")); + ReadonlyConfig config = multiTableReadonlyConfig(primaryKeys); + CatalogTable table = createCatalogTable("TEST_TABLE", Collections.singletonList("id")); + + Optional<List<String>> resolved = factory.resolveMultiTablePrimaryKeys(config, table); + + Assertions.assertTrue(resolved.isPresent()); + Assertions.assertEquals(Arrays.asList("id", "DATA_SOURCE"), resolved.get()); + } + + @Test + void testResolveMultiTablePrimaryKeysUnmatched() { + Map<String, Object> primaryKeys = new LinkedHashMap<>(); + primaryKeys.put("^OTHER_.*$", Arrays.asList("id_txn_ctrl")); + ReadonlyConfig config = multiTableReadonlyConfig(primaryKeys); + CatalogTable table = createCatalogTable("TEST_TABLE", Collections.singletonList("id")); + + Assertions.assertFalse(factory.resolveMultiTablePrimaryKeys(config, table).isPresent()); + } + + @Test + void testResolveMultiTablePrimaryKeysMissingPrimaryKeyFails() { + Map<String, Object> primaryKeys = new LinkedHashMap<>(); + primaryKeys.put("^TEST_.*$", Arrays.asList("${primary_key}", "DATA_SOURCE")); + ReadonlyConfig config = multiTableReadonlyConfig(primaryKeys); + CatalogTable table = createCatalogTable("TEST_TABLE", Collections.emptyList()); + + Assertions.assertThrows( + JdbcConnectorException.class, + () -> factory.resolveMultiTablePrimaryKeys(config, table)); + } + + @Test + void testResolveMultiTablePrimaryKeysFirstMatchWins() { + Map<String, Object> primaryKeys = new LinkedHashMap<>(); + primaryKeys.put("^TEST_.*$", Arrays.asList("id")); + primaryKeys.put("^T.*$", Arrays.asList("other")); + ReadonlyConfig config = multiTableReadonlyConfig(primaryKeys); + CatalogTable table = createCatalogTable("TEST_TABLE", Collections.singletonList("id")); + + Assertions.assertEquals( + Collections.singletonList("id"), + factory.resolveMultiTablePrimaryKeys(config, table).get()); + } + + /** + * Parses the same option from a real HOCON string (like the engine does) and keeps the pattern + * set non-alphabetical on purpose: {@code ^table.*$} is declared last although it matches first + * alphabetically, so a map that does not preserve the declaration order makes it win for both + * tables and fails this test. + */ + @Test + void testResolveMultiTablePrimaryKeysFromHoconFirstMatchWins() { + // Shaped like the config file used by the E2E test. The SeaTunnel HOCON parser turns a top + // level `sink` block into a list whose entries carry `plugin_name`, hence the list access. + Config hoconConfig = + ConfigFactory.parseString( + "sink {\n" + + " Jdbc {\n" + + " url = \"jdbc:mysql://localhost:3306/seatunnel\"\n" + + " database = \"sink\"\n" + + " table = \"${table_name}\"\n" + + " generate_sink_sql = true\n" + + " primary_keys = [\"c_bigint\"]\n" + + " multi_table_config {\n" + + " primary_keys {\n" + + " \"^table2$\" = [\"c_mediumint\"]\n" + + " \"^table1$\" = [\"c_int\", \"c_integer\"]\n" + + " \"^table.*$\" = [\"c_smallint\"]\n" + + " }\n" + + " }\n" + + " }\n" + + "}"); + ReadonlyConfig sinkConfig = + ReadonlyConfig.fromConfig(hoconConfig.getConfigList("sink").get(0)); + + Assertions.assertEquals( + Arrays.asList("c_int", "c_integer"), + factory.resolveMultiTablePrimaryKeys( + sinkConfig, + createCatalogTable("table1", Collections.singletonList("id"))) + .get()); + Assertions.assertEquals( + Collections.singletonList("c_mediumint"), + factory.resolveMultiTablePrimaryKeys( + sinkConfig, + createCatalogTable("table2", Collections.singletonList("id"))) + .get()); + } + + @Test + void testResolveMultiTablePrimaryKeysStringValue() { + Map<String, Object> primaryKeys = new LinkedHashMap<>(); + primaryKeys.put("^TEST_.*$", "id, DATA_SOURCE"); + ReadonlyConfig config = multiTableReadonlyConfig(primaryKeys); + CatalogTable table = createCatalogTable("TEST_TABLE", Collections.singletonList("id")); + + Assertions.assertEquals( + Arrays.asList("id", "DATA_SOURCE"), + factory.resolveMultiTablePrimaryKeys(config, table).get()); + } + + @Test + void testResolveMultiTablePrimaryKeysInvalidRegexFails() { + Map<String, Object> primaryKeys = new LinkedHashMap<>(); + primaryKeys.put("^[unclosed", Arrays.asList("id")); + ReadonlyConfig config = multiTableReadonlyConfig(primaryKeys); + CatalogTable table = createCatalogTable("TEST_TABLE", Collections.singletonList("id")); + + Assertions.assertThrows( + JdbcConnectorException.class, + () -> factory.resolveMultiTablePrimaryKeys(config, table)); + } + + @Test + void testCompilePatternCachesCompiledPattern() { + Pattern first = factory.compilePattern("^TEST_.*$"); + Pattern second = factory.compilePattern("^TEST_.*$"); + + Assertions.assertSame(first, second); + } + + @Test + void testCompilePatternInvalidFails() { + Assertions.assertThrows( + JdbcConnectorException.class, () -> factory.compilePattern("^[unclosed")); + } + + @Test + void testFactoryContextWithMultiTableConfigInvalidColumnFails() { + Map<String, Object> cfg = baseConfig(); + Map<String, Object> primaryKeys = new LinkedHashMap<>(); + primaryKeys.put("^TEST_.*$", Arrays.asList("id", "")); + Map<String, Object> multiTableConfig = new LinkedHashMap<>(); + multiTableConfig.put("primary_keys", primaryKeys); + cfg.put("multi_table_config", multiTableConfig); + + Assertions.assertThrows( + JdbcConnectorException.class, () -> createSinkViaFactoryContext(cfg, true)); + } + + @Test + void testFactoryContextWithMultiTableConfigValid() { + Map<String, Object> cfg = baseConfig(); + Map<String, Object> primaryKeys = new LinkedHashMap<>(); + primaryKeys.put("^TEST_.*$", Arrays.asList("${primary_key}", "DATA_SOURCE")); + Map<String, Object> multiTableConfig = new LinkedHashMap<>(); + multiTableConfig.put("primary_keys", primaryKeys); + cfg.put("multi_table_config", multiTableConfig); + + Assertions.assertDoesNotThrow(() -> createSinkViaFactoryContext(cfg, true)); + } + + /** + * The patterns must match the upstream table name even when the resolved sink table name + * differs, e.g. via a static {@code table} remap. Otherwise the mapping silently falls back to + * catalog metadata and the wrong key columns end up in the generated SQL. + */ + @Test + void testFactoryContextWithMultiTableConfigMatchesUpstreamTableName() { + Map<String, Object> cfg = baseConfig(); + cfg.put("table", "target_TEST_TABLE"); + + Map<String, Object> primaryKeys = new LinkedHashMap<>(); + primaryKeys.put("^TEST_TABLE$", Collections.singletonList("id")); + Map<String, Object> multiTableConfig = new LinkedHashMap<>(); + multiTableConfig.put("primary_keys", primaryKeys); + cfg.put("multi_table_config", multiTableConfig); + + TableSink tableSink = createSinkViaFactoryContext(cfg, false); + SeaTunnelSink<?, ?, ?, ?> sink = tableSink.createSink(); + CatalogTable writeTable = sink.getWriteCatalogTable().get(); + + Assertions.assertNotNull(writeTable.getTableSchema().getPrimaryKey()); + Assertions.assertEquals( + Collections.singletonList("id"), + writeTable.getTableSchema().getPrimaryKey().getColumnNames()); + } + + /** + * Guards the legacy top-level {@code primary_keys} branch: it keeps rebuilding the catalog + * primary key without rewriting the option value and without applying the new column + * validation, which is what this branch did before {@code multi_table_config} was added. + */ + @Test + void testLegacyPrimaryKeysStillRebuildCatalogPrimaryKey() { + Map<String, Object> cfg = baseConfig(); + cfg.put("primary_keys", Collections.singletonList("name")); + + TableSink tableSink = createSinkViaFactoryContext(cfg, false); + SeaTunnelSink<?, ?, ?, ?> sink = tableSink.createSink(); + CatalogTable writeTable = sink.getWriteCatalogTable().get(); + + Assertions.assertNotNull(writeTable.getTableSchema().getPrimaryKey()); + Assertions.assertEquals( + Collections.singletonList("name"), + writeTable.getTableSchema().getPrimaryKey().getColumnNames()); + } + + /** + * Pins the order between the engine-level TablePlaceholder pass and the connector-level + * expansion. The engine pass rewrites only top-level {@code String} values and single-element + * {@code String} lists, so a top-level {@code primary_keys = ["${primary_key}"]} is expanded + * there, while the nested {@code multi_table_config.primary_keys} map reaches the sink + * untouched and is expanded afterwards by {@link JdbcSinkFactory#resolveMultiTablePrimaryKeys}. + */ + @Test + void testPlaceholderOrderBetweenEngineAndConnectorPasses() { + Map<String, Object> cfg = baseConfig(); + cfg.put("primary_keys", Collections.singletonList("${primary_key}")); + Map<String, Object> primaryKeys = new LinkedHashMap<>(); + primaryKeys.put("^TEST_.*$", Arrays.asList("${primary_key}", "DATA_SOURCE")); + Map<String, Object> multiTableConfig = new LinkedHashMap<>(); + multiTableConfig.put("primary_keys", primaryKeys); + cfg.put("multi_table_config", multiTableConfig); + + CatalogTable upstream = createCatalogTable("TEST_TABLE", Collections.singletonList("id")); + ReadonlyConfig afterEnginePass = + TablePlaceholderProcessor.replaceTablePlaceholder( + ReadonlyConfig.fromMap(cfg), upstream); + + // The engine pass expands the top-level single-element placeholder list ... + Assertions.assertEquals( + Collections.singletonList("id"), afterEnginePass.get(JdbcSinkOptions.PRIMARY_KEYS)); + // ... leaves the nested multi_table_config map untouched ... + Map<?, ?> nestedPrimaryKeys = + (Map<?, ?>) + afterEnginePass.get(JdbcSinkOptions.MULTI_TABLE_CONFIG).get("primary_keys"); + Assertions.assertEquals( + Arrays.asList("${primary_key}", "DATA_SOURCE"), nestedPrimaryKeys.get("^TEST_.*$")); + // ... and the connector expands it afterwards, per matched table. + Assertions.assertEquals( + Arrays.asList("id", "DATA_SOURCE"), + factory.resolveMultiTablePrimaryKeys(afterEnginePass, upstream).get()); + } } diff --git a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigBuilder.java b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigBuilder.java index 754695826d..97e12c35e1 100644 --- a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigBuilder.java +++ b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigBuilder.java @@ -258,7 +258,27 @@ public class ConfigBuilder { processVariablesMap(resolvedConfigMap, userConfigMap); - return ConfigFactory.parseMap(resolvedConfigMap); + return rebuildWithLiteralKeys(resolvedConfigMap) + .resolve(ConfigResolveOptions.defaults().setAllowUnresolved(true)); + } + + /** + * Rebuilds a config from a map whose keys must be taken literally, for example the regex keys + * of {@code multi_table_config.primary_keys}. {@link ConfigFactory#parseMap} treats every key + * as a path expression and rejects such keys with {@code ConfigException.BadPath}, so the map + * is rendered as JSON and parsed with the JSON syntax instead, which keeps the keys as they + * are. + * + * <p>This is shared with {@link ConfigShadeUtils#processConfig} so that both rebuilds of a + * loaded config keep literal keys and cannot drift apart. + * + * @param configMap the config map to rebuild + * @return the rebuilt config, with the keys of {@code configMap} kept literal + */ + static Config rebuildWithLiteralKeys(Map<String, Object> configMap) { + return ConfigFactory.parseString( + JsonUtils.toJsonString(configMap), + ConfigParseOptions.defaults().setSyntax(ConfigSyntax.JSON)); } public static Config substituteUserConfig( diff --git a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigShadeUtils.java b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigShadeUtils.java index 8ec4b875a3..957a4cc1d3 100644 --- a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigShadeUtils.java +++ b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigShadeUtils.java @@ -225,7 +225,7 @@ public final class ConfigShadeUtils { configMap.put(Constants.SOURCE, sources); configMap.put(Constants.SINK, sinks); configMap.put(Constants.TRANSFORM, transforms); - return ConfigFactory.parseMap(configMap); + return ConfigBuilder.rebuildWithLiteralKeys(configMap); } catch (Exception e) { // Log desensitized error information log.error( diff --git a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigBuilderTest.java b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigBuilderTest.java index 9a86c43b98..56f2c0c7b2 100644 --- a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigBuilderTest.java +++ b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigBuilderTest.java @@ -17,11 +17,18 @@ package org.apache.seatunnel.core.starter.utils; +import org.apache.seatunnel.shade.com.typesafe.config.Config; +import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory; + import org.apache.seatunnel.core.starter.exception.ConfigCheckException; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.ArrayList; import java.util.Arrays; import java.util.LinkedHashMap; @@ -30,6 +37,76 @@ import java.util.Map; public class ConfigBuilderTest { + /** + * Every starter loads a job config through {@code ConfigBuilder.of(path, variables)}, which + * rebuilds the config after substituting user variables. That rebuild must keep option keys + * literal: a regex key such as {@code ^t_nova_.*$}, used by {@code + * multi_table_config.primary_keys}, is not a valid HOCON path, so rebuilding the map with + * {@code ConfigFactory.parseMap} rejected the whole config with {@code ConfigException.BadPath} + * (see #12491). + */ + @Test + public void testOfPreservesLiteralRegexOptionKeys(@TempDir Path tempDir) throws IOException { + Path conf = writeMultiTableConfigConf(tempDir); + + // The starters pass null when no -i variable is given. + assertLoadedConfig(conf, ConfigBuilder.of(conf, null), "jdbc:mysql://${db_host}/sink"); + + // A -i variable is substituted, and must not disturb the shape of the config. + assertLoadedConfig( + conf, + ConfigBuilder.of(conf, Arrays.asList("db_host=localhost:3306")), + "jdbc:mysql://localhost:3306/sink"); + } + + private static void assertLoadedConfig(Path conf, Config config, String expectedUrl) { + Assertions.assertEquals(expectedUrl, config.getConfigList("sink").get(0).getString("url")); + + // Ordinary dotted keys keep the shape the file parser produced, so nothing changes for the + // options that already worked. + Assertions.assertEquals( + ConfigFactory.parseFile(conf.toFile()).getConfig("env").root().unwrapped(), + config.getConfig("env").root().unwrapped()); + + // The regex key survives as one literal key, instead of being rejected as an invalid path + // expression. + Map<?, ?> primaryKeys = + config.getConfigList("sink") + .get(0) + .getConfig("multi_table_config") + .getConfig("primary_keys") + .root() + .unwrapped(); + Assertions.assertEquals( + Arrays.asList("${primary_key}", "DATA_SOURCE"), primaryKeys.get("^t_nova_.*$")); + } + + private static Path writeMultiTableConfigConf(Path tempDir) throws IOException { + Path conf = tempDir.resolve("multi_table_config.conf"); + Files.write( + conf, + Arrays.asList( + "env {", + " job.mode = \"BATCH\"", + "}", + "source {", + " FakeSource {", + " plugin_output = \"fake\"", + " }", + "}", + "sink {", + " Jdbc {", + " url = \"jdbc:mysql://${db_host}/sink\"", + " multi_table_config {", + " primary_keys {", + " \"^t_nova_.*$\" = [\"${primary_key}\", \"DATA_SOURCE\"]", + " }", + " }", + " }", + "}")); + return conf; + } + @Test public void testConfigDesensitizationSort() { Map<String, Object> config = new LinkedHashMap<>(); diff --git a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigShadeTest.java b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigShadeTest.java index a463ad50cc..478d0f5a23 100644 --- a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigShadeTest.java +++ b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigShadeTest.java @@ -573,6 +573,65 @@ public class ConfigShadeTest { token, encryptedConfig.getConfigList("sink").get(0).getString("token")); } + @Test + public void testDecryptPreservesSpecialCharacterKeys() { + Config input = + ConfigFactory.parseString( + "source { FakeSource { plugin_output = \"fake\" } }\n" + + "sink { Jdbc { url = \"jdbc:mysql://localhost:3306/db\", " + + "username = \"u\", password = \"p\", " + + "multi_table_config { primary_keys { " + + "\"^t_nova_.*$\" = [\"${primary_key}\", \"DATA_SOURCE\"], " + + "\"^t_tyuen_txn_.*$\" = [\"id_txn_ctrl\", \"DATA_SOURCE\"] " + + "} } } }"); + + Config decrypted = ConfigShadeUtils.decryptConfig(input); + + Map<String, Object> primaryKeys = + decrypted + .getConfigList("sink") + .get(0) + .getConfig("multi_table_config") + .getConfig("primary_keys") + .root() + .unwrapped(); + + Assertions.assertTrue(primaryKeys.containsKey("^t_nova_.*$")); + Assertions.assertTrue(primaryKeys.containsKey("^t_tyuen_txn_.*$")); + Assertions.assertEquals( + Arrays.asList("${primary_key}", "DATA_SOURCE"), primaryKeys.get("^t_nova_.*$")); + Assertions.assertEquals( + Arrays.asList("id_txn_ctrl", "DATA_SOURCE"), primaryKeys.get("^t_tyuen_txn_.*$")); + } + + /** + * Guards the other half of the JSON round trip in {@code processConfig}: ordinary config shapes + * must survive the rebuild unchanged. The key below is unquoted (so HOCON expands it to a + * nested object), {@code dfs.replication} is a quoted literal key inside a nested map, and both + * values are numbers. + */ + @Test + public void testDecryptPreservesOrdinaryConfigShapes() { + Config input = + ConfigFactory.parseString( + "env { job.mode = \"BATCH\" }\n" + + "source { FakeSource { plugin_output = \"fake\", " + + "split.size = 8096, \"dfs.replication\" = 3 } }\n" + + "sink { Jdbc { url = \"jdbc:mysql://localhost:3306/db\" } }"); + + Config decrypted = ConfigShadeUtils.decryptConfig(input); + + Assertions.assertEquals( + input.getConfig("env").root().unwrapped(), + decrypted.getConfig("env").root().unwrapped()); + Assertions.assertEquals( + input.getConfigList("source").get(0).root().unwrapped(), + decrypted.getConfigList("source").get(0).root().unwrapped()); + Assertions.assertEquals( + input.getConfigList("sink").get(0).root().unwrapped(), + decrypted.getConfigList("sink").get(0).root().unwrapped()); + } + public static class ConfigShadeWithProps implements ConfigShade { private String suffix; diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlMultipleTablesIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlMultipleTablesIT.java index 88256d8442..89702eeb63 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlMultipleTablesIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlMultipleTablesIT.java @@ -80,6 +80,25 @@ public class JdbcMysqlMultipleTablesIT extends TestSuiteBase implements TestReso private static final List<String> SINK_TABLES = TABLES.stream().map(table -> SINK_DATABASE + "." + table).collect(Collectors.toList()); + // Dedicated fixture for the multi_table_config phase. Its key columns are NOT NULL because + // MySQL rejects a column that is explicitly declared NULL and is part of a PRIMARY KEY; the + // shared source.table1/source.table2 fixture used by the other tests is left untouched. + private static final List<String> MULTI_TABLE_CONFIG_TABLES = + Arrays.asList("mtc_table1", "mtc_table2"); + private static final List<String> MULTI_TABLE_CONFIG_SOURCE_TABLES = + MULTI_TABLE_CONFIG_TABLES.stream() + .map(table -> SOURCE_DATABASE + "." + table) + .collect(Collectors.toList()); + private static final int MULTI_TABLE_CONFIG_ROW_COUNT = 3; + private static final String CREATE_MULTI_TABLE_CONFIG_TABLE_SQL = + "CREATE TABLE IF NOT EXISTS %s\n" + + "(\n" + + " `id` bigint(20) NOT NULL,\n" + + " `c_int` int(11) NOT NULL,\n" + + " `c_integer` int(11) NOT NULL,\n" + + " `c_mediumint` mediumint(9) NOT NULL,\n" + + " `c_varchar` varchar(64) DEFAULT NULL\n" + + ");"; private static final String CREATE_TABLE_SQL = "CREATE TABLE IF NOT EXISTS %s\n" + "(\n" @@ -146,6 +165,8 @@ public class JdbcMysqlMultipleTablesIT extends TestSuiteBase implements TestReso createTables(SOURCE_DATABASE, TABLES); createTables(SINK_DATABASE, TABLES); initSourceTablesData(); + createMultiTableConfigSourceTables(); + initMultiTableConfigSourceTablesData(); } @TestTemplate @@ -197,6 +218,43 @@ public class JdbcMysqlMultipleTablesIT extends TestSuiteBase implements TestReso container.executeJob("/jdbc_mysql_source_and_sink_with_multiple_tables.sql"); Assertions.assertEquals( 0, sqlConfEexecResult.getExitCode(), sqlConfEexecResult.getStderr()); + + // The dedicated sink tables are dropped so that the job creates them itself: only then the + // primary keys resolved from multi_table_config.primary_keys end up in the generated + // create-table DDL and can be asserted below. + dropMultiTableConfigSinkTables(); + try { + Container.ExecResult multiTableConfigExecResult = + container.executeJob( + "/jdbc_mysql_source_and_sink_with_multi_table_config.conf"); + Assertions.assertEquals( + 0, + multiTableConfigExecResult.getExitCode(), + multiTableConfigExecResult.getStderr()); + + // The catch-all pattern is declared last on purpose, so first-match-wins gives + // mtc_table1 a composite key and mtc_table2 a single key instead of the catch-all or + // the fallback key. + Assertions.assertAll( + () -> + Assertions.assertIterableEquals( + Arrays.asList("c_int", "c_integer"), + queryPrimaryKeyColumns(SINK_DATABASE, "mtc_table1")), + () -> + Assertions.assertIterableEquals( + Arrays.asList("c_mediumint"), + queryPrimaryKeyColumns(SINK_DATABASE, "mtc_table2")), + () -> + Assertions.assertEquals( + MULTI_TABLE_CONFIG_ROW_COUNT, + queryRowCount(SINK_DATABASE, "mtc_table1")), + () -> + Assertions.assertEquals( + MULTI_TABLE_CONFIG_ROW_COUNT, + queryRowCount(SINK_DATABASE, "mtc_table2"))); + } finally { + dropMultiTableConfigSinkTables(); + } } @TestTemplate @@ -328,6 +386,44 @@ public class JdbcMysqlMultipleTablesIT extends TestSuiteBase implements TestReso } } + private void createMultiTableConfigSourceTables() throws SQLException { + try (Statement statement = connection.createStatement()) { + statement.execute("create database if not exists " + SOURCE_DATABASE); + for (String table : MULTI_TABLE_CONFIG_TABLES) { + statement.execute( + String.format( + CREATE_MULTI_TABLE_CONFIG_TABLE_SQL, + SOURCE_DATABASE + "." + table)); + } + } + } + + private void initMultiTableConfigSourceTablesData() throws SQLException { + String sql = + "INSERT INTO %s (`id`, `c_int`, `c_integer`, `c_mediumint`, `c_varchar`)" + + " VALUES (?, ?, ?, ?, ?)"; + for (String table : MULTI_TABLE_CONFIG_SOURCE_TABLES) { + try (PreparedStatement statement = + connection.prepareStatement(String.format(sql, table))) { + for (int i = 0; i < MULTI_TABLE_CONFIG_ROW_COUNT; i++) { + statement.setLong(1, i + 1); + statement.setInt(2, 100 + i); + statement.setInt(3, 200 + i); + statement.setInt(4, 300 + i); + statement.setString(5, "row_" + i); + statement.addBatch(); + } + statement.executeBatch(); + } + } + } + + private void dropMultiTableConfigSinkTables() throws SQLException { + for (String table : MULTI_TABLE_CONFIG_TABLES) { + dropTable(SINK_DATABASE, table); + } + } + private List<List<Object>> query(String sql) { try (Statement statement = connection.createStatement(); ResultSet resultSet = statement.executeQuery(sql)) { @@ -347,6 +443,24 @@ public class JdbcMysqlMultipleTablesIT extends TestSuiteBase implements TestReso } } + private List<String> queryPrimaryKeyColumns(String database, String table) { + return query( + String.format( + "SELECT COLUMN_NAME FROM information_schema.STATISTICS" + + " WHERE TABLE_SCHEMA = '%s' AND TABLE_NAME = '%s'" + + " AND INDEX_NAME = 'PRIMARY' ORDER BY SEQ_IN_INDEX", + database, table)) + .stream() + .map(row -> String.valueOf(row.get(0))) + .collect(Collectors.toList()); + } + + private int queryRowCount(String database, String table) { + List<List<Object>> result = + query(String.format("SELECT COUNT(*) FROM %s.%s", database, table)); + return Integer.parseInt(String.valueOf(result.get(0).get(0))); + } + private void clearSinkTables() throws SQLException { for (String table : SINK_TABLES) { String sql = "truncate table " + table; diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_mysql_source_and_sink_with_multi_table_config.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_mysql_source_and_sink_with_multi_table_config.conf new file mode 100644 index 0000000000..531a800cfc --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_mysql_source_and_sink_with_multi_table_config.conf @@ -0,0 +1,79 @@ +# +# 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 { + jdbc { + url = "jdbc:mysql://mysql-e2e:3306/seatunnel" + driver = "com.mysql.cj.jdbc.Driver" + connection_check_timeout_sec = 100 + username = "root" + password = "Abc!@#135_seatunnel" + + table_list = [ + { + table_path = "source.mtc_table1" + }, + { + table_path = "source.mtc_table2" + } + ] + where_condition = "where c_int >= 0" + split.size = 8096 + split.even-distribution.factor.upper-bound = 100 + split.even-distribution.factor.lower-bound = 0.05 + split.sample-sharding.threshold = 1000 + split.inverse-sampling.rate = 1000 + } +} + +sink { + jdbc { + url = "jdbc:mysql://mysql-e2e:3306/seatunnel" + driver = "com.mysql.cj.jdbc.Driver" + username = "root" + password = "Abc!@#135_seatunnel" + + database = "sink" + table = "${table_name}" + generate_sink_sql = true + # The E2E test drops the sink tables before this job starts, so the sink creates them here and + # the primary keys resolved from multi_table_config.primary_keys end up in the generated DDL. + schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST" + create_index = true + + # Fallback for tables that match no pattern in multi_table_config.primary_keys. It must not be + # used by mtc_table1/mtc_table2, otherwise their keys would resolve to `id` instead of the + # mapped columns. + primary_keys = ["id"] + + multi_table_config { + primary_keys { + # The first matching pattern in declaration order wins. The catch-all is declared last on + # purpose: it also matches mtc_table1/mtc_table2 and it sorts first alphabetically, so it + # would win if the declaration order were not preserved. + "^mtc_table2$" = ["c_mediumint"] + "^mtc_table1$" = ["c_int", "c_integer"] + "^mtc_.*$" = ["c_int"] + } + } + } +}
