This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 f3e1c2ee3b [Fix][Transform-V2] Rebind Replace fields on schema refresh
(#12365)
f3e1c2ee3b is described below
commit f3e1c2ee3b83efd4a1e7fb4cc4fe40b44060c7d0
Author: Goutam Adwant <[email protected]>
AuthorDate: Sat Oct 31 12:53:59 2026 +0000
[Fix][Transform-V2] Rebind Replace fields on schema refresh (#12365)
Signed-off-by: Goutam Adwant <[email protected]>
---
.../introduction/concepts/incompatible-changes.md | 5 +
docs/en/transforms/replace.md | 5 +-
.../introduction/concepts/incompatible-changes.md | 5 +
docs/zh/transforms/replace.md | 4 +
.../cdc/mysql/MysqlCDCWithSchemaChangeIT.java | 144 +++++++++++++
...sqlcdc_to_mysql_with_replace_schema_change.conf | 59 ++++++
.../transform/replace/ReplaceTransform.java | 2 +
.../transform/replace/ReplaceTransformTest.java | 230 +++++++++++++++++++++
8 files changed, 453 insertions(+), 1 deletion(-)
diff --git a/docs/en/introduction/concepts/incompatible-changes.md
b/docs/en/introduction/concepts/incompatible-changes.md
index 41e06d61f2..62a1967254 100644
--- a/docs/en/introduction/concepts/incompatible-changes.md
+++ b/docs/en/introduction/concepts/incompatible-changes.md
@@ -272,6 +272,11 @@ The Zeta SQL ARRAY function now declares TINYINT elements
as `ARRAY<TINYINT>`, m
(`existingConfigMap`) and restrict the member port (5801) with a
`NetworkPolicy`. Restart the pods
after upgrading so the new configuration is applied.
+### Replace Transform Schema Refresh
+
+- **Behavior change**: Replace resolves `replace_fields` against the refreshed
input schema after schema changes and checkpoint restoration. Previously cached
positions could replace an unrelated column or leave the configured field
unchanged after column positions shifted.
+- **Impact and migration**: If a configured field was dropped or renamed,
schema refresh now fails with the existing missing-field error instead of
processing subsequent rows with stale positions. Update `replace_fields` to
match the source schema before restarting. No option, default, API, or
checkpoint format changes are introduced.
+
### SQL and Calcite Vector Arithmetic
- **Behavior correction**: `VECTOR_NORM`, `INNER_PRODUCT`, `COSINE_DISTANCE`,
diff --git a/docs/en/transforms/replace.md b/docs/en/transforms/replace.md
index aba1c3a193..e50bf746b0 100644
--- a/docs/en/transforms/replace.md
+++ b/docs/en/transforms/replace.md
@@ -24,6 +24,10 @@ The fields you want to replace
For backward compatibility, legacy key `replace_field = "name"` is still
supported, but new configurations should use `replace_fields`.
+Field names are resolved again when the input schema is refreshed, including
checkpoint schema restoration. Adding, removing, or reordering other columns
therefore does not change which fields receive replacements. If a configured
field is absent from the refreshed schema (for example, it was dropped or
renamed), the refresh fails with a missing-field error before subsequent rows
are processed. Update `replace_fields` to match the source schema before
restarting the job.
+
+This field-binding behavior does not change the engine's schema-evolution
support or the existing value-conversion and output-schema behavior of Replace.
+
### pattern [string]
The old string that will be replaced
@@ -122,4 +126,3 @@ sink {
### new version
- Add Replace Transform Connector
-
diff --git a/docs/zh/introduction/concepts/incompatible-changes.md
b/docs/zh/introduction/concepts/incompatible-changes.md
index 7db9438db4..0c5342f45c 100644
--- a/docs/zh/introduction/concepts/incompatible-changes.md
+++ b/docs/zh/introduction/concepts/incompatible-changes.md
@@ -221,6 +221,11 @@ Zeta SQL ARRAY 函数现在将 TINYINT 元素声明为 `ARRAY<TINYINT>`，与实
并使用 `schema_save_mode = "ERROR_WHEN_SCHEMA_NOT_EXIST"` 保留该表结构。不要对手动定义的目标表使用
`RECREATE_SCHEMA`。任意指定索引前缀长度可能拒绝前缀相同但完整值不同的源主键，因此无法保持原有主键语义。
+### Replace 转换的模式刷新
+
+- **行为变更**：模式变更或检查点恢复后，Replace 会根据刷新的输入模式重新解析
`replace_fields`。此前，列位置变化后，缓存的位置可能导致替换无关列或漏掉配置的字段。
+- **影响和迁移**：如果配置的字段已被删除或重命名，模式刷新现在会报出现有的字段不存在错误，而不是使用过期位置继续处理数据行。重新启动作业之前，请更新
`replace_fields`，使其与源表模式一致。此更改不涉及选项、默认值、API 或检查点格式变更。
+
### SQL 和 Calcite 向量运算
- **行为修正**：`VECTOR_NORM`、`INNER_PRODUCT`、`COSINE_DISTANCE`、`L1_DISTANCE`、
diff --git a/docs/zh/transforms/replace.md b/docs/zh/transforms/replace.md
index b1222f04ed..466e4dc35e 100644
--- a/docs/zh/transforms/replace.md
+++ b/docs/zh/transforms/replace.md
@@ -24,6 +24,10 @@
为了向后兼容，旧配置 `replace_field = "name"` 仍然支持，但新配置建议统一使用 `replace_fields`。
+输入模式刷新时（包括从检查点恢复模式），会重新解析配置的字段名。因此，添加、删除或重排其他列不会改变需要执行替换的字段。如果配置的字段在刷新后的模式中不存在（例如已被删除或重命名），刷新会在处理后续数据行之前报出字段不存在错误。重新启动作业之前，请更新
`replace_fields`，使其与源表模式一致。
+
+此字段绑定行为不会改变引擎对模式演变的支持，也不会改变 Replace 现有的值转换和输出模式行为。
+
### pattern [string]
将被替换的旧字符串
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java
index ad1fd293ee..fdf60ce136 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java
@@ -94,6 +94,9 @@ public class MysqlCDCWithSchemaChangeIT extends TestSuiteBase
implements TestRes
/** Dedicated sink table used by the event-type filter regression
coverage. */
private static final String SINK_TABLE_FILTER =
"mysql_cdc_e2e_sink_table_schema_change_filter";
+ private static final String REPLACE_SOURCE_TABLE = "replace_schema_source";
+ private static final String REPLACE_SINK_TABLE = "replace_schema_sink";
+
/** Stable projection used after add-column evolution to compare source
and sink rows. */
private static final String STABLE_QUERY =
"select id,name,description,weight from %s.%s order by id";
@@ -264,6 +267,147 @@ public class MysqlCDCWithSchemaChangeIT extends
TestSuiteBase implements TestRes
MYSQL_DATABASE, SOURCE_TABLE, "Updated product catalog with
sports equipment");
}
+ /** Verifies field binding after a live position shift and subsequent
checkpoint restoration. */
+ @Order(5)
+ @TestTemplate
+ public void
testReplaceFieldBindingAfterSchemaChangeAndRestore(TestContainer container)
+ throws Exception {
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("DROP TABLE IF EXISTS shop." +
REPLACE_SOURCE_TABLE);
+ statement.execute("DROP TABLE IF EXISTS shop." +
REPLACE_SINK_TABLE);
+ statement.execute(
+ "CREATE TABLE shop."
+ + REPLACE_SOURCE_TABLE
+ + " (guard_value VARCHAR(64), name VARCHAR(64),
title VARCHAR(64), id INT PRIMARY KEY)");
+ statement.execute(
+ "CREATE TABLE shop."
+ + REPLACE_SINK_TABLE
+ + " LIKE shop."
+ + REPLACE_SOURCE_TABLE);
+ statement.execute(
+ "INSERT INTO shop."
+ + REPLACE_SOURCE_TABLE
+ + " VALUES ('before guard', 'before name', 'before
title', 1)");
+ }
+ String jobId = String.valueOf(JobIdGenerator.newJobId());
+ String config = "/mysqlcdc_to_mysql_with_replace_schema_change.conf";
+ CompletableFuture<?> execution =
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ Assertions.assertEquals(
+ 0, container.executeJob(config,
jobId).getExitCode());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ Throwable testFailure = null;
+ try {
+ assertReplaceRow(execution, "'before prefix'", "after name",
"after title");
+ waitForReplaceIncrementalRead(container, execution);
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ "ALTER TABLE shop."
+ + REPLACE_SOURCE_TABLE
+ + " ADD COLUMN prefix VARCHAR(64) DEFAULT
'before prefix' FIRST");
+ statement.execute(
+ "UPDATE shop."
+ + REPLACE_SOURCE_TABLE
+ + " SET name='before live', title='before
changed' WHERE id=1");
+ }
+ assertReplaceRow(execution, "prefix", "after live", "after
changed");
+ Assertions.assertEquals(0,
container.savepointJob(jobId).getExitCode());
+ execution.get(60, TimeUnit.SECONDS);
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ "UPDATE shop."
+ + REPLACE_SOURCE_TABLE
+ + " SET name='before restored', title='before
recovered' WHERE id=1");
+ }
+ execution =
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ Assertions.assertEquals(
+ 0, container.restoreJob(config,
jobId).getExitCode());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ assertReplaceRow(execution, "prefix", "after restored", "after
recovered");
+ } catch (Exception | AssertionError failure) {
+ testFailure = failure;
+ throw failure;
+ } finally {
+ try {
+ try {
+ container.cancelJob(jobId);
+ } finally {
+ execution.get(60, TimeUnit.SECONDS);
+ }
+ } catch (Exception | AssertionError cleanupFailure) {
+ if (testFailure != null) {
+ testFailure.addSuppressed(cleanupFailure);
+ } else {
+ throw cleanupFailure;
+ }
+ }
+ }
+ }
+
+ private void waitForReplaceIncrementalRead(
+ TestContainer container, CompletableFuture<?> execution) {
+ await().atMost(DEFAULT_TABLE_SYNC_TIMEOUT_MS, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () -> {
+ if (execution.isCompletedExceptionally()) {
+ execution.join();
+ }
+ Assertions.assertTrue(
+
Arrays.stream(container.getServerLogs().split("\\R"))
+ .anyMatch(
+ line ->
+
line.contains(INCREMENTAL_READ_MARKER)
+ &&
line.contains(
+
MYSQL_DATABASE
+
+ "."
+
+ REPLACE_SOURCE_TABLE)),
+ "The Replace source has not started
incremental reading");
+ });
+ }
+
+ private void assertReplaceRow(
+ CompletableFuture<?> execution, String prefixExpression, String
name, String title) {
+ await().atMost(STRUCTURE_AND_DATA_ASSERT_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS)
+ .ignoreExceptionsMatching(
+ error ->
+ error.getCause() instanceof SQLException
+ && ((SQLException)
error.getCause()).getErrorCode() == 1054)
+ .untilAsserted(
+ () -> {
+ if (execution.isCompletedExceptionally()) {
+ execution.join();
+ }
+ Assertions.assertEquals(
+ Arrays.asList(
+ Arrays.asList(
+ "before prefix",
+ "before guard",
+ name,
+ title,
+ 1)),
+ query(
+ "SELECT "
+ + prefixExpression
+ +
",guard_value,name,title,id FROM shop."
+ + REPLACE_SINK_TABLE
+ + " ORDER BY id"));
+ });
+ }
+
@Order(2)
@TestTemplate
public void testMysqlCdcWithSchemaEvolutionCaseExactlyOnce(TestContainer
container) {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_replace_schema_change.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_replace_schema_change.conf
new file mode 100644
index 0000000000..b076c43d38
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_replace_schema_change.conf
@@ -0,0 +1,59 @@
+#
+# 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 = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ MySQL-CDC {
+ plugin_output = "source"
+ server-id = 5852-5853
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["shop.replace_schema_source"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/shop"
+ schema-changes.enabled = true
+ }
+}
+
+transform {
+ Replace {
+ plugin_input = "source"
+ plugin_output = "replaced"
+ replace_fields = ["name", "title"]
+ pattern = "before"
+ replacement = "after"
+ }
+}
+
+sink {
+ jdbc {
+ plugin_input = "replaced"
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/shop"
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "st_user_sink"
+ password = "mysqlpw"
+ generate_sink_sql = true
+ database = shop
+ table = replace_schema_sink
+ primary_keys = ["id"]
+ multi_table_sink_replica = 1
+ }
+}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/replace/ReplaceTransform.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/replace/ReplaceTransform.java
index 0ceb749667..0864d1006a 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/replace/ReplaceTransform.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/replace/ReplaceTransform.java
@@ -119,6 +119,8 @@ public class ReplaceTransform extends
AbstractCatalogSupportMapTransform {
@Override
protected TableSchema transformTableSchema() {
+ // Schema refresh and input handoff must rebuild the physical field
bindings.
+ initializeFieldIndexes();
return inputCatalogTable.getTableSchema();
}
diff --git
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/replace/ReplaceTransformTest.java
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/replace/ReplaceTransformTest.java
index a823c77c9f..a6af50bdb4 100644
---
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/replace/ReplaceTransformTest.java
+++
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/replace/ReplaceTransformTest.java
@@ -26,7 +26,14 @@ import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
import org.apache.seatunnel.api.table.catalog.TableIdentifier;
import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.RestoreTableSchemaEvent;
+import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.RowKind;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
import org.apache.seatunnel.transform.exception.TransformException;
@@ -37,6 +44,7 @@ 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.Map;
@@ -44,6 +52,228 @@ class ReplaceTransformTest {
private static CatalogTable catalogTable;
+ @Test
+ void testRestoreSchemaRebindsFieldsBeforeNextRow() {
+ ReplaceTransform transform = replacementTransform("name", "title");
+ transform.getProducedCatalogTable();
+ CatalogTable restored = reorderedTable("title", "id", "name");
+ RestoreTableSchemaEvent event = new RestoreTableSchemaEvent(restored);
+
+ Assertions.assertSame(event, transform.mapSchemaChangeEvent(event));
+ Assertions.assertArrayEquals(
+ new Object[] {"after title", 1, "after name"},
+ transform
+ .map(new SeaTunnelRow(new Object[] {"before title", 1,
"before name"}))
+ .getFields());
+ Assertions.assertEquals(restored.getTableSchema(),
event.getChangeAfter().getTableSchema());
+ }
+
+ @Test
+ void testInputSchemaHandoffRebindsFieldsBeforeNextRow() {
+ ReplaceTransform transform = replacementTransform("name", "title");
+ transform.getProducedCatalogTable();
+ CatalogTable reordered = reorderedTable("title", "name", "id");
+ transform.setInputCatalogTable(reordered);
+
+ Assertions.assertArrayEquals(
+ new Object[] {"after title", "after name", 1},
+ transform
+ .map(new SeaTunnelRow(new Object[] {"before title",
"before name", 1}))
+ .getFields());
+ Assertions.assertEquals(
+ reordered.getTableSchema(),
transform.getProducedCatalogTable().getTableSchema());
+ }
+
+ private static ReplaceTransform replacementTransform(String... fields) {
+ return new ReplaceTransform(replacementConfig(fields), catalogTable);
+ }
+
+ private static ReadonlyConfig replacementConfig(String... fields) {
+ Map<String, Object> config = new HashMap<>();
+ config.put("replace_fields", Arrays.asList(fields));
+ config.put("pattern", "before");
+ config.put("replacement", "after");
+ return ReadonlyConfig.fromMap(config);
+ }
+
+ private static CatalogTable reorderedTable(String... fields) {
+ TableSchema.Builder schema = TableSchema.builder();
+ for (String field : fields) {
+ schema.column(catalogTable.getTableSchema().getColumn(field));
+ }
+ return CatalogTable.of(
+ catalogTable.getTableId(),
+ schema.build(),
+ catalogTable.getOptions(),
+ catalogTable.getPartitionKeys(),
+ catalogTable.getComment());
+ }
+
+ @Test
+ void testAddDropAndReorderRebindFields() {
+ ReplaceTransform transform = replacementTransform("name", "title");
+ transform.mapSchemaChangeEvent(
+ AlterTableAddColumnEvent.addFirst(
+ catalogTable.getTableId(),
+ PhysicalColumn.of("prefix", BasicType.STRING_TYPE,
10L, true, null, null)));
+ Assertions.assertArrayEquals(
+ new Object[] {"before prefix", 1, "after name", "after title"},
+ transform
+ .map(
+ new SeaTunnelRow(
+ new Object[] {
+ "before prefix", 1, "before name",
"before title"
+ }))
+ .getFields());
+
+ transform.mapSchemaChangeEvent(
+ new AlterTableDropColumnEvent(catalogTable.getTableId(),
"id"));
+ Assertions.assertArrayEquals(
+ new Object[] {"before prefix", "after name", "after title"},
+ transform
+ .map(
+ new SeaTunnelRow(
+ new Object[] {
+ "before prefix", "before name",
"before title"
+ }))
+ .getFields());
+
+ transform.mapSchemaChangeEvent(
+ AlterTableModifyColumnEvent.modifyFirst(
+ catalogTable.getTableId(),
+ catalogTable.getTableSchema().getColumn("title")));
+ Assertions.assertArrayEquals(
+ new Object[] {"after title", "before prefix", "after name"},
+ transform
+ .map(
+ new SeaTunnelRow(
+ new Object[] {
+ "before title", "before prefix",
"before name"
+ }))
+ .getFields());
+ }
+
+ @Test
+ void testMissingTargetFailsDuringSchemaRefresh() {
+ SchemaChangeEvent[] events = {
+ new RestoreTableSchemaEvent(reorderedTable("id", "title")),
+ new AlterTableDropColumnEvent(catalogTable.getTableId(), "name"),
+ AlterTableChangeColumnEvent.change(
+ catalogTable.getTableId(),
+ "name",
+ PhysicalColumn.of("renamed", BasicType.STRING_TYPE, 10L,
true, null, null))
+ };
+ for (SchemaChangeEvent event : events) {
+ ReplaceTransform transform = replacementTransform("title", "name");
+ TransformException error =
+ Assertions.assertThrows(
+ TransformException.class, () ->
transform.mapSchemaChangeEvent(event));
+ Assertions.assertTrue(error.getMessage().contains("name"));
+ }
+ ReplaceTransform transform = replacementTransform("name");
+ Assertions.assertThrows(
+ TransformException.class,
+ () -> transform.setInputCatalogTable(reorderedTable("id",
"title")));
+ }
+
+ @Test
+ void testRestorePreservesRowMetadataNullsAndInput() {
+ ReplaceTransform transform = replacementTransform("name", "title");
+ transform.mapSchemaChangeEvent(
+ new RestoreTableSchemaEvent(reorderedTable("title", "id",
"name")));
+ SeaTunnelRow input = new SeaTunnelRow(new Object[] {null, 1, "before
name"});
+ input.setRowKind(RowKind.UPDATE_AFTER);
+ input.setTableId(catalogTable.getTableId().toTablePath().toString());
+ input.getOptions().put("sequence", 123L);
+ SeaTunnelRow output = transform.map(input);
+ Assertions.assertArrayEquals(new Object[] {null, 1, "after name"},
output.getFields());
+ Assertions.assertArrayEquals(new Object[] {null, 1, "before name"},
input.getFields());
+ Assertions.assertEquals(input.getRowKind(), output.getRowKind());
+ Assertions.assertEquals(input.getTableId(), output.getTableId());
+ Assertions.assertEquals(input.getOptions(), output.getOptions());
+ }
+
+ @Test
+ void testTypeChangeKeepsExistingValueConversion() {
+ ReplaceTransform transform = replacementTransform("name");
+ transform.mapSchemaChangeEvent(
+ AlterTableModifyColumnEvent.modifyFirst(
+ catalogTable.getTableId(),
+ PhysicalColumn.of("name", BasicType.INT_TYPE, 10L,
true, null, null)));
+ Assertions.assertArrayEquals(
+ new Object[] {"123", 1, "before title"},
+ transform.map(new SeaTunnelRow(new Object[] {123, 1, "before
title"})).getFields());
+ Assertions.assertEquals(
+ BasicType.INT_TYPE,
+
transform.getProducedCatalogTable().getSeaTunnelRowType().getFieldType(0));
+ }
+
+ @Test
+ void testMultiTableRestoreOnlyRebindsAffectedTable() {
+ CatalogTable second =
+ CatalogTable.of(
+ TableIdentifier.of("catalog", TablePath.of("other")),
+ catalogTable.getTableSchema(),
+ new HashMap<>(),
+ new ArrayList<>(),
+ "other");
+ ReplaceMultiCatalogTransform transform =
+ new ReplaceMultiCatalogTransform(
+ Arrays.asList(catalogTable, second),
replacementConfig("name", "title"));
+ CatalogTable restored = reorderedTable("title", "id", "name");
+ transform.mapSchemaChangeEvent(new RestoreTableSchemaEvent(restored));
+
+ SeaTunnelRow firstInput = new SeaTunnelRow(new Object[] {"before
title", 1, "before name"});
+
firstInput.setTableId(catalogTable.getTableId().toTablePath().toString());
+ Assertions.assertArrayEquals(
+ new Object[] {"after title", 1, "after name"},
+ transform.map(firstInput).getFields());
+ SeaTunnelRow secondInput =
+ new SeaTunnelRow(new Object[] {2, "before other", "before
title"});
+ secondInput.setTableId(second.getTableId().toTablePath().toString());
+ Assertions.assertArrayEquals(
+ new Object[] {2, "after other", "after title"},
+ transform.map(secondInput).getFields());
+ Assertions.assertEquals(
+ restored.getTableSchema(),
+ transform.getProducedCatalogTables().get(0).getTableSchema());
+ Assertions.assertEquals(
+ second.getTableSchema(),
+ transform.getProducedCatalogTables().get(1).getTableSchema());
+ }
+
+ @Test
+ void testAllMatchChainRebindsOnRestoreAndInputHandoff() {
+ Map<String, Object> firstRule = new HashMap<>();
+ firstRule.put("table_path",
catalogTable.getTableId().toTablePath().toString());
+ firstRule.put("replace_fields", Arrays.asList("name", "title"));
+ firstRule.put("pattern", "before");
+ firstRule.put("replacement", "middle");
+ Map<String, Object> secondRule = new HashMap<>(firstRule);
+ secondRule.put("pattern", "middle");
+ secondRule.put("replacement", "after");
+ Map<String, Object> config = new HashMap<>();
+ config.put("rule_match_mode", "ALL_MATCH");
+ config.put("table_transform", Arrays.asList(firstRule, secondRule));
+ ReplaceMultiCatalogTransform transform =
+ new ReplaceMultiCatalogTransform(
+ Collections.singletonList(catalogTable),
ReadonlyConfig.fromMap(config));
+ transform.mapSchemaChangeEvent(
+ new RestoreTableSchemaEvent(reorderedTable("title", "id",
"name")));
+ Assertions.assertArrayEquals(
+ new Object[] {"after title", 1, "after name"},
+ transform
+ .map(new SeaTunnelRow(new Object[] {"before title", 1,
"before name"}))
+ .getFields());
+ transform.setInputCatalogTables(
+ Collections.singletonList(reorderedTable("name", "title",
"id")));
+ Assertions.assertArrayEquals(
+ new Object[] {"after name", "after title", 1},
+ transform
+ .map(new SeaTunnelRow(new Object[] {"before name",
"before title", 1}))
+ .getFields());
+ }
+
@BeforeAll
static void setUp() {
catalogTable =
