fix MaterializeInc/database-issues#11312

Author: patrickwwbutler

src/mysql-util/src/decoding.rs

@@ -37,37 +37,37 @@ pub fn pack_mysql_row(
     // This is a fallback for MySQL servers that do not have `binlog_row_metadata` set to
     // `FULL`. If the first column name does not begin with '@', then we can assume that
     // full metadata is available and we can match columns by name.
-    let row_values: Vec<Value> = if row
+    let zip_values: Vec<EitherOrBoth<&MySqlColumnDesc, Value>> = if row
         .columns_ref()
         .first()
         .is_some_and(|col| col.name_ref().starts_with(b"@"))
     {
-        row.unwrap()
+        table_desc
+            .columns
+            .iter()
+            .zip_longest(row.unwrap())
+            .collect()
     } else {
-        row.columns_ref()
+        table_desc
+            .columns
             .iter()
-            .enumerate()
-            .filter(|(_, col)| {
-                table_desc
-                    .columns
+            .filter(|col| col.column_type.is_some())
+            .map(|col| {
+                let pos = row
+                    .columns_ref()
                     .iter()
-                    .filter(|col| col.column_type.is_some())
-                    .any(|c| c.name.as_str() == col.name_str())
-            })
-            .map(|(i, _)| {
-                row.as_ref(i)
-                    .expect("Can't unwrap row if some of columns was taken")
-                    .clone()
+                    .position(|row_col| row_col.name_str() == col.name.as_str())
+                    .expect("column in table desc not found in row metadata");
+                EitherOrBoth::Both(
+                    col,
+                    row.get(pos)
+                        .expect("Can't unwrap row if some of columns was taken"),
+                )
             })
             .collect()
     };
 
-    for values in table_desc
-        .columns
-        .iter()
-        .filter(|col| col.column_type.is_some())
-        .zip_longest(row_values)
-    {
+    for values in zip_values {
         let (col_desc, value) = match values {
             EitherOrBoth::Both(col_desc, value) => (col_desc, value),
             EitherOrBoth::Left(col_desc) => {

test/mysql-cdc/binlog-backward-compat.td

@@ -61,6 +61,47 @@ c cherry
 
 > DROP SOURCE mysql_src CASCADE;
 
+$ mysql-execute name=mysql
+SET GLOBAL binlog_row_metadata = FULL;
+USE public;
+CREATE TABLE bar (a INT, b INT, c INT);
+INSERT INTO bar VALUES (1, 2, 3), (4, 5, 6);
+
+> CREATE SOURCE mysql_src2 FROM MYSQL CONNECTION mysql_conn;
+
+> CREATE TABLE bar FROM SOURCE mysql_src2 (REFERENCE public.bar) WITH (EXCLUDE COLUMNS (b));
+
+> SELECT * FROM bar;
+1 3
+4 6
+
+$ mysql-execute name=mysql
+SET GLOBAL binlog_row_metadata = MINIMAL;
+INSERT INTO bar VALUES (7, 8, 9);
+
+> SELECT * FROM bar;
+1 3
+4 6
+7 9
+
+$ mysql-execute name=mysql
+UPDATE bar SET c = 30 WHERE a = 1;
+
+> SELECT * FROM bar;
+1 30
+4 6
+7 9
+
+$ mysql-execute name=mysql
+DELETE FROM bar WHERE a = 4;
+
+> SELECT * FROM bar;
+1 30
+7 9
+
+> DROP SOURCE mysql_src2 CASCADE;
+
+
 # Restore to a clean state so other tests are unaffected.
 $ mysql-execute name=mysql
 SET GLOBAL binlog_row_metadata = FULL;