support to read tables of inconsistent schema

flink-connector-debezium/src/main/java/com/ververica/cdc/debezium/table/RowDataDebeziumDeserializeSchema.java

@@ -43,6 +43,7 @@
 import io.debezium.time.NanoTimestamp;
 import io.debezium.time.Timestamp;
 import org.apache.kafka.connect.data.Decimal;
+import org.apache.kafka.connect.data.Field;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.source.SourceRecord;
@@ -499,11 +500,16 @@ public Object convert(Object dbzObj, Schema schema) throws Exception {
                 GenericRowData row = new GenericRowData(arity);
                 for (int i = 0; i < arity; i++) {
                     String fieldName = fieldNames[i];
-                    Object fieldValue = struct.get(fieldName);
-                    Schema fieldSchema = schema.field(fieldName).schema();
-                    Object convertedField =
-                            convertField(fieldConverters[i], fieldValue, fieldSchema);
-                    row.setField(i, convertedField);
+                    Field field = schema.field(fieldName);
+                    if (field == null) {
+                        row.setField(i, null);
+                    } else {
+                        Object fieldValue = struct.get(field);
+                        Schema fieldSchema = schema.field(fieldName).schema();
+                        Object convertedField =
+                                convertField(fieldConverters[i], fieldValue, fieldSchema);
+                        row.setField(i, convertedField);
+                    }
                 }
                 return row;
             }

flink-connector-mysql-cdc/src/test/java/com/ververica/cdc/connectors/mysql/table/MySqlConnectorITCase.java

@@ -62,6 +62,11 @@ public class MySqlConnectorITCase extends MySqlParallelSourceTestBase {
     private final UniqueDatabase customerDatabase =
             new UniqueDatabase(MYSQL_CONTAINER, "customer", TEST_USER, TEST_PASSWORD);
 
+    private final UniqueDatabase userDatabase1 =
+            new UniqueDatabase(MYSQL_CONTAINER, "user_1", TEST_USER, TEST_PASSWORD);
+    private final UniqueDatabase userDatabase2 =
+            new UniqueDatabase(MYSQL_CONTAINER, "user_2", TEST_USER, TEST_PASSWORD);
+
     private final StreamExecutionEnvironment env =
             StreamExecutionEnvironment.getExecutionEnvironment();
     private final StreamTableEnvironment tEnv =
@@ -616,6 +621,91 @@ public void testPrimaryKeyWithSnowflakeAlgorithm() throws Exception {
         result.getJobClient().get().cancel().get();
     }
 
+    @Test
+    public void testInconsistentSchema() throws Exception {
+        userDatabase1.createAndInitialize();
+        userDatabase2.createAndInitialize();
+        String sourceDDL =
+                String.format(
+                        "CREATE TABLE `user` ("
+                                + " `id` DECIMAL(20, 0) NOT NULL,"
+                                + " name STRING,"
+                                + " address STRING,"
+                                + " phone_number STRING,"
+                                + " email STRING,"
+                                + " age INT,"
+                                + " primary key (`id`) not enforced"
+                                + ") WITH ("
+                                + " 'connector' = 'mysql-cdc',"
+                                + " 'hostname' = '%s',"
+                                + " 'port' = '%s',"
+                                + " 'username' = '%s',"
+                                + " 'password' = '%s',"
+                                + " 'database-name' = '%s',"
+                                + " 'table-name' = '%s',"
+                                + " 'debezium.internal.implementation' = '%s',"
+                                + " 'scan.incremental.snapshot.enabled' = '%s',"
+                                + " 'server-id' = '%s',"
+                                + " 'scan.incremental.snapshot.chunk.size' = '%s'"
+                                + ")",
+                        MYSQL_CONTAINER.getHost(),
+                        MYSQL_CONTAINER.getDatabasePort(),
+                        userDatabase1.getUsername(),
+                        userDatabase1.getPassword(),
+                        "user_.*",
+                        "user_table_.*",
+                        getDezImplementation(),
+                        incrementalSnapshot,
+                        getServerId(),
+                        getSplitSize());
+        tEnv.executeSql(sourceDDL);
+
+        // async submit job
+        TableResult result = tEnv.executeSql("SELECT * FROM `user`");
+
+        CloseableIterator<Row> iterator = result.collect();
+        waitForSnapshotStarted(iterator);
+
+        try (Connection connection = userDatabase1.getJdbcConnection();
+                Statement statement = connection.createStatement()) {
+            statement.execute("UPDATE user_table_1_1 SET email = '[email protected]' WHERE id=111;");
+        }
+
+        try (Connection connection = userDatabase2.getJdbcConnection();
+                Statement statement = connection.createStatement()) {
+            statement.execute("UPDATE user_table_2_2 SET age = 20 WHERE id=221;");
+        }
+
+        String[] expected =
+                new String[] {
+                    "+I[111, user_111, Shanghai, 123567891234, [email protected], null]",
+                    "-U[111, user_111, Shanghai, 123567891234, [email protected], null]",
+                    "+U[111, user_111, Shanghai, 123567891234, [email protected], null]",
+                    "+I[121, user_121, Shanghai, 123567891234, null, null]",
+                    "+I[211, user_211, Shanghai, 123567891234, null, null]",
+                    "+I[221, user_221, Shanghai, 123567891234, null, 18]",
+                    "-U[221, user_221, Shanghai, 123567891234, null, 18]",
+                    "+U[221, user_221, Shanghai, 123567891234, null, 20]",
+                };
+
+        assertEqualsInAnyOrder(
+                Arrays.asList(expected), fetchRows(result.collect(), expected.length));
+        result.getJobClient().get().cancel().get();
+
+        // should drop the userDatabase1 and userDatabase2 for the test will run
+        // three times and create multiply databases with name like user_xxx.
+        // otherwise it'll read the database created by previous tests for we use `user_.*` to match
+        // database
+        try (Connection connection = userDatabase1.getJdbcConnection();
+                Statement statement = connection.createStatement()) {
+            statement.execute("drop database " + userDatabase1.getDatabaseName());
+        }
+        try (Connection connection = userDatabase2.getJdbcConnection();
+                Statement statement = connection.createStatement()) {
+            statement.execute("drop database " + userDatabase2.getDatabaseName());
+        }
+    }
+
     @Ignore
     @Test
     public void testStartupFromSpecificOffset() throws Exception {

flink-connector-mysql-cdc/src/test/resources/ddl/user_1.sql

@@ -0,0 +1,41 @@
+-- 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.
+
+-- ----------------------------------------------------------------------------------------------------------------
+-- DATABASE:  user_1
+-- ----------------------------------------------------------------------------------------------------------------
+
+-- Create user_table_1_1 table
+CREATE TABLE user_table_1_1 (
+    id INTEGER NOT NULL PRIMARY KEY,
+    name VARCHAR(255) NOT NULL DEFAULT 'flink',
+    address VARCHAR(1024),
+    phone_number VARCHAR(512),
+    email VARCHAR(255)
+);
+
+-- Create user_table_1_2 table
+CREATE TABLE user_table_1_2 (
+    id INTEGER NOT NULL PRIMARY KEY,
+    name VARCHAR(255) NOT NULL DEFAULT 'flink',
+    address VARCHAR(1024),
+    phone_number VARCHAR(512)
+);
+
+INSERT INTO user_table_1_1
+VALUES (111,"user_111","Shanghai","123567891234","[email protected]");
+
+INSERT INTO user_table_1_2
+VALUES (121,"user_121","Shanghai","123567891234");

flink-connector-mysql-cdc/src/test/resources/ddl/user_2.sql

@@ -0,0 +1,41 @@
+-- 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.
+
+-- ----------------------------------------------------------------------------------------------------------------
+-- DATABASE:  user_2
+-- ----------------------------------------------------------------------------------------------------------------
+
+-- Create user_table_2_1 table
+CREATE TABLE user_table_2_1 (
+    id INTEGER NOT NULL PRIMARY KEY,
+    name VARCHAR(255) NOT NULL DEFAULT 'flink',
+    address VARCHAR(1024),
+    phone_number VARCHAR(512)
+);
+
+-- Create user_table_2_2 table
+CREATE TABLE user_table_2_2 (
+    id INTEGER NOT NULL PRIMARY KEY,
+    name VARCHAR(255) NOT NULL DEFAULT 'flink',
+    address VARCHAR(1024),
+    phone_number VARCHAR(512),
+    age INTEGER
+);
+
+INSERT INTO user_table_2_1
+VALUES (211,"user_211","Shanghai","123567891234");
+
+INSERT INTO user_table_2_2
+VALUES (221,"user_221","Shanghai","123567891234", 18);