chore(connector-file,transforms-v2): adapt to upstream/dev API changes after rebase

Mechanical fixups so the rebased PR branch compiles and all schema-evolution tests pass
on top of latest upstream/dev. No behaviour change in the schema-evolution feature itself.

Adapted to upstream API changes:
- CanalJsonSerializationSchema / DebeziumJsonSerializationSchema /
  MaxWellJsonSerializationSchema constructors now require a third arg (mergeUpdateEventFlag)
  when called with a Charset. onSchemaChanged() in each strategy now passes the field that
  was already stored in the writer.
- FileSinkConfig constructor signature changed to take ReadonlyConfig (was Config).
  Test sites in FileSchemaEvolutionTest, ParquetWriteStrategyEvolutionTest, and
  ParquetTypeCoercionTest wrapped with ReadonlyConfig.fromConfig(...). Imports added.
- FieldFieldMultiCatalogTransform was renamed to FilterFieldMultiCatalogTransform upstream.
  ChainTimestampPreservationTest and ProductionPipelineSchemaChangeTest updated accordingly.

Test contract updates following the partition_by guard removal in this PR:
- testPartitionByWithSchemaEvolutionEnabledThrowsAtConfig replaced with
  testPartitionByWithSchemaEvolutionEnabledIsAccepted — partition_by + schema_evolution
  is now supported (writer-local partitionFieldsIndexInRow rebuilds on every ALTER via
  name-based lookup; drop/rename of partition column is the only case rejected, with an
  explicit IllegalStateException at rebuild time).

Restored applySchemaChange's silent no-op behavior when schema_evolution_enabled=false.
The fail-fast variant is left as an open design question for maintainers — current
behavior preserves backward compatibility but leaves a latent ClassCastException trap
when schema-changes.enabled=true at source + schema_evolution_enabled=false at sink
(documented inline). Awaiting maintainer guidance on the preferred UX.

All 30 schema-evolution tests pass:
- FileSchemaEvolutionTest (19)
- ParquetWriteStrategyEvolutionTest (2), ParquetTypeCoercionTest (3)
- ChainTimestampPreservationTest, MetadataMultiCatalogSchemaChangeTest,
  ProductionPipelineSchemaChangeTest, TransformChainLiveAlterTest,
  SQLMultiCatalogSchemaChangeTest

Author: ricky2129

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/AbstractWriteStrategy.java

@@ -40,7 +40,6 @@
 import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSinkOptions;
 import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
 import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
-import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorErrorCode;
 import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
 import org.apache.seatunnel.connectors.seatunnel.file.hadoop.HadoopFileSystemProxy;
 import org.apache.seatunnel.connectors.seatunnel.file.sink.commit.FileCommitInfo;
@@ -193,22 +192,12 @@ public void setCatalogTable(CatalogTable catalogTable) {
     @Override
     public void applySchemaChange(SchemaChangeEvent event) throws IOException {
         if (!fileSinkConfig.isSchemaEvolutionEnabled()) {
-            // Fail-fast guard: if a real schema change arrives but the sink is configured to
-            // ignore them, silently swallowing the event leaves sinkColumnsIndexInRow stale.
-            // The next data row will read row[idx] with stale catalog assumptions and produce a
-            // confusing ClassCastException several rows later. Throw immediately with an
-            // actionable message instead — it's a config mismatch, not a runtime fault.
-            throw new FileConnectorException(
-                    FileConnectorErrorCode.FORMAT_NOT_SUPPORT,
-                    "Received schema-change event ["
-                            + event.getClass().getSimpleName()
-                            + "] but schema_evolution_enabled=false on the file sink. "
-                            + "This combination is unsafe — the sink would keep writing with the "
-                            + "pre-ALTER schema and corrupt subsequent rows. "
-                            + "Resolve by either: "
-                            + "(a) setting schema_evolution_enabled=true on the sink, OR "
-                            + "(b) setting schema-changes.enabled=false on the CDC source so "
-                            + "ALTER events are not emitted in the first place.");
+            // No-op when schema evolution is disabled at the sink. NOTE: if the upstream CDC
+            // source has schema-changes.enabled=true, this leaves sinkColumnsIndexInRow stale and
+            // the next data row may produce a ClassCastException several rows later. Open
+            // discussion with maintainers on whether this should fail-fast instead — the silent
+            // return is the historical behavior preserved here for backwards compatibility.
+            return;
         }
         log.info(
                 "[FileSchemaEvolution] applying {} — before: rowType={}, sinkColumns={}, indices={}",

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/CanalJsonWriteStrategy.java

@@ -71,7 +71,9 @@ public void setCatalogTable(CatalogTable catalogTable) {
     protected void onSchemaChanged() {
         this.serializationSchema =
                 new CanalJsonSerializationSchema(
-                        buildSchemaWithRowType(seaTunnelRowType, sinkColumnsIndexInRow), charset);
+                        buildSchemaWithRowType(seaTunnelRowType, sinkColumnsIndexInRow),
+                        charset,
+                        mergeUpdateEventFlag);
     }
 
     @Override

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/DebeziumJsonWriteStrategy.java

@@ -71,7 +71,9 @@ public void setCatalogTable(CatalogTable catalogTable) {
     protected void onSchemaChanged() {
         this.serializationSchema =
                 new DebeziumJsonSerializationSchema(
-                        buildSchemaWithRowType(seaTunnelRowType, sinkColumnsIndexInRow), charset);
+                        buildSchemaWithRowType(seaTunnelRowType, sinkColumnsIndexInRow),
+                        charset,
+                        mergeUpdateEventFlag);
     }
 
     @Override

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/MaxWellJsonWriteStrategy.java

@@ -71,7 +71,9 @@ public void setCatalogTable(CatalogTable catalogTable) {
     protected void onSchemaChanged() {
         this.serializationSchema =
                 new MaxWellJsonSerializationSchema(
-                        buildSchemaWithRowType(seaTunnelRowType, sinkColumnsIndexInRow), charset);
+                        buildSchemaWithRowType(seaTunnelRowType, sinkColumnsIndexInRow),
+                        charset,
+                        mergeUpdateEventFlag);
     }
 
     @Override

seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/writer/FileSchemaEvolutionTest.java

@@ -488,7 +488,8 @@ private static NoOpWriteStrategy buildTypeZooStrategy() {
                         "path = \"/tmp/test\"\n"
                                 + "file_format_type = \"parquet\"\n"
                                 + "schema_evolution_enabled = true");
-        FileSinkConfig sinkConfig = new FileSinkConfig(config, TYPE_ZOO_ROW_TYPE);
+        FileSinkConfig sinkConfig =
+                new FileSinkConfig(ReadonlyConfig.fromConfig(config), TYPE_ZOO_ROW_TYPE);
         NoOpWriteStrategy strategy = new NoOpWriteStrategy(sinkConfig);
         strategy.setCatalogTable(
                 org.apache.seatunnel.api.table.catalog.CatalogTableUtil.getCatalogTable(
@@ -654,22 +655,25 @@ public void testBinaryWithSchemaEvolutionEnabledThrowsAtConfig() {
                                 + "schema_evolution_enabled = true");
         Assertions.assertThrows(
                 FileConnectorException.class,
-                () -> new FileSinkConfig(config, BASE_ROW_TYPE),
+                () -> new FileSinkConfig(ReadonlyConfig.fromConfig(config), BASE_ROW_TYPE),
                 "binary format must reject schema_evolution_enabled=true at config time");
     }
 
     @Test
-    public void testPartitionByWithSchemaEvolutionEnabledThrowsAtConfig() {
+    public void testPartitionByWithSchemaEvolutionEnabledIsAccepted() {
+        // partition_by + schema_evolution_enabled is supported: applySchemaChange rebuilds
+        // partitionFieldsIndexInRow on every ALTER via name-based lookup against the post-ALTER
+        // row type, mirroring sinkColumnsIndexInRow. Drop/rename of a partition column itself is
+        // rejected at rebuild time with an explicit IllegalStateException (covered separately).
         Config config =
                 ConfigFactory.parseString(
                         "path = \"/tmp/test\"\n"
                                 + "file_format_type = \"text\"\n"
                                 + "partition_by = [\"age\"]\n"
                                 + "schema_evolution_enabled = true");
-        Assertions.assertThrows(
-                FileConnectorException.class,
-                () -> new FileSinkConfig(config, BASE_ROW_TYPE),
-                "partition_by must reject schema_evolution_enabled=true at config time");
+        Assertions.assertDoesNotThrow(
+                () -> new FileSinkConfig(ReadonlyConfig.fromConfig(config), BASE_ROW_TYPE),
+                "partition_by + schema_evolution_enabled=true must be supported");
     }
 
     /**

seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/writer/ParquetTypeCoercionTest.java

@@ -19,6 +19,7 @@
 
 import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory;
 
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.source.Collector;
 import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
 import org.apache.seatunnel.api.table.type.BasicType;
@@ -214,7 +215,8 @@ private static List<Object[]> writeAndReadBack(
         writeConfig.put("file_format_type", FileFormat.PARQUET.name());
 
         FileSinkConfig sinkConfig =
-                new FileSinkConfig(ConfigFactory.parseMap(writeConfig), rowType);
+                new FileSinkConfig(
+                        ReadonlyConfig.fromConfig(ConfigFactory.parseMap(writeConfig)), rowType);
         ParquetWriteStrategy strategy = new ParquetWriteStrategy(sinkConfig);
         ParquetReadStrategyTest.LocalConf hadoopConf =
                 new ParquetReadStrategyTest.LocalConf(FS_DEFAULT_NAME_DEFAULT);

seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/writer/ParquetWriteStrategyEvolutionTest.java

@@ -19,6 +19,7 @@
 
 import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory;
 
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.source.Collector;
 import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
 import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
@@ -86,7 +87,9 @@ public void testAddColumnRotatesFileAndPreservesAllRows() throws Exception {
         writeConfig.put("schema_evolution_enabled", true);
 
         FileSinkConfig sinkConfig =
-                new FileSinkConfig(ConfigFactory.parseMap(writeConfig), baseRowType);
+                new FileSinkConfig(
+                        ReadonlyConfig.fromConfig(ConfigFactory.parseMap(writeConfig)),
+                        baseRowType);
         ParquetWriteStrategy strategy = new ParquetWriteStrategy(sinkConfig);
         ParquetReadStrategyTest.LocalConf hadoopConf =
                 new ParquetReadStrategyTest.LocalConf(FS_DEFAULT_NAME_DEFAULT);
@@ -187,7 +190,9 @@ public void testDropColumnRotatesFile() throws Exception {
         writeConfig.put("schema_evolution_enabled", true);
 
         FileSinkConfig sinkConfig =
-                new FileSinkConfig(ConfigFactory.parseMap(writeConfig), baseRowType);
+                new FileSinkConfig(
+                        ReadonlyConfig.fromConfig(ConfigFactory.parseMap(writeConfig)),
+                        baseRowType);
         ParquetWriteStrategy strategy = new ParquetWriteStrategy(sinkConfig);
         ParquetReadStrategyTest.LocalConf hadoopConf =
                 new ParquetReadStrategyTest.LocalConf(FS_DEFAULT_NAME_DEFAULT);

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogTransform.java

@@ -19,8 +19,6 @@
 
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
-import org.apache.seatunnel.api.table.catalog.TableIdentifier;
-import org.apache.seatunnel.api.table.catalog.TableSchema;
 import org.apache.seatunnel.api.table.schema.event.AlterTableEvent;
 import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;

seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/metadata/ChainTimestampPreservationTest.java

@@ -34,7 +34,7 @@
 import org.apache.seatunnel.api.table.type.LocalTimeType;
 import org.apache.seatunnel.api.table.type.MetadataUtil;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
-import org.apache.seatunnel.transform.filter.FieldFieldMultiCatalogTransform;
+import org.apache.seatunnel.transform.filter.FilterFieldMultiCatalogTransform;
 import org.apache.seatunnel.transform.filterrowkind.FieldRowKindMultiCatalogTransform;
 import org.apache.seatunnel.transform.rowkind.RowKindExtractorMultiCatalogTransform;
 import org.apache.seatunnel.transform.sql.SQLMultiCatalogFlatMapTransform;
@@ -162,8 +162,8 @@ void chainPreservesLocalDateTimeAtTimestampColumn() {
         CatalogTable sqlOut = sql.getProducedCatalogTables().get(0);
         Map<String, Object> filterFieldCfg = new HashMap<>();
         filterFieldCfg.put("exclude_fields", Arrays.asList("c_delay"));
-        FieldFieldMultiCatalogTransform filterField =
-                new FieldFieldMultiCatalogTransform(
+        FilterFieldMultiCatalogTransform filterField =
+                new FilterFieldMultiCatalogTransform(
                         Collections.singletonList(sqlOut), ReadonlyConfig.fromMap(filterFieldCfg));
 
         List<org.apache.seatunnel.api.transform.SeaTunnelTransform<SeaTunnelRow>> chain =

seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/metadata/ProductionPipelineSchemaChangeTest.java

@@ -33,7 +33,7 @@
 import org.apache.seatunnel.api.table.type.CommonOptions;
 import org.apache.seatunnel.api.table.type.MetadataUtil;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
-import org.apache.seatunnel.transform.filter.FieldFieldMultiCatalogTransform;
+import org.apache.seatunnel.transform.filter.FilterFieldMultiCatalogTransform;
 import org.apache.seatunnel.transform.filterrowkind.FieldRowKindMultiCatalogTransform;
 import org.apache.seatunnel.transform.rowkind.RowKindExtractorMultiCatalogTransform;
 import org.apache.seatunnel.transform.sql.SQLMultiCatalogFlatMapTransform;
@@ -148,12 +148,12 @@ void productionWrapperPipelinePreservesPostAlterValues() {
                 new SQLMultiCatalogFlatMapTransform(
                         Collections.singletonList(rkeOut), ReadonlyConfig.fromMap(sqlCfg));
 
-        // 5. FieldFieldMultiCatalogTransform — exclude c_delay
+        // 5. FilterFieldMultiCatalogTransform — exclude c_delay
         CatalogTable sqlOut = sql.getProducedCatalogTables().get(0);
         Map<String, Object> filterFieldCfg = new HashMap<>();
         filterFieldCfg.put("exclude_fields", Arrays.asList("c_delay"));
-        FieldFieldMultiCatalogTransform filterField =
-                new FieldFieldMultiCatalogTransform(
+        FilterFieldMultiCatalogTransform filterField =
+                new FilterFieldMultiCatalogTransform(
                         Collections.singletonList(sqlOut), ReadonlyConfig.fromMap(filterFieldCfg));
 
         // List of all wrappers in chain order — engine iterates in this order during ALTER.