[Improve][Connector-V2] Support schema evolution for mysql-cdc and mysql-jdbc (#6929)

Author: dailai

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/Column.java

@@ -191,4 +191,7 @@ protected Column(
 
     /** Returns a copy of the column with a replaced name. */
     public abstract Column rename(String newColumnName);
+
+    /** Returns a copy of the column with a replaced sourceType. */
+    public abstract Column reSourceType(String sourceType);
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/MetadataColumn.java

@@ -83,4 +83,9 @@ public Column rename(String newColumnName) {
                 defaultValue,
                 comment);
     }
+
+    @Override
+    public Column reSourceType(String sourceType) {
+        throw new UnsupportedOperationException("Not implemented");
+    }
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/PhysicalColumn.java

@@ -325,4 +325,22 @@ public Column rename(String newColumnName) {
                 bitLen,
                 longColumnLength);
     }
+
+    @Override
+    public Column reSourceType(String newSourceType) {
+        return new PhysicalColumn(
+                name,
+                dataType,
+                columnLength,
+                scale,
+                nullable,
+                defaultValue,
+                comment,
+                newSourceType,
+                options,
+                isUnsigned,
+                isZeroFill,
+                bitLen,
+                longColumnLength);
+    }
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/event/AlterTableColumnEvent.java

@@ -23,6 +23,7 @@
 
 @ToString(callSuper = true)
 public abstract class AlterTableColumnEvent extends AlterTableEvent {
+
     public AlterTableColumnEvent(TableIdentifier tableIdentifier) {
         super(tableIdentifier);
     }

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/event/TableEvent.java

@@ -33,6 +33,7 @@ public abstract class TableEvent implements SchemaChangeEvent {
     protected final TableIdentifier tableIdentifier;
     @Getter @Setter private String jobId;
     @Getter @Setter private String statement;
+    @Getter @Setter protected String sourceDialectName;
 
     @Override
     public TableIdentifier tableIdentifier() {

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/schema/AbstractSchemaChangeResolver.java

@@ -0,0 +1,58 @@
+/*
+ * 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.connectors.cdc.base.schema;
+
+import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
+import org.apache.seatunnel.connectors.cdc.base.utils.SourceRecordUtils;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+import com.google.common.collect.Lists;
+import io.debezium.relational.history.HistoryRecord;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.List;
+
+@Slf4j
+public abstract class AbstractSchemaChangeResolver implements SchemaChangeResolver {
+
+    protected static final List<String> SUPPORT_DDL = Lists.newArrayList("ALTER TABLE");
+
+    protected JdbcSourceConfig jdbcSourceConfig;
+
+    public AbstractSchemaChangeResolver(JdbcSourceConfig jdbcSourceConfig) {
+        this.jdbcSourceConfig = jdbcSourceConfig;
+    }
+
+    @Override
+    public boolean support(SourceRecord record) {
+        String ddl = SourceRecordUtils.getDdl(record);
+        Struct value = (Struct) record.value();
+        List<Struct> tableChanges = value.getArray(HistoryRecord.Fields.TABLE_CHANGES);
+        if (tableChanges == null || tableChanges.isEmpty()) {
+            log.warn("Ignoring statement for non-captured table {}", ddl);
+            return false;
+        }
+        return StringUtils.isNotBlank(ddl)
+                && SUPPORT_DDL.stream()
+                        .map(String::toUpperCase)
+                        .anyMatch(prefix -> ddl.toUpperCase().contains(prefix));
+    }
+}

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/utils/SourceRecordUtils.java

@@ -28,6 +28,7 @@
 import io.debezium.data.Envelope;
 import io.debezium.document.DocumentReader;
 import io.debezium.relational.TableId;
+import io.debezium.relational.history.HistoryRecord;
 import io.debezium.util.SchemaNameAdjuster;
 
 import java.math.BigDecimal;
@@ -214,4 +215,9 @@ public static TablePath getTablePath(SourceRecord record) {
         }
         return TablePath.of(databaseName, schemaName, tableName);
     }
+
+    public static String getDdl(SourceRecord record) {
+        Struct schemaChangeStruct = (Struct) record.value();
+        return schemaChangeStruct.getString(HistoryRecord.Fields.DDL_STATEMENTS);
+    }
 }

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializeSchema.java

@@ -119,10 +119,9 @@ private void deserializeSchemaChangeRecord(
             SourceRecord record, Collector<SeaTunnelRow> collector) {
         SchemaChangeEvent schemaChangeEvent = schemaChangeResolver.resolve(record, resultTypeInfo);
         if (schemaChangeEvent == null) {
-            log.info("Unsupported resolve schemaChangeEvent {}, just skip.", record);
+            log.warn("Unsupported resolve schemaChangeEvent {}, just skip.", record);
             return;
         }
-
         if (resultTypeInfo instanceof MultipleRowType) {
             Map<String, SeaTunnelRowType> newRowTypeMap = new HashMap<>();
             for (Map.Entry<String, SeaTunnelRowType> entry : (MultipleRowType) resultTypeInfo) {

seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/config/MySqlSourceConfigFactory.java

@@ -76,7 +76,8 @@ public MySqlSourceConfig create(int subtaskId) {
         // Note: the includeSchemaChanges parameter is used to control emitting the schema record,
         // only DataStream API program need to emit the schema record, the Table API need not
 
-        // TODO Not yet supported
+        // Some scenarios do not require automatic capture of table structure changes, so the
+        // default setting is false.
         props.setProperty("include.schema.changes", String.valueOf(false));
         // disable the offset flush totally
         props.setProperty("offset.flush.interval.ms", String.valueOf(Long.MAX_VALUE));

seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSource.java

@@ -102,6 +102,8 @@ public DebeziumDeserializationSchema<T> createDebeziumDeserializationSchema(
                         .setPhysicalRowType(physicalRowType)
                         .setResultTypeInfo(physicalRowType)
                         .setServerTimeZone(ZoneId.of(zoneId))
+                        .setSchemaChangeResolver(
+                                new MySqlSchemaChangeResolver(createSourceConfigFactory(config)))
                         .build();
     }