[Feature][Transform][Replace]Replace support schema change

docs/en/concept/schema-evolution.md

@@ -26,6 +26,11 @@ Schema Evolution means that the schema of a data table can be changed and the da
 [Doris](https://github.com/apache/seatunnel/blob/dev/docs/en/connector-v2/sink/Doris.md)
 [Paimon](https://github.com/apache/seatunnel/blob/dev/docs/en/connector-v2/sink/Paimon.md#Schema-Evolution)
 
+## Supported transforms
+
+### Transform
+[Replace](https://github.com/apache/seatunnel/blob/dev/docs/en/transform-v2/replace.md)
+
 Note: The schema evolution is not support the transform at now. The schema evolution of different types of databases（Oracle-CDC -> Jdbc-Mysql）is currently not supported the default value of the column in ddl.
 
 When you use the Oracle-CDC，you can not use the username named `SYS` or `SYSTEM` to modify the table schema, otherwise the ddl event will be filtered out which can lead to the schema evolution not working.

docs/zh/concept/schema-evolution.md

@@ -26,6 +26,11 @@
 [Doris](https://github.com/apache/seatunnel/blob/dev/docs/zh/connector-v2/sink/Doris.md)
 [Paimon](https://github.com/apache/seatunnel/blob/dev/docs/zh/connector-v2/sink/Paimon.md#模式演变)
 
+## 已支持的转换
+
+### 转换
+[Replace](https://github.com/apache/seatunnel/blob/dev/docs/zh/transform-v2/replace.md)
+
 注意: 目前模式演进不支持transform。不同类型数据库(Oracle-CDC -> Jdbc-Mysql)的模式演进目前不支持ddl中列的默认值。
 
 当你使用Oracle-CDC时，你不能使用用户名`SYS`或`SYSTEM`来修改表结构，否则ddl事件将被过滤，这可能导致模式演进不起作用；

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/replace/ReplaceMultiCatalogTransform.java

@@ -19,7 +19,9 @@
 
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 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 org.apache.seatunnel.transform.common.AbstractMultiCatalogMapTransform;
 
@@ -37,6 +39,13 @@ public String getPluginName() {
         return "Replace";
     }
 
+    @Override
+    public SchemaChangeEvent mapSchemaChangeEvent(SchemaChangeEvent event) {
+        return ((SeaTunnelMapTransform<SeaTunnelRow>)
+                        transformMap.get(event.tablePath().toString()))
+                .mapSchemaChangeEvent(event);
+    }
+
     @Override
     protected SeaTunnelTransform<SeaTunnelRow> buildTransform(
             CatalogTable inputCatalogTable, ReadonlyConfig config) {

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/replace/ReplaceTransform.java

@@ -17,9 +17,15 @@
 
 package org.apache.seatunnel.transform.replace;
 
+import org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTesting;
+
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.catalog.Column;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
+import org.apache.seatunnel.api.table.schema.handler.TableSchemaChangeEventDispatcher;
+import org.apache.seatunnel.api.table.schema.handler.TableSchemaChangeEventHandler;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowAccessor;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.transform.common.SingleFieldOutputTransform;
@@ -36,13 +42,16 @@ public class ReplaceTransform extends SingleFieldOutputTransform {
     private final ReadonlyConfig config;
     private int inputFieldIndex;
 
+    private TableSchemaChangeEventHandler tableSchemaChangeEventHandler;
+
     public ReplaceTransform(
             @NonNull ReadonlyConfig config, @NonNull CatalogTable inputCatalogTable) {
         super(inputCatalogTable);
         this.config = config;
         initOutputFields(
                 inputCatalogTable.getTableSchema().toPhysicalRowDataType(),
                 this.config.get(ReplaceTransformConfig.KEY_REPLACE_FIELD));
+        this.tableSchemaChangeEventHandler = new TableSchemaChangeEventDispatcher();
     }
 
     @Override
@@ -58,6 +67,28 @@ private void initOutputFields(SeaTunnelRowType inputRowType, String replaceField
         }
     }
 
+    @Override
+    public SchemaChangeEvent mapSchemaChangeEvent(SchemaChangeEvent event) {
+        TableSchema newTableSchema =
+                tableSchemaChangeEventHandler
+                        .reset(this.inputCatalogTable.getTableSchema())
+                        .apply(event);
+        this.inputCatalogTable =
+                CatalogTable.of(
+                        inputCatalogTable.getTableId(),
+                        newTableSchema,
+                        inputCatalogTable.getOptions(),
+                        inputCatalogTable.getPartitionKeys(),
+                        inputCatalogTable.getComment());
+        initOutputFields(
+                inputCatalogTable.getTableSchema().toPhysicalRowDataType(),
+                this.config.get(ReplaceTransformConfig.KEY_REPLACE_FIELD));
+
+        transformTableSchema();
+
+        return event;
+    }
+
     @Override
     protected Object getOutputFieldValue(SeaTunnelRowAccessor inputRow) {
         Object inputFieldValue = inputRow.getField(inputFieldIndex);
@@ -108,4 +139,9 @@ protected Column getOutputColumn() {
         }
         return collect.get(0).copy();
     }
+
+    @VisibleForTesting
+    public void initRowContainerGenerator() {
+        transformTableSchema();
+    }
 }

seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/replace/ReplaceTransformTest.java

@@ -0,0 +1,129 @@
+/*
+ * 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.replace;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.ConstraintKey;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.PrimaryKey;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+
+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.Map;
+
+public class ReplaceTransformTest {
+
+    private static final CatalogTable DEFAULT_TABLE =
+            CatalogTable.of(
+                    TableIdentifier.of("test", "Database-x", "Schema-x", "Table-x"),
+                    TableSchema.builder()
+                            .column(
+                                    PhysicalColumn.of(
+                                            "f1",
+                                            BasicType.LONG_TYPE,
+                                            null,
+                                            null,
+                                            false,
+                                            null,
+                                            null))
+                            .column(
+                                    PhysicalColumn.of(
+                                            "f2",
+                                            BasicType.STRING_TYPE,
+                                            null,
+                                            null,
+                                            true,
+                                            null,
+                                            null))
+                            .column(
+                                    PhysicalColumn.of(
+                                            "f3",
+                                            BasicType.LONG_TYPE,
+                                            null,
+                                            null,
+                                            true,
+                                            null,
+                                            null))
+                            .primaryKey(PrimaryKey.of("pk1", Arrays.asList("f1")))
+                            .constraintKey(
+                                    ConstraintKey.of(
+                                            ConstraintKey.ConstraintType.UNIQUE_KEY,
+                                            "uk1",
+                                            Arrays.asList(
+                                                    ConstraintKey.ConstraintKeyColumn.of(
+                                                            "f2", ConstraintKey.ColumnSortType.ASC),
+                                                    ConstraintKey.ConstraintKeyColumn.of(
+                                                            "f3",
+                                                            ConstraintKey.ColumnSortType.ASC))))
+                            .build(),
+                    Collections.emptyMap(),
+                    Collections.singletonList("f2"),
+                    null);
+
+    @Test
+    public void testReplaceTransformSchemaChange() {
+
+        // before schema change
+        SeaTunnelRow inputRow = new SeaTunnelRow(new Object[] {1L, "old string", 1L});
+        inputRow.setTableId(DEFAULT_TABLE.getTablePath().getFullName());
+
+        Map map = new HashMap<>();
+        map.put("replace_field", "f2");
+        map.put("pattern", "old string");
+        map.put("replacement", "new string");
+
+        ReadonlyConfig readonlyConfig = ReadonlyConfig.fromMap(map);
+        ReplaceTransform replaceTransform = new ReplaceTransform(readonlyConfig, DEFAULT_TABLE);
+
+        replaceTransform.initRowContainerGenerator();
+
+        Assertions.assertEquals(1, replaceTransform.getFieldIndex());
+        SeaTunnelRow outputRow = replaceTransform.transform(inputRow);
+
+        Assertions.assertEquals(
+                "SeaTunnelRow{tableId=Database-x.Schema-x.Table-x, kind=+I, fields=[1, new string, 1]}",
+                outputRow.toString());
+
+        // after schema change
+        AlterTableAddColumnEvent addColumnEvent =
+                AlterTableAddColumnEvent.addAfter(
+                        DEFAULT_TABLE.getTableId(),
+                        PhysicalColumn.of("f4", BasicType.LONG_TYPE, null, null, true, null, null),
+                        "f1");
+
+        replaceTransform.mapSchemaChangeEvent(addColumnEvent);
+
+        SeaTunnelRow inputRowTwo = new SeaTunnelRow(new Object[] {2L, 2L, "old string", 2L});
+        inputRowTwo.setTableId(DEFAULT_TABLE.getTablePath().getFullName());
+        SeaTunnelRow outputRowTwo = replaceTransform.transform(inputRowTwo);
+
+        Assertions.assertEquals(
+                "SeaTunnelRow{tableId=Database-x.Schema-x.Table-x, kind=+I, fields=[2, 2, new string, 2]}",
+                outputRowTwo.toString());
+    }
+}