fix MaterializeInc/database-issues#11313

Author: patrickwwbutler

src/mysql-util/src/desc.rs

@@ -101,10 +101,11 @@ impl MySqlTableDesc {
         // If we do have full metadata, then we can match columns by name and just check that all
         // columns in `self.columns` are present and compatible with columns in `other.columns`.
         let mut other_columns = other.columns.iter();
-        for self_column in &self.columns {
+        for self_column in self.columns.iter().filter(|col| col.column_type.is_some()) {
             let other_column = if full_metadata {
-                other_columns
-                    .by_ref()
+                other
+                    .columns
+                    .iter()
                     .find(|c| c.name == self_column.name)
                     .ok_or_else(|| {
                         anyhow::anyhow!(

test/mysql-cdc/upstream-schema-changes.td

@@ -103,14 +103,18 @@ f fig
 # dropping an excluded column should have no effect
 $ mysql-execute name=mysql
 ALTER TABLE foo DROP COLUMN meta_col;
+INSERT INTO foo VALUES ('g', 'grape');
 
+# The INSERT after the DROP COLUMN forces the binlog to advance past the ALTER
+# TABLE event, so the SELECT must succeed after the schema change is processed.
 > SELECT * FROM foo3;
 a apple
 b banana
 c cherry
 d date
 e elderberry
 f fig
+g grape
 
 > DROP TABLE foo3;
 
@@ -123,9 +127,10 @@ c cherry
 d date
 e elderberry
 f fig
+g grape
 
 $ mysql-execute name=mysql
-INSERT INTO foo VALUES ('g', 'grape');
+INSERT INTO foo VALUES ('h', 'honeydew');
 
 > SELECT * FROM foo4;
 a apple
@@ -158,16 +163,23 @@ b banana 7
 
 $ mysql-execute name=mysql
 ALTER TABLE bar DROP COLUMN meta_col;
+INSERT INTO bar VALUES ('c', 'cherry', 9, 'grove');
 
+# The INSERT after the DROP COLUMN forces the binlog to advance past the ALTER
+# TABLE event, so the SELECT must succeed after the schema change is processed.
 > SELECT * FROM bar1;
 a apple 5
 b banana 7
+c cherry 9
 
 $ mysql-execute name=mysql
 ALTER TABLE bar ADD COLUMN lastname VARCHAR(16) AFTER `name`;
+INSERT INTO bar VALUES ('d', 'date_lastname', 'date', 10, 'oasis');
 
 > SELECT * FROM bar1;
 a apple 5
 b banana 7
+c cherry 9
+d date 10
 
 > DROP SOURCE mysql_src CASCADE;