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 018e360cbc [Feature][Transform-V2] Add multi-table rule match mode
(#11099)
018e360cbc is described below
commit 018e360cbc0992671629ce10b4dd604693433b5f
Author: Goutam Adwant <[email protected]>
AuthorDate: Sun Jul 5 01:10:53 2026 -0700
[Feature][Transform-V2] Add multi-table rule match mode (#11099)
Signed-off-by: goutamadwant <[email protected]>
Co-authored-by: DanielLeens <[email protected]>
---
docs/en/transforms/transform-multi-table.md | 35 +-
docs/zh/transforms/transform-multi-table.md | 33 +-
.../adaptsink/DefineSinkTypeTransformFactory.java | 1 +
.../AbstractMultiCatalogFlatMapTransform.java | 10 +
.../common/AbstractMultiCatalogMapTransform.java | 10 +
.../common/AbstractMultiCatalogTransform.java | 102 ++++-
.../transform/common/ChainedFlatMapTransform.java | 116 +++++
.../transform/common/ChainedMapTransform.java | 110 +++++
.../transform/common/TransformCommonOptions.java | 11 +
.../transform/copy/CopyFieldTransformFactory.java | 1 +
.../DynamicCompileTransformFactory.java | 1 +
.../encrypt/FieldEncryptTransformFactory.java | 1 +
.../fieldmapper/FieldMapperTransformFactory.java | 1 +
.../filter/FilterFieldTransformFactory.java | 1 +
.../FilterRowKindTransformFactory.java | 1 +
.../jsonpath/JsonPathTransformFactory.java | 1 +
.../metadata/MetadataTransformFactory.java | 1 +
.../embedding/EmbeddingTransformFactory.java | 1 +
.../nlpmodel/llm/LLMTransformFactory.java | 1 +
.../regexextract/RegexExtractTransformFactory.java | 3 +-
.../rename/FieldRenameTransformFactory.java | 1 +
.../rename/TableRenameTransformFactory.java | 1 +
.../transform/replace/ReplaceTransformFactory.java | 1 +
.../rowkind/RowKindExtractorTransformFactory.java | 1 +
.../transform/split/SplitTransformFactory.java | 1 +
.../transform/sql/SQLTransformFactory.java | 1 +
.../table/TableFilterTransformFactory.java | 1 +
.../table/TableMergeTransformFactory.java | 1 +
.../validator/DataValidatorTransformFactory.java | 1 +
.../common/AbstractMultiCatalogTransformTest.java | 476 +++++++++++++++++++++
.../copy/CopyFieldMultiCatalogTransformTest.java | 94 ++++
31 files changed, 998 insertions(+), 22 deletions(-)
diff --git a/docs/en/transforms/transform-multi-table.md
b/docs/en/transforms/transform-multi-table.md
index 2e14f801f0..3097000bb8 100644
--- a/docs/en/transforms/transform-multi-table.md
+++ b/docs/en/transforms/transform-multi-table.md
@@ -18,7 +18,8 @@ Multi-table Transform has no limitations on Transform
capabilities; any Transfor
|----------------------------|--------|----------|---------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| table_match_regex | String | No | .* | A regular
expression to match the tables that require transformation. By default, it
matches all tables. Note that this table name refers to the actual upstream
table name, not `plugin_output`.
|
| table_transform | List | No | - | You can use a
list in `table_transform` to specify rules for individual tables. If a
transformation rule is configured for a specific table in `table_transform`,
the outer rules will not apply to that table. The rules in `table_transform`
take precedence. |
-| table_transform.table_path | String | No | - | When configuring
a transformation rule for a table in `table_transform`, you need to specify the
table path using the `table_path` field. The table path should include
`databaseName[.schemaName].tableName`.
|
+| table_transform.table_path | String | No | - | When configuring
a transformation rule for a table in `table_transform`, you need to specify the
table path using the `table_path` field. The table path should include
`databaseName[.schemaName].tableName` and is matched exactly.
|
+| rule_match_mode | String | No | - | Controls how
multiple `table_transform` entries with the same exact `table_path` are
evaluated. Supported values are `FIRST_MATCH` and `ALL_MATCH`.
|
## Matching Logic
@@ -66,6 +67,36 @@ This allows us to handle transformations for multiple tables
within a single tra
For each table, the priority of configuration is: `table_transform` >
`table_match_regex`. If no rules match a table, no transformation will be
applied.
+`table_transform.table_path` uses exact table path matching. `rule_match_mode`
only controls the case where multiple `table_transform` entries use the same
exact `table_path`:
+
+- If `rule_match_mode` is not configured, duplicate exact `table_path` entries
are rejected during config resolution.
+- `FIRST_MATCH`: apply the first matching `table_transform` entry in
declaration order.
+- `ALL_MATCH`: apply all matching `table_transform` entries in declaration
order. The output of an earlier matching rule becomes the input of the next
matching rule for the same table.
+
+For example, when `rule_match_mode` is `ALL_MATCH`, the following
configuration first copies `name` to `name2` for `test.xyz`, and then copies
`name2` to `name3` for the same table:
+
+```hocon
+transform {
+ Copy {
+ rule_match_mode = "ALL_MATCH"
+
+ table_transform = [{
+ table_path = "test.xyz"
+ src_field = "name"
+ dest_field = "name2"
+ }, {
+ table_path = "test.xyz"
+ src_field = "name2"
+ dest_field = "name3"
+ }]
+ }
+}
+```
+
+Output structure:
+
+| id | name | age | name2 | name3 |
+
Below are the transform configurations for each table:
- **test.abc** and **test.abcd**
@@ -125,4 +156,4 @@ Output structure:
| id | name | age |
-In this example, we used the Copy transform, but all transforms in SeaTunnel
support multi-table transformations, and you can configure them similarly
within the corresponding transform block.
\ No newline at end of file
+In this example, we used the Copy transform, but all transforms in SeaTunnel
support multi-table transformations, and you can configure them similarly
within the corresponding transform block.
diff --git a/docs/zh/transforms/transform-multi-table.md
b/docs/zh/transforms/transform-multi-table.md
index 8fa346a422..ae0f08882b 100644
--- a/docs/zh/transforms/transform-multi-table.md
+++ b/docs/zh/transforms/transform-multi-table.md
@@ -20,7 +20,8 @@ SeaTunnel 的 Multi-Table Transform 支持在上游插件输出多个表时，
|----------------------------|--------|----------|---------|--------------------------------------------------------------------------------------------------|
| table_match_regex | String | No | .* |
表名的正则表达式，通过正则表达式来匹配需要进行转换的表，默认匹配所有的表。注意这个表名是上游的真正表名，不是`plugin_output`。
|
| table_transform | List | No | - |
可以通过table_transform列表来指定部分表的规则，当在table_transform中配置某个表的转换规则后，外层针对当前表的规则不会生效，以table_transform中的为准
|
-| table_transform.table_path | String | No | - |
当在table_transform中配置某个表的转换规则后，需要使用table_path字段指定表名，表名需要包含`databaseName[.schemaName].tableName`。
|
+| table_transform.table_path | String | No | - |
当在table_transform中配置某个表的转换规则后，需要使用table_path字段指定表名，表名需要包含`databaseName[.schemaName].tableName`，并且按完整表路径精确匹配。
|
+| rule_match_mode | String | No | - | 控制多个
`table_transform` 配置使用相同精确 `table_path` 时的执行方式，支持 `FIRST_MATCH` 和 `ALL_MATCH`。 |
## 匹配逻辑
@@ -62,6 +63,36 @@ transform {
对于每个表来说，配置的优先级是：`table_transform` >
`table_match_regex`。如果所有的规则都没有匹配到，那么该表将不会进行任何转换操作。
+`table_transform.table_path` 使用精确表路径匹配。`rule_match_mode` 只控制多个
`table_transform` 配置使用相同精确 `table_path` 的场景：
+
+- 如果没有配置 `rule_match_mode`，重复的精确 `table_path` 配置会在配置解析阶段失败。
+- `FIRST_MATCH`：按声明顺序使用第一个匹配的 `table_transform` 配置。
+- `ALL_MATCH`：按声明顺序执行所有匹配的 `table_transform` 配置。前一个规则的输出会作为同一张表上后一个规则的输入。
+
+例如，当 `rule_match_mode` 为 `ALL_MATCH` 时，下面的配置会先为 `test.xyz` 将 `name` 复制为
`name2`，然后继续将 `name2` 复制为 `name3`：
+
+```hocon
+transform {
+ Copy {
+ rule_match_mode = "ALL_MATCH"
+
+ table_transform = [{
+ table_path = "test.xyz"
+ src_field = "name"
+ dest_field = "name2"
+ }, {
+ table_path = "test.xyz"
+ src_field = "name2"
+ dest_field = "name3"
+ }]
+ }
+}
+```
+
+输出表结构：
+
+| id | name | age | name2 | name3 |
+
针对每个表来说，他们的Transform配置是：
- **test.abc**和**test.abcd**
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/adaptsink/DefineSinkTypeTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/adaptsink/DefineSinkTypeTransformFactory.java
index 2f0d986b22..bbdb6d4650 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/adaptsink/DefineSinkTypeTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/adaptsink/DefineSinkTypeTransformFactory.java
@@ -54,6 +54,7 @@ public class DefineSinkTypeTransformFactory implements
TableTransformFactory {
new
ColumnsStructureValidator())))
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogFlatMapTransform.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogFlatMapTransform.java
index b1deb453d6..3e89ae62c6 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogFlatMapTransform.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogFlatMapTransform.java
@@ -21,6 +21,7 @@ import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.transform.SeaTunnelFlatMapTransform;
+import org.apache.seatunnel.api.transform.SeaTunnelTransform;
import java.util.List;
@@ -33,6 +34,15 @@ public abstract class AbstractMultiCatalogFlatMapTransform
extends AbstractMulti
super(inputCatalogTables, config);
}
+ @Override
+ protected SeaTunnelTransform<SeaTunnelRow> composeTransforms(
+ List<SeaTunnelTransform<SeaTunnelRow>> transforms) {
+ if (transforms.size() == 1) {
+ return transforms.get(0);
+ }
+ return new ChainedFlatMapTransform(transforms);
+ }
+
@Override
public List<SeaTunnelRow> flatMap(SeaTunnelRow row) {
if (transformMap.size() == 1) {
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogMapTransform.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogMapTransform.java
index 9e7d184c09..463803d366 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogMapTransform.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogMapTransform.java
@@ -21,6 +21,7 @@ import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.transform.SeaTunnelMapTransform;
+import org.apache.seatunnel.api.transform.SeaTunnelTransform;
import java.util.List;
@@ -33,6 +34,15 @@ public abstract class AbstractMultiCatalogMapTransform
extends AbstractMultiCata
super(inputCatalogTables, config);
}
+ @Override
+ protected SeaTunnelTransform<SeaTunnelRow> composeTransforms(
+ List<SeaTunnelTransform<SeaTunnelRow>> transforms) {
+ if (transforms.size() == 1) {
+ return transforms.get(0);
+ }
+ return new ChainedMapTransform(transforms);
+ }
+
@Override
public SeaTunnelRow map(SeaTunnelRow row) {
if (transformMap.size() == 1) {
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogTransform.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogTransform.java
index 0d3a5365c6..f3806109a7 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogTransform.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogTransform.java
@@ -25,10 +25,11 @@ import
org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.transform.SeaTunnelTransform;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.function.Function;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
@@ -50,28 +51,42 @@ public abstract class AbstractMultiCatalogTransform
implements SeaTunnelTransfor
this.transformMap = new HashMap<>();
Pattern tableMatchRegex =
Pattern.compile(config.get(TransformCommonOptions.TABLE_MATCH_REGEX));
- Map<String, ReadonlyConfig> singleTableConfig =
+ boolean ruleMatchModeConfigured =
+
config.getOptional(TransformCommonOptions.RULE_MATCH_MODE).isPresent();
+ TransformCommonOptions.RuleMatchMode ruleMatchMode =
+ config.get(TransformCommonOptions.RULE_MATCH_MODE);
+ List<ReadonlyConfig> singleTableConfigs =
config.get(TransformCommonOptions.MULTI_TABLES).stream()
.map(ReadonlyConfig::fromMap)
.filter(c -> c.get(TransformCommonOptions.TABLE_PATH)
!= null)
- .collect(
- Collectors.toMap(
- c ->
c.get(TransformCommonOptions.TABLE_PATH),
- Function.identity()));
+ .collect(Collectors.toList());
+ if (!ruleMatchModeConfigured) {
+ rejectDuplicateTablePaths(singleTableConfigs);
+ }
inputCatalogTables.forEach(
inputCatalogTable -> {
String tableId =
inputCatalogTable.getTableId().toTablePath().toString();
- ReadonlyConfig tableConfig;
- if (singleTableConfig.containsKey(tableId)) {
- tableConfig = singleTableConfig.get(tableId);
+ List<ReadonlyConfig> tableConfigs =
+ singleTableConfigs.stream()
+ .filter(
+ c ->
+ tableId.equals(
+ c.get(
+
TransformCommonOptions
+
.TABLE_PATH)))
+ .collect(Collectors.toList());
+ if (!tableConfigs.isEmpty()) {
+ transformMap.put(
+ tableId,
+ buildTransform(
+ inputCatalogTable,
+ selectTableConfigs(tableConfigs,
ruleMatchMode)));
} else if (tableMatchRegex.matcher(tableId).matches()) {
- tableConfig = config;
- } else {
- tableConfig = null;
- }
- if (tableConfig != null) {
- transformMap.put(tableId,
buildTransform(inputCatalogTable, tableConfig));
+ transformMap.put(
+ tableId,
+ buildTransform(
+ inputCatalogTable,
Collections.singletonList(config)));
} else {
transformMap.put(tableId,
createIdentityTransform(inputCatalogTable));
}
@@ -91,6 +106,49 @@ public abstract class AbstractMultiCatalogTransform
implements SeaTunnelTransfor
protected abstract SeaTunnelTransform<SeaTunnelRow> buildTransform(
CatalogTable inputCatalogTable, ReadonlyConfig config);
+ private void rejectDuplicateTablePaths(List<ReadonlyConfig> tableConfigs) {
+ Map<String, Boolean> tablePaths = new HashMap<>();
+ for (ReadonlyConfig tableConfig : tableConfigs) {
+ String tablePath =
tableConfig.get(TransformCommonOptions.TABLE_PATH);
+ if (tablePaths.put(tablePath, true) != null) {
+ throw new IllegalStateException(
+ String.format(
+ "Duplicate table_transform rules are
configured for table_path [%s]",
+ tablePath));
+ }
+ }
+ }
+
+ private List<ReadonlyConfig> selectTableConfigs(
+ List<ReadonlyConfig> tableConfigs,
TransformCommonOptions.RuleMatchMode ruleMatchMode) {
+ if (ruleMatchMode == TransformCommonOptions.RuleMatchMode.FIRST_MATCH)
{
+ return Collections.singletonList(tableConfigs.get(0));
+ }
+ return tableConfigs;
+ }
+
+ private SeaTunnelTransform<SeaTunnelRow> buildTransform(
+ CatalogTable inputCatalogTable, List<ReadonlyConfig> configs) {
+ List<SeaTunnelTransform<SeaTunnelRow>> transforms = new ArrayList<>();
+ CatalogTable currentCatalogTable = inputCatalogTable;
+ for (ReadonlyConfig config : configs) {
+ SeaTunnelTransform<SeaTunnelRow> transform =
+ buildTransform(currentCatalogTable, config);
+ transforms.add(transform);
+ currentCatalogTable = transform.getProducedCatalogTable();
+ }
+ return composeTransforms(transforms);
+ }
+
+ protected SeaTunnelTransform<SeaTunnelRow> composeTransforms(
+ List<SeaTunnelTransform<SeaTunnelRow>> transforms) {
+ if (transforms.size() == 1) {
+ return transforms.get(0);
+ }
+ throw new UnsupportedOperationException(
+ String.format("%s does not support chained transforms",
getPluginName()));
+ }
+
protected abstract SeaTunnelTransform<SeaTunnelRow>
createIdentityTransform(
CatalogTable catalogTable);
@@ -118,18 +176,24 @@ public abstract class AbstractMultiCatalogTransform
implements SeaTunnelTransfor
String targetTableId = event.tablePath().toString();
SeaTunnelTransform<SeaTunnelRow> targetTransform =
transformMap.get(targetTableId);
if (targetTransform != null) {
- targetTransform.mapSchemaChangeEvent(event);
+ SchemaChangeEvent mappedEvent =
targetTransform.mapSchemaChangeEvent(event);
+ if (mappedEvent == null) {
+ return null;
+ }
refreshOutputCatalogTables();
// Propagate this transform's actual produced catalog through the
chain via the event's
// changeAfter field (existing API designed for exactly this).
Downstream transforms
// and the sink read changeAfter to adopt upstream's actual row
layout instead of
// re-applying ALTER locally — which would diverge from the actual
row order.
- if (event instanceof AlterTableEvent) {
+ if (mappedEvent instanceof AlterTableEvent) {
outputCatalogTables.stream()
.filter(t ->
t.getTableId().toTablePath().toString().equals(targetTableId))
.findFirst()
- .ifPresent(produced -> ((AlterTableEvent)
event).setChangeAfter(produced));
+ .ifPresent(
+ produced ->
+ ((AlterTableEvent)
mappedEvent).setChangeAfter(produced));
}
+ return mappedEvent;
}
return event;
}
@@ -151,6 +215,8 @@ public abstract class AbstractMultiCatalogTransform
implements SeaTunnelTransfor
SeaTunnelTransform<SeaTunnelRow> inner = transformMap.get(tableId);
if (inner instanceof AbstractSeaTunnelTransform) {
((AbstractSeaTunnelTransform<?, ?>)
inner).setInputCatalogTable(newInput);
+ } else if (inner != null) {
+
inner.setInputCatalogTables(Collections.singletonList(newInput));
}
}
refreshOutputCatalogTables();
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/ChainedFlatMapTransform.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/ChainedFlatMapTransform.java
new file mode 100644
index 0000000000..28e5895240
--- /dev/null
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/ChainedFlatMapTransform.java
@@ -0,0 +1,116 @@
+/*
+ * 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.transform.common;
+
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.transform.SeaTunnelFlatMapTransform;
+import org.apache.seatunnel.api.transform.SeaTunnelTransform;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+class ChainedFlatMapTransform implements
SeaTunnelFlatMapTransform<SeaTunnelRow> {
+
+ private final List<SeaTunnelTransform<SeaTunnelRow>> transforms;
+
+ ChainedFlatMapTransform(List<SeaTunnelTransform<SeaTunnelRow>> transforms)
{
+ this.transforms = transforms;
+ }
+
+ @Override
+ public String getPluginName() {
+ return transforms.get(0).getPluginName();
+ }
+
+ @Override
+ public void open() {
+ transforms.forEach(SeaTunnelTransform::open);
+ }
+
+ @Override
+ public List<SeaTunnelRow> flatMap(SeaTunnelRow row) {
+ List<SeaTunnelRow> currentRows = Collections.singletonList(row);
+ for (SeaTunnelTransform<SeaTunnelRow> transform : transforms) {
+ List<SeaTunnelRow> nextRows = new ArrayList<>();
+ for (SeaTunnelRow currentRow : currentRows) {
+ List<SeaTunnelRow> rows =
+ ((SeaTunnelFlatMapTransform<SeaTunnelRow>)
transform).flatMap(currentRow);
+ if (rows != null) {
+ nextRows.addAll(rows);
+ }
+ }
+ currentRows = nextRows;
+ }
+ return currentRows;
+ }
+
+ @Override
+ public CatalogTable getProducedCatalogTable() {
+ return transforms.get(transforms.size() - 1).getProducedCatalogTable();
+ }
+
+ @Override
+ public List<CatalogTable> getProducedCatalogTables() {
+ return Collections.singletonList(getProducedCatalogTable());
+ }
+
+ @Override
+ public SchemaChangeEvent mapSchemaChangeEvent(SchemaChangeEvent
schemaChangeEvent) {
+ SchemaChangeEvent currentEvent = schemaChangeEvent;
+ for (int i = 0; i < transforms.size(); i++) {
+ SeaTunnelTransform<SeaTunnelRow> transform = transforms.get(i);
+ currentEvent = transform.mapSchemaChangeEvent(currentEvent);
+ if (currentEvent == null) {
+ return null;
+ }
+ if (i + 1 < transforms.size()) {
+ setTransformInput(transforms.get(i + 1),
transform.getProducedCatalogTable());
+ }
+ }
+ return currentEvent;
+ }
+
+ @Override
+ public void setInputCatalogTables(List<CatalogTable> inputCatalogTables) {
+ if (inputCatalogTables == null || inputCatalogTables.isEmpty()) {
+ return;
+ }
+ CatalogTable currentCatalogTable = inputCatalogTables.get(0);
+ for (SeaTunnelTransform<SeaTunnelRow> transform : transforms) {
+ setTransformInput(transform, currentCatalogTable);
+ currentCatalogTable = transform.getProducedCatalogTable();
+ }
+ }
+
+ @Override
+ public void close() {
+ transforms.forEach(SeaTunnelTransform::close);
+ }
+
+ private void setTransformInput(
+ SeaTunnelTransform<SeaTunnelRow> transform, CatalogTable
catalogTable) {
+ if (transform instanceof AbstractSeaTunnelTransform) {
+ ((AbstractSeaTunnelTransform<?, ?>)
transform).setInputCatalogTable(catalogTable);
+ } else {
+
transform.setInputCatalogTables(Collections.singletonList(catalogTable));
+ }
+ }
+}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/ChainedMapTransform.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/ChainedMapTransform.java
new file mode 100644
index 0000000000..26750f869d
--- /dev/null
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/ChainedMapTransform.java
@@ -0,0 +1,110 @@
+/*
+ * 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.transform.common;
+
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.transform.SeaTunnelMapTransform;
+import org.apache.seatunnel.api.transform.SeaTunnelTransform;
+
+import java.util.Collections;
+import java.util.List;
+
+class ChainedMapTransform implements SeaTunnelMapTransform<SeaTunnelRow> {
+
+ private final List<SeaTunnelTransform<SeaTunnelRow>> transforms;
+
+ ChainedMapTransform(List<SeaTunnelTransform<SeaTunnelRow>> transforms) {
+ this.transforms = transforms;
+ }
+
+ @Override
+ public String getPluginName() {
+ return transforms.get(0).getPluginName();
+ }
+
+ @Override
+ public void open() {
+ transforms.forEach(SeaTunnelTransform::open);
+ }
+
+ @Override
+ public SeaTunnelRow map(SeaTunnelRow row) {
+ SeaTunnelRow currentRow = row;
+ for (SeaTunnelTransform<SeaTunnelRow> transform : transforms) {
+ if (currentRow == null) {
+ return null;
+ }
+ currentRow = ((SeaTunnelMapTransform<SeaTunnelRow>)
transform).map(currentRow);
+ }
+ return currentRow;
+ }
+
+ @Override
+ public CatalogTable getProducedCatalogTable() {
+ return transforms.get(transforms.size() - 1).getProducedCatalogTable();
+ }
+
+ @Override
+ public List<CatalogTable> getProducedCatalogTables() {
+ return Collections.singletonList(getProducedCatalogTable());
+ }
+
+ @Override
+ public SchemaChangeEvent mapSchemaChangeEvent(SchemaChangeEvent
schemaChangeEvent) {
+ SchemaChangeEvent currentEvent = schemaChangeEvent;
+ for (int i = 0; i < transforms.size(); i++) {
+ SeaTunnelTransform<SeaTunnelRow> transform = transforms.get(i);
+ currentEvent = transform.mapSchemaChangeEvent(currentEvent);
+ if (currentEvent == null) {
+ return null;
+ }
+ if (i + 1 < transforms.size()) {
+ setTransformInput(transforms.get(i + 1),
transform.getProducedCatalogTable());
+ }
+ }
+ return currentEvent;
+ }
+
+ @Override
+ public void setInputCatalogTables(List<CatalogTable> inputCatalogTables) {
+ if (inputCatalogTables == null || inputCatalogTables.isEmpty()) {
+ return;
+ }
+ CatalogTable currentCatalogTable = inputCatalogTables.get(0);
+ for (SeaTunnelTransform<SeaTunnelRow> transform : transforms) {
+ setTransformInput(transform, currentCatalogTable);
+ currentCatalogTable = transform.getProducedCatalogTable();
+ }
+ }
+
+ @Override
+ public void close() {
+ transforms.forEach(SeaTunnelTransform::close);
+ }
+
+ private void setTransformInput(
+ SeaTunnelTransform<SeaTunnelRow> transform, CatalogTable
catalogTable) {
+ if (transform instanceof AbstractSeaTunnelTransform) {
+ ((AbstractSeaTunnelTransform<?, ?>)
transform).setInputCatalogTable(catalogTable);
+ } else {
+
transform.setInputCatalogTables(Collections.singletonList(catalogTable));
+ }
+ }
+}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/TransformCommonOptions.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/TransformCommonOptions.java
index daafc08e2d..268bf1cef9 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/TransformCommonOptions.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/TransformCommonOptions.java
@@ -47,6 +47,12 @@ public class TransformCommonOptions {
.defaultValue(".*")
.withDescription("The regex to match the table path");
+ public static final Option<RuleMatchMode> RULE_MATCH_MODE =
+ Options.key("rule_match_mode")
+ .enumType(RuleMatchMode.class)
+ .defaultValue(RuleMatchMode.FIRST_MATCH)
+ .withDescription("The rule match mode for table transform
config");
+
public static final Option<ErrorHandleWay> ROW_ERROR_HANDLE_WAY_OPTION =
Options.key("row_error_handle_way")
.singleChoice(
@@ -76,4 +82,9 @@ public class TransformCommonOptions {
.noDefaultValue()
.withDescription(
"Target table name for routing invalid data when
error_handle_way is ROUTE_TO_TABLE");
+
+ public enum RuleMatchMode {
+ FIRST_MATCH,
+ ALL_MATCH
+ }
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/copy/CopyFieldTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/copy/CopyFieldTransformFactory.java
index 5de39766ff..4d69bf2ca2 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/copy/CopyFieldTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/copy/CopyFieldTransformFactory.java
@@ -50,6 +50,7 @@ public class CopyFieldTransformFactory implements
TableTransformFactory {
.optional(CopyTransformConfig.DEST_FIELD)
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/dynamiccompile/DynamicCompileTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/dynamiccompile/DynamicCompileTransformFactory.java
index 02141fd0a9..a79fd7104a 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/dynamiccompile/DynamicCompileTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/dynamiccompile/DynamicCompileTransformFactory.java
@@ -57,6 +57,7 @@ public class DynamicCompileTransformFactory implements
TableTransformFactory {
Conditions.notBlank(DynamicCompileTransformConfig.ABSOLUTE_PATH))
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/encrypt/FieldEncryptTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/encrypt/FieldEncryptTransformFactory.java
index 3c33a095ff..9544b65803 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/encrypt/FieldEncryptTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/encrypt/FieldEncryptTransformFactory.java
@@ -52,6 +52,7 @@ public class FieldEncryptTransformFactory implements
TableTransformFactory {
Conditions.greaterThan(FieldEncryptTransformConfig.MAX_FIELD_LENGTH, 0))
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/fieldmapper/FieldMapperTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/fieldmapper/FieldMapperTransformFactory.java
index 8edb4419e3..f9e0f15852 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/fieldmapper/FieldMapperTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/fieldmapper/FieldMapperTransformFactory.java
@@ -43,6 +43,7 @@ public class FieldMapperTransformFactory implements
TableTransformFactory {
Conditions.mapNotEmpty(FieldMapperTransformConfig.FIELD_MAPPER))
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/filter/FilterFieldTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/filter/FilterFieldTransformFactory.java
index bdc273ed27..7860a50dee 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/filter/FilterFieldTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/filter/FilterFieldTransformFactory.java
@@ -50,6 +50,7 @@ public class FilterFieldTransformFactory implements
TableTransformFactory {
Conditions.notEmpty(FilterFieldTransformConfig.EXCLUDE_FIELDS))
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/filterrowkind/FilterRowKindTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/filterrowkind/FilterRowKindTransformFactory.java
index 80dea1693f..5023f37e60 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/filterrowkind/FilterRowKindTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/filterrowkind/FilterRowKindTransformFactory.java
@@ -41,6 +41,7 @@ public class FilterRowKindTransformFactory implements
TableTransformFactory {
FilterRowKinkTransformConfig.INCLUDE_KINDS)
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/jsonpath/JsonPathTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/jsonpath/JsonPathTransformFactory.java
index d0411fddb3..59b4d36d00 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/jsonpath/JsonPathTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/jsonpath/JsonPathTransformFactory.java
@@ -52,6 +52,7 @@ public class JsonPathTransformFactory implements
TableTransformFactory {
new ColumnsValidator())))
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.optional(TransformCommonOptions.ROW_ERROR_HANDLE_WAY_OPTION)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/metadata/MetadataTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/metadata/MetadataTransformFactory.java
index 479376b888..2988d7ae38 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/metadata/MetadataTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/metadata/MetadataTransformFactory.java
@@ -42,6 +42,7 @@ public class MetadataTransformFactory implements
TableTransformFactory {
Conditions.mapNotEmpty(MetadataTransformConfig.METADATA_FIELDS))
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/nlpmodel/embedding/EmbeddingTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/nlpmodel/embedding/EmbeddingTransformFactory.java
index f75a1eb430..95cc465b73 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/nlpmodel/embedding/EmbeddingTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/nlpmodel/embedding/EmbeddingTransformFactory.java
@@ -81,6 +81,7 @@ public class EmbeddingTransformFactory implements
TableTransformFactory {
EmbeddingTransformConfig.DIMENSION)
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/nlpmodel/llm/LLMTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/nlpmodel/llm/LLMTransformFactory.java
index e703bc9457..67f855b4c6 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/nlpmodel/llm/LLMTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/nlpmodel/llm/LLMTransformFactory.java
@@ -66,6 +66,7 @@ public class LLMTransformFactory implements
TableTransformFactory {
LLMTransformConfig.CustomRequestConfig.CUSTOM_CONFIG)
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/regexextract/RegexExtractTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/regexextract/RegexExtractTransformFactory.java
index 0502bd27fc..5eee24a304 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/regexextract/RegexExtractTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/regexextract/RegexExtractTransformFactory.java
@@ -52,7 +52,8 @@ public class RegexExtractTransformFactory implements
TableTransformFactory {
Conditions.extension(
RegexExtractTransformConfig.KEY_DEFAULT_VALUES,
new DefaultValuesLengthValidator()))
- .optional(TransformCommonOptions.MULTI_TABLES)
+ .optional(
+ TransformCommonOptions.MULTI_TABLES,
TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rename/FieldRenameTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rename/FieldRenameTransformFactory.java
index 299bbb5255..78200dfa67 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rename/FieldRenameTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rename/FieldRenameTransformFactory.java
@@ -46,6 +46,7 @@ public class FieldRenameTransformFactory implements
TableTransformFactory {
.optional(CONVERT_CASE, PREFIX, SUFFIX,
REPLACEMENTS_WITH_REGEX, SPECIFIC)
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rename/TableRenameTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rename/TableRenameTransformFactory.java
index 840bfff209..a581849fa9 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rename/TableRenameTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rename/TableRenameTransformFactory.java
@@ -44,6 +44,7 @@ public class TableRenameTransformFactory implements
TableTransformFactory {
.optional(CONVERT_CASE, PREFIX, SUFFIX,
REPLACEMENTS_WITH_REGEX)
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/replace/ReplaceTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/replace/ReplaceTransformFactory.java
index c7c7e1ff64..5e3c32b1ed 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/replace/ReplaceTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/replace/ReplaceTransformFactory.java
@@ -46,6 +46,7 @@ public class ReplaceTransformFactory implements
TableTransformFactory {
.optional(ReplaceTransformConfig.KEY_REPLACE_FIRST)
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rowkind/RowKindExtractorTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rowkind/RowKindExtractorTransformFactory.java
index 99011726fc..9837c1b019 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rowkind/RowKindExtractorTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/rowkind/RowKindExtractorTransformFactory.java
@@ -40,6 +40,7 @@ public class RowKindExtractorTransformFactory implements
TableTransformFactory {
.optional(RowKindExtractorTransformConfig.TRANSFORM_TYPE)
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/split/SplitTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/split/SplitTransformFactory.java
index d5385a4f59..b65e5fbefd 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/split/SplitTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/split/SplitTransformFactory.java
@@ -44,6 +44,7 @@ public class SplitTransformFactory implements
TableTransformFactory {
Conditions.notEmpty(SplitTransformConfig.KEY_OUTPUT_FIELDS))
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLTransformFactory.java
index 4a3b6db9c5..c1ddda2af7 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLTransformFactory.java
@@ -41,6 +41,7 @@ public class SQLTransformFactory implements
TableTransformFactory {
.required(KEY_QUERY)
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/table/TableFilterTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/table/TableFilterTransformFactory.java
index 701d9f5697..cde7685f86 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/table/TableFilterTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/table/TableFilterTransformFactory.java
@@ -56,6 +56,7 @@ public class TableFilterTransformFactory implements
TableTransformFactory {
.optional(TableFilterConfig.PATTERN_MODE)
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/table/TableMergeTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/table/TableMergeTransformFactory.java
index 6bf876a235..45bfde346c 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/table/TableMergeTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/table/TableMergeTransformFactory.java
@@ -41,6 +41,7 @@ public class TableMergeTransformFactory implements
TableTransformFactory {
.optional(TableMergeConfig.DATABASE, TableMergeConfig.SCHEMA)
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.build();
}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/validator/DataValidatorTransformFactory.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/validator/DataValidatorTransformFactory.java
index b391d8339a..e633bfa8db 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/validator/DataValidatorTransformFactory.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/validator/DataValidatorTransformFactory.java
@@ -53,6 +53,7 @@ public class DataValidatorTransformFactory implements
TableTransformFactory {
.and(Conditions.extension(FIELD_RULES, new
FieldRulesValidator())))
.optional(TransformCommonOptions.MULTI_TABLES)
.optional(TransformCommonOptions.TABLE_MATCH_REGEX)
+ .optional(TransformCommonOptions.RULE_MATCH_MODE)
.optional(TransformCommonOptions.ROW_ERROR_HANDLE_WAY_OPTION)
.optional(TransformCommonOptions.ERROR_TABLE_OPTION)
.build();
diff --git
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogTransformTest.java
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogTransformTest.java
new file mode 100644
index 0000000000..5dc403dc70
--- /dev/null
+++
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogTransformTest.java
@@ -0,0 +1,476 @@
+/*
+ * 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.transform.common;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.catalog.Column;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.api.transform.SeaTunnelTransform;
+
+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.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+class AbstractMultiCatalogTransformTest {
+
+ private static final Option<String> FIELD_NAME =
+ Options.key("field_name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Field name to append");
+
+ private static final Option<String> REQUIRED_FIELD =
+ Options.key("required_field")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Field that must exist before this rule
is built");
+
+ private static final Option<Boolean> FILTER_SCHEMA_CHANGE_EVENT =
+ Options.key("filter_schema_change_event")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Return null for schema-change events");
+
+ private static final Option<Boolean> FAIL_ON_SCHEMA_CHANGE_EVENT =
+ Options.key("fail_on_schema_change_event")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Fail if a schema-change event reaches
this rule");
+
+ @Test
+ void defaultModeRejectsDuplicateExactTablePathRules() {
+ CatalogTable table = catalogTable();
+ String tablePath = tablePath(table);
+ ReadonlyConfig config =
+ config(tableRule(tablePath, "first"), tableRule(tablePath,
"second"));
+
+ IllegalStateException exception =
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ new TestMapMultiCatalogTransform(
+ Collections.singletonList(table),
config));
+
+ Assertions.assertTrue(exception.getMessage().contains(tablePath));
+ }
+
+ @Test
+ void defaultModeRejectsDuplicateExactTablePathRulesBeforeTableMatching() {
+ CatalogTable table = catalogTable();
+ String duplicateTablePath = "database.schema.customers";
+ ReadonlyConfig config =
+ config(
+ tableRule(duplicateTablePath, "first"),
+ tableRule(duplicateTablePath, "second"));
+
+ IllegalStateException exception =
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ new TestMapMultiCatalogTransform(
+ Collections.singletonList(table),
config));
+
+
Assertions.assertTrue(exception.getMessage().contains(duplicateTablePath));
+ }
+
+ @Test
+ void firstMatchModeUsesOnlyFirstExactTablePathRule() {
+ CatalogTable table = catalogTable();
+ String tablePath = tablePath(table);
+ ReadonlyConfig config =
+ config(
+ TransformCommonOptions.RuleMatchMode.FIRST_MATCH,
+ tableRule(tablePath, "first"),
+ tableRule(tablePath, "second"));
+ TestMapMultiCatalogTransform transform =
+ new
TestMapMultiCatalogTransform(Collections.singletonList(table), config);
+
+ SeaTunnelRow output = transform.map(row(tablePath));
+
+ Assertions.assertEquals(Arrays.asList("id", "first"),
fieldNames(transform));
+ Assertions.assertEquals(2, output.getArity());
+ Assertions.assertEquals("first", output.getField(1));
+ }
+
+ @Test
+ void allMatchModeAppliesExactTablePathRulesInDeclarationOrder() {
+ CatalogTable table = catalogTable();
+ String tablePath = tablePath(table);
+ Map<String, Object> secondRule = tableRule(tablePath, "second");
+ secondRule.put(REQUIRED_FIELD.key(), "first");
+ ReadonlyConfig config =
+ config(
+ TransformCommonOptions.RuleMatchMode.ALL_MATCH,
+ tableRule(tablePath, "first"),
+ secondRule);
+ TestMapMultiCatalogTransform transform =
+ new
TestMapMultiCatalogTransform(Collections.singletonList(table), config);
+
+ SeaTunnelRow output = transform.map(row(tablePath));
+
+ Assertions.assertEquals(Arrays.asList("id", "first", "second"),
fieldNames(transform));
+ Assertions.assertEquals(3, output.getArity());
+ Assertions.assertEquals("first", output.getField(1));
+ Assertions.assertEquals("second", output.getField(2));
+ }
+
+ @Test
+ void allMatchModeWorksForFlatMapTransforms() {
+ CatalogTable table = catalogTable();
+ String tablePath = tablePath(table);
+ ReadonlyConfig config =
+ config(
+ TransformCommonOptions.RuleMatchMode.ALL_MATCH,
+ tableRule(tablePath, "first"),
+ tableRule(tablePath, "second"));
+ TestFlatMapMultiCatalogTransform transform =
+ new
TestFlatMapMultiCatalogTransform(Collections.singletonList(table), config);
+
+ List<SeaTunnelRow> outputRows = transform.flatMap(row(tablePath));
+
+ Assertions.assertEquals(Arrays.asList("id", "first", "second"),
fieldNames(transform));
+ Assertions.assertEquals(1, outputRows.size());
+ Assertions.assertEquals("first", outputRows.get(0).getField(1));
+ Assertions.assertEquals("second", outputRows.get(0).getField(2));
+ }
+
+ @Test
+ void chainedMapTransformStopsSchemaChangeAfterFilteredEvent() {
+ CatalogTable table = catalogTable();
+ String tablePath = tablePath(table);
+ Map<String, Object> firstRule = tableRule(tablePath, "first");
+ firstRule.put(FILTER_SCHEMA_CHANGE_EVENT.key(), true);
+ Map<String, Object> secondRule = tableRule(tablePath, "second");
+ secondRule.put(FAIL_ON_SCHEMA_CHANGE_EVENT.key(), true);
+ ReadonlyConfig config =
+ config(TransformCommonOptions.RuleMatchMode.ALL_MATCH,
firstRule, secondRule);
+ TestMapMultiCatalogTransform transform =
+ new
TestMapMultiCatalogTransform(Collections.singletonList(table), config);
+
+ SchemaChangeEvent event =
transform.mapSchemaChangeEvent(addColumnEvent(table));
+
+ Assertions.assertNull(event);
+ }
+
+ @Test
+ void chainedFlatMapTransformStopsSchemaChangeAfterFilteredEvent() {
+ CatalogTable table = catalogTable();
+ String tablePath = tablePath(table);
+ Map<String, Object> firstRule = tableRule(tablePath, "first");
+ firstRule.put(FILTER_SCHEMA_CHANGE_EVENT.key(), true);
+ Map<String, Object> secondRule = tableRule(tablePath, "second");
+ secondRule.put(FAIL_ON_SCHEMA_CHANGE_EVENT.key(), true);
+ ReadonlyConfig config =
+ config(TransformCommonOptions.RuleMatchMode.ALL_MATCH,
firstRule, secondRule);
+ TestFlatMapMultiCatalogTransform transform =
+ new
TestFlatMapMultiCatalogTransform(Collections.singletonList(table), config);
+
+ SchemaChangeEvent event =
transform.mapSchemaChangeEvent(addColumnEvent(table));
+
+ Assertions.assertNull(event);
+ }
+
+ @Test
+ void tableMatchRegexIsFallbackWhenExactTablePathRuleDoesNotMatch() {
+ CatalogTable table = catalogTable();
+ String tablePath = tablePath(table);
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put(TransformCommonOptions.TABLE_MATCH_REGEX.key(),
".*orders");
+ configMap.put(FIELD_NAME.key(), "fallback");
+ configMap.put(
+ TransformCommonOptions.MULTI_TABLES.key(),
+
Collections.singletonList(tableRule("database.schema.customers", "specific")));
+ TestMapMultiCatalogTransform transform =
+ new TestMapMultiCatalogTransform(
+ Collections.singletonList(table),
ReadonlyConfig.fromMap(configMap));
+
+ SeaTunnelRow output = transform.map(row(tablePath));
+
+ Assertions.assertEquals(Arrays.asList("id", "fallback"),
fieldNames(transform));
+ Assertions.assertEquals("fallback", output.getField(1));
+ }
+
+ @Test
+ void identityTransformIsUsedWhenNoRuleMatches() {
+ CatalogTable table = catalogTable();
+ String tablePath = tablePath(table);
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put(TransformCommonOptions.TABLE_MATCH_REGEX.key(),
"does-not-match");
+ TestMapMultiCatalogTransform transform =
+ new TestMapMultiCatalogTransform(
+ Collections.singletonList(table),
ReadonlyConfig.fromMap(configMap));
+ SeaTunnelRow input = row(tablePath);
+
+ SeaTunnelRow output = transform.map(input);
+
+ Assertions.assertSame(input, output);
+ Assertions.assertEquals(Collections.singletonList("id"),
fieldNames(transform));
+ }
+
+ private static ReadonlyConfig config(Map<String, Object>... tableRules) {
+ return config(null, tableRules);
+ }
+
+ private static ReadonlyConfig config(
+ TransformCommonOptions.RuleMatchMode ruleMatchMode, Map<String,
Object>... tableRules) {
+ Map<String, Object> configMap = new HashMap<>();
+ if (ruleMatchMode != null) {
+ configMap.put(TransformCommonOptions.RULE_MATCH_MODE.key(),
ruleMatchMode);
+ }
+ configMap.put(TransformCommonOptions.MULTI_TABLES.key(),
Arrays.asList(tableRules));
+ return ReadonlyConfig.fromMap(configMap);
+ }
+
+ private static Map<String, Object> tableRule(String tablePath, String
fieldName) {
+ Map<String, Object> rule = new HashMap<>();
+ rule.put(TransformCommonOptions.TABLE_PATH.key(), tablePath);
+ rule.put(FIELD_NAME.key(), fieldName);
+ return rule;
+ }
+
+ private static CatalogTable catalogTable() {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"id"}, new SeaTunnelDataType[]
{BasicType.INT_TYPE});
+ return CatalogTableUtil.getCatalogTable("catalog", "database",
"schema", "orders", rowType);
+ }
+
+ private static String tablePath(CatalogTable table) {
+ return table.getTableId().toTablePath().toString();
+ }
+
+ private static SeaTunnelRow row(String tablePath) {
+ SeaTunnelRow row = new SeaTunnelRow(new Object[] {1});
+ row.setTableId(tablePath);
+ return row;
+ }
+
+ private static SchemaChangeEvent addColumnEvent(CatalogTable table) {
+ return AlterTableAddColumnEvent.add(
+ table.getTableId(),
+ PhysicalColumn.of("created_at", BasicType.STRING_TYPE, 0L,
true, null, null));
+ }
+
+ private static List<String> fieldNames(SeaTunnelTransform<SeaTunnelRow>
transform) {
+ return
transform.getProducedCatalogTable().getTableSchema().getColumns().stream()
+ .map(Column::getName)
+ .collect(Collectors.toList());
+ }
+
+ private static SeaTunnelRow appendField(SeaTunnelRow inputRow, String
fieldName) {
+ Object[] fields = Arrays.copyOf(inputRow.getFields(),
inputRow.getArity() + 1);
+ fields[fields.length - 1] = fieldName;
+ SeaTunnelRow outputRow = new SeaTunnelRow(fields);
+ outputRow.setTableId(inputRow.getTableId());
+ outputRow.setRowKind(inputRow.getRowKind());
+ if (inputRow.getOptionsOrNull() != null) {
+ outputRow.setOptions(inputRow.getOptionsOrNull());
+ }
+ return outputRow;
+ }
+
+ private static TableSchema appendField(CatalogTable inputCatalogTable,
String fieldName) {
+ List<Column> columns =
+ inputCatalogTable.getTableSchema().getColumns().stream()
+ .map(Column::copy)
+ .collect(Collectors.toCollection(ArrayList::new));
+ columns.add(PhysicalColumn.of(fieldName, BasicType.STRING_TYPE, 0L,
true, null, null));
+ return TableSchema.builder()
+ .columns(columns)
+ .primaryKey(inputCatalogTable.getTableSchema().getPrimaryKey())
+
.constraintKey(inputCatalogTable.getTableSchema().getConstraintKeys())
+ .build();
+ }
+
+ private static void assertRequiredField(CatalogTable inputCatalogTable,
ReadonlyConfig config) {
+ config.getOptional(REQUIRED_FIELD)
+ .ifPresent(
+ field ->
+ inputCatalogTable
+ .getTableSchema()
+ .toPhysicalRowDataType()
+ .indexOf(field));
+ }
+
+ private static class TestMapMultiCatalogTransform extends
AbstractMultiCatalogMapTransform {
+
+ private TestMapMultiCatalogTransform(
+ List<CatalogTable> inputCatalogTables, ReadonlyConfig config) {
+ super(inputCatalogTables, config);
+ }
+
+ @Override
+ public String getPluginName() {
+ return "TestMap";
+ }
+
+ @Override
+ protected SeaTunnelTransform<SeaTunnelRow> buildTransform(
+ CatalogTable inputCatalogTable, ReadonlyConfig config) {
+ return new AppendFieldMapTransform(inputCatalogTable, config);
+ }
+
+ @Override
+ protected SeaTunnelTransform<SeaTunnelRow> createIdentityTransform(
+ CatalogTable catalogTable) {
+ return new IdentityMapTransform(catalogTable);
+ }
+ }
+
+ private static class TestFlatMapMultiCatalogTransform
+ extends AbstractMultiCatalogFlatMapTransform {
+
+ private TestFlatMapMultiCatalogTransform(
+ List<CatalogTable> inputCatalogTables, ReadonlyConfig config) {
+ super(inputCatalogTables, config);
+ }
+
+ @Override
+ public String getPluginName() {
+ return "TestFlatMap";
+ }
+
+ @Override
+ protected SeaTunnelTransform<SeaTunnelRow> buildTransform(
+ CatalogTable inputCatalogTable, ReadonlyConfig config) {
+ return new AppendFieldFlatMapTransform(inputCatalogTable, config);
+ }
+
+ @Override
+ protected SeaTunnelTransform<SeaTunnelRow> createIdentityTransform(
+ CatalogTable catalogTable) {
+ return new IdentityFlatMapTransform(catalogTable);
+ }
+ }
+
+ private static class AppendFieldMapTransform extends
AbstractCatalogSupportMapTransform {
+
+ private final String fieldName;
+
+ private final boolean filterSchemaChangeEvent;
+
+ private final boolean failOnSchemaChangeEvent;
+
+ private AppendFieldMapTransform(CatalogTable inputCatalogTable,
ReadonlyConfig config) {
+ super(inputCatalogTable);
+ assertRequiredField(inputCatalogTable, config);
+ this.fieldName = config.get(FIELD_NAME);
+ this.filterSchemaChangeEvent =
config.get(FILTER_SCHEMA_CHANGE_EVENT);
+ this.failOnSchemaChangeEvent =
config.get(FAIL_ON_SCHEMA_CHANGE_EVENT);
+ }
+
+ @Override
+ public String getPluginName() {
+ return "AppendFieldMap";
+ }
+
+ @Override
+ protected SeaTunnelRow transformRow(SeaTunnelRow inputRow) {
+ return appendField(inputRow, fieldName);
+ }
+
+ @Override
+ public SchemaChangeEvent mapSchemaChangeEvent(SchemaChangeEvent event)
{
+ if (failOnSchemaChangeEvent) {
+ throw new AssertionError("Schema-change event should not reach
this rule");
+ }
+ if (filterSchemaChangeEvent) {
+ return null;
+ }
+ return super.mapSchemaChangeEvent(event);
+ }
+
+ @Override
+ protected TableSchema transformTableSchema() {
+ return appendField(inputCatalogTable, fieldName);
+ }
+
+ @Override
+ protected TableIdentifier transformTableIdentifier() {
+ return inputCatalogTable.getTableId().copy();
+ }
+ }
+
+ private static class AppendFieldFlatMapTransform
+ extends AbstractCatalogSupportFlatMapTransform {
+
+ private final String fieldName;
+
+ private final boolean filterSchemaChangeEvent;
+
+ private final boolean failOnSchemaChangeEvent;
+
+ private AppendFieldFlatMapTransform(CatalogTable inputCatalogTable,
ReadonlyConfig config) {
+ super(inputCatalogTable);
+ assertRequiredField(inputCatalogTable, config);
+ this.fieldName = config.get(FIELD_NAME);
+ this.filterSchemaChangeEvent =
config.get(FILTER_SCHEMA_CHANGE_EVENT);
+ this.failOnSchemaChangeEvent =
config.get(FAIL_ON_SCHEMA_CHANGE_EVENT);
+ }
+
+ @Override
+ public String getPluginName() {
+ return "AppendFieldFlatMap";
+ }
+
+ @Override
+ protected List<SeaTunnelRow> transformRow(SeaTunnelRow inputRow) {
+ return Collections.singletonList(appendField(inputRow, fieldName));
+ }
+
+ @Override
+ public SchemaChangeEvent mapSchemaChangeEvent(SchemaChangeEvent event)
{
+ if (failOnSchemaChangeEvent) {
+ throw new AssertionError("Schema-change event should not reach
this rule");
+ }
+ if (filterSchemaChangeEvent) {
+ return null;
+ }
+ return event;
+ }
+
+ @Override
+ protected TableSchema transformTableSchema() {
+ return appendField(inputCatalogTable, fieldName);
+ }
+
+ @Override
+ protected TableIdentifier transformTableIdentifier() {
+ return inputCatalogTable.getTableId().copy();
+ }
+ }
+}
diff --git
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/copy/CopyFieldMultiCatalogTransformTest.java
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/copy/CopyFieldMultiCatalogTransformTest.java
new file mode 100644
index 0000000000..5429984aa7
--- /dev/null
+++
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/copy/CopyFieldMultiCatalogTransformTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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.transform.copy;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.catalog.Column;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.transform.common.TransformCommonOptions;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+class CopyFieldMultiCatalogTransformTest {
+
+ @Test
+ void allMatchModeAppliesCopyRulesInDeclarationOrder() {
+ CatalogTable table = catalogTable();
+ String tablePath = table.getTableId().toTablePath().toString();
+ CopyFieldMultiCatalogTransform transform =
+ new CopyFieldMultiCatalogTransform(
+ Collections.singletonList(table),
allMatchConfig(tablePath));
+ SeaTunnelRow row = new SeaTunnelRow(new Object[] {1, "name-value"});
+ row.setTableId(tablePath);
+
+ SeaTunnelRow output = transform.map(row);
+
+ Assertions.assertEquals(
+ Arrays.asList("id", "name", "name2", "name3"),
fieldNames(transform));
+ Assertions.assertEquals(4, output.getArity());
+ Assertions.assertEquals("name-value", output.getField(2));
+ Assertions.assertEquals("name-value", output.getField(3));
+ }
+
+ private static ReadonlyConfig allMatchConfig(String tablePath) {
+ Map<String, Object> firstRule = copyRule(tablePath, "name", "name2");
+ Map<String, Object> secondRule = copyRule(tablePath, "name2", "name3");
+ Map<String, Object> config = new HashMap<>();
+ config.put(
+ TransformCommonOptions.RULE_MATCH_MODE.key(),
+ TransformCommonOptions.RuleMatchMode.ALL_MATCH);
+ config.put(TransformCommonOptions.MULTI_TABLES.key(),
Arrays.asList(firstRule, secondRule));
+ return ReadonlyConfig.fromMap(config);
+ }
+
+ private static Map<String, Object> copyRule(
+ String tablePath, String srcField, String destField) {
+ Map<String, Object> rule = new HashMap<>();
+ rule.put(TransformCommonOptions.TABLE_PATH.key(), tablePath);
+ rule.put(CopyTransformConfig.SRC_FIELD.key(), srcField);
+ rule.put(CopyTransformConfig.DEST_FIELD.key(), destField);
+ return rule;
+ }
+
+ private static CatalogTable catalogTable() {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"id", "name"},
+ new SeaTunnelDataType[] {BasicType.INT_TYPE,
BasicType.STRING_TYPE});
+ return CatalogTableUtil.getCatalogTable("catalog", "database",
"schema", "orders", rowType);
+ }
+
+ private static List<String> fieldNames(CopyFieldMultiCatalogTransform
transform) {
+ return
transform.getProducedCatalogTable().getTableSchema().getColumns().stream()
+ .map(Column::getName)
+ .collect(Collectors.toList());
+ }
+}
