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-12365-d2fc207903516c23f98f52f073e28c1cd63fe2f0 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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 =
