mysql: verify schema compatibility using column names when binlog_full_metadata=true

Adds a `full_metadata: bool` parameter to `MySqlTableDesc::determine_compatibility()`.
When true, columns are matched by name (allowing upstream reordering and safe
addition of new columns). When false, uses the original positional prefix check.
`verify_schemas()` now passes `output.binlog_full_metadata` to drive the choice.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: patrickwwbutler

src/mysql-util/src/desc.rs

@@ -75,7 +75,11 @@ impl MySqlTableDesc {
     /// exceptions:
     /// - `self`'s columns are a prefix of `other`'s columns.
     /// - `self`'s keys are all present in `other`
-    pub fn determine_compatibility(&self, other: &MySqlTableDesc) -> Result<(), anyhow::Error> {
+    pub fn determine_compatibility(
+        &self,
+        other: &MySqlTableDesc,
+        full_metadata: bool,
+    ) -> Result<(), anyhow::Error> {
         if self == other {
             return Ok(());
         }
@@ -90,12 +94,29 @@ impl MySqlTableDesc {
             );
         }
 
-        // `columns` is ordered by the ordinal_position of each column in the table,
-        // so as long as `self.columns` is a compatible prefix of `other.columns`, we can
-        // ignore extra columns from `other.columns`.
+        // In the case that we don't have full binlog row metadata, `columns` is ordered by the
+        // ordinal_position of each column in the table, so as long as `self.columns` is a
+        // compatible prefix of `other.columns`, we can ignore extra columns from `other.columns`.
+        //
+        // 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 {
-            let other_column = other_columns.next().ok_or_else(|| {
+        for self_column in self.columns.iter() {
+            let other_column = if full_metadata {
+                if self_column.column_type.is_none() {
+                    // This is an excluded column and can be ignored, as it may not have a
+                    // corresponding column in `other.columns` if the column was dropped upstream.
+                    continue;
+                }
+                other.columns.iter().find(|c| c.name == self_column.name)
+            } else {
+                other_columns.next()
+            };
+            if self_column.column_type.is_none() {
+                // This is an excluded column and can be ignored.
+                continue;
+            }
+            let other_column = other_column.ok_or_else(|| {
                 anyhow::anyhow!(
                     "column {} no longer present in table {}",
                     self_column.name,
@@ -110,7 +131,6 @@ impl MySqlTableDesc {
                 );
             }
         }
-
         // Our keys are all still present in exactly the same shape.
         // TODO: Implement a more relaxed key compatibility check:
         // We should check that for all keys that we know about there exists an upstream key whose

src/storage/src/source/mysql/schemas.rs

@@ -64,13 +64,18 @@ where
                             )),
                         );
                         match new_desc {
-                            Ok(desc) => match output.desc.determine_compatibility(&desc) {
-                                Ok(()) => None,
-                                Err(err) => Some((
-                                    output,
-                                    DefiniteError::IncompatibleSchema(err.to_string()),
-                                )),
-                            },
+                            Ok(desc) => {
+                                match output
+                                    .desc
+                                    .determine_compatibility(&desc, output.binlog_full_metadata)
+                                {
+                                    Ok(()) => None,
+                                    Err(err) => Some((
+                                        output,
+                                        DefiniteError::IncompatibleSchema(err.to_string()),
+                                    )),
+                                }
+                            }
                             Err(err) => {
                                 Some((output, DefiniteError::IncompatibleSchema(err.to_string())))
                             }

test/mysql-cdc/35-exclude-columns.td

@@ -68,5 +68,6 @@ contains:column "f2" does not exist
 $ mysql-execute name=mysql
 ALTER TABLE t1 DROP COLUMN f2;
 
-! select * from t1;
-contains:incompatible schema change
+> select * from t1;
+1 "test"
+1 "test"

test/mysql-cdc/alter-column-irrelevant.td

@@ -59,10 +59,12 @@ INSERT INTO t1 VALUES (2, 2);
 # add a new column to t1 at the beginning of
 $ mysql-execute name=mysql
 ALTER TABLE t1 ADD COLUMN f3 INTEGER FIRST;
-INSERT INTO t1 VALUES (3, 3, 3);
+INSERT INTO t1 VALUES (0, 3, 3);
 
-! SELECT * FROM t1;
-contains:incompatible schema change
+> SELECT * FROM t1;
+1
+2
+3
 
 # add a new column to t2
 $ mysql-execute name=mysql

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

@@ -0,0 +1,185 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# Perform various schema updates to the upstream table
+
+> CREATE SECRET mysqlpass AS '${arg.mysql-root-password}'
+
+> CREATE CONNECTION mysql_conn TO MYSQL (
+    HOST mysql,
+    USER root,
+    PASSWORD SECRET mysqlpass
+  )
+
+$ mysql-connect name=mysql url=mysql://root@mysql password=${arg.mysql-root-password}
+
+$ mysql-execute name=mysql
+DROP DATABASE IF EXISTS public;
+CREATE DATABASE public;
+USE public;
+CREATE TABLE foo (name VARCHAR(16), value VARCHAR(32));
+INSERT INTO foo VALUES ('a', 'apple'), ('b', 'banana');
+
+> CREATE SOURCE mysql_src FROM MYSQL CONNECTION mysql_conn;
+> CREATE TABLE foo1 FROM SOURCE mysql_src (REFERENCE public.foo);
+
+> SELECT * FROM foo1;
+a apple
+b banana
+
+$ mysql-execute name=mysql
+ALTER TABLE foo ADD COLUMN meta_col VARCHAR(32);
+INSERT INTO foo VALUES ('c', 'cherry', 'wild');
+
+# Adding a column to the upstream table does not affect foo1 — it continues to
+# replicate only the columns it was created with, ignoring the new meta_col.
+> SELECT * FROM foo1;
+a apple
+b banana
+c cherry
+
+$ mysql-execute name=mysql
+ALTER TABLE foo MODIFY meta_col VARCHAR(64);
+INSERT INTO foo VALUES ('d', 'date', 'ajwa');
+
+# Altering the newly added `meta_col` column does not brick `foo1` because `foo1` is
+# not following/replicating `meta_col`, so schema updates involving it are inconsequential.
+> SELECT * FROM foo1;
+a apple
+b banana
+c cherry
+d date
+
+> DROP TABLE foo1;
+
+# Unlike SQL Server CDC, MySQL binlog-based replication does not require a new capture
+# instance for the new column. Creating a table from the existing source will include
+# meta_col from the snapshot onward.
+> CREATE TABLE foo2 FROM SOURCE mysql_src (REFERENCE public.foo);
+> SELECT * FROM foo2;
+a apple <null>
+b banana <null>
+c cherry wild
+d date ajwa
+
+$ mysql-execute name=mysql
+INSERT INTO foo VALUES ('e', 'elderberry', 'montypython');
+
+> SELECT * FROM foo2;
+a apple <null>
+b banana <null>
+c cherry wild
+d date ajwa
+e elderberry montypython
+
+> DROP TABLE foo2;
+
+# We can also use EXCLUDE COLUMNS to exclude meta_col if desired.
+> CREATE TABLE foo3 FROM SOURCE mysql_src (REFERENCE public.foo) WITH (EXCLUDE COLUMNS = (meta_col));
+> SELECT * FROM foo3;
+a apple
+b banana
+c cherry
+d date
+e elderberry
+
+$ mysql-execute name=mysql
+INSERT INTO foo VALUES ('f', 'fig', 'sweet');
+
+> SELECT * FROM foo3;
+a apple
+b banana
+c cherry
+d date
+e elderberry
+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;
+
+# After meta_col has been dropped upstream, foo4 can be created without excluding it.
+> CREATE TABLE foo4 FROM SOURCE mysql_src (REFERENCE public.foo);
+> SELECT * FROM foo4;
+a apple
+b banana
+c cherry
+d date
+e elderberry
+f fig
+g grape
+
+$ mysql-execute name=mysql
+INSERT INTO foo VALUES ('h', 'honeydew');
+
+> SELECT * FROM foo4;
+a apple
+b banana
+c cherry
+d date
+e elderberry
+f fig
+g grape
+
+# Dropping a non-excluded (tracked) column stalls the source.
+$ mysql-execute name=mysql
+ALTER TABLE foo DROP COLUMN value;
+
+! SELECT * FROM foo4;
+contains:incompatible schema change
+
+> DROP TABLE foo4;
+
+# Test with multiple excluded columns.
+$ mysql-execute name=mysql
+CREATE TABLE bar (name VARCHAR(16), value VARCHAR(32), age BIGINT, location VARCHAR(32), meta_col VARCHAR(32));
+INSERT INTO bar VALUES ('a', 'apple', 5, 'orchard', 'blah'), ('b', 'banana', 7, 'tree', 'blahblah');
+
+> CREATE TABLE bar1 FROM SOURCE mysql_src (REFERENCE public.bar) WITH (EXCLUDE COLUMNS = (meta_col, location));
+
+> SELECT * FROM bar1;
+a apple 5
+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;