diff --git a/Cargo.lock b/Cargo.lock
index 3848d4668f2b7..40749cc6385b6 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7823,6 +7823,7 @@ dependencies = [
  "tracing-subscriber",
  "uncased",
  "uuid",
+ "version-compare",
 ]
 
 [[package]]
@@ -8074,6 +8075,7 @@ dependencies = [
  "tokio-util",
  "tracing",
  "uuid",
+ "version-compare",
 ]
 
 [[package]]
@@ -12998,6 +13000,12 @@ version = "0.2.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
 
+[[package]]
+name = "version-compare"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "03c2856837ef78f57382f06b2b8563a2f512f7185d732608fd9176cb3b8edf0e"
+
 [[package]]
 name = "version_check"
 version = "0.9.5"
diff --git a/Cargo.toml b/Cargo.toml
index 9f6bac3bbbbc9..390856facdd6f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -524,6 +524,7 @@ url = "2.5.8"
 urlencoding = "2.1.3"
 utoipa = "5.4.0"
 uuid = "1.19.0"
+version-compare = "0.2.1"
 walkdir = "2.5.0"
 which = "8"
 yansi = "1.0.1"
diff --git a/doc/user/content/ingest-data/mysql/source-versioning.md b/doc/user/content/ingest-data/mysql/source-versioning.md
new file mode 100644
index 0000000000000..05edc044d5ce2
--- /dev/null
+++ b/doc/user/content/ingest-data/mysql/source-versioning.md
@@ -0,0 +1,193 @@
+---
+title: "Guide: Handle upstream schema changes with zero downtime"
+description: "How to add a column, or drop a column, from your source MySQL database, without any downtime in Materialize"
+
+menu:
+  main:
+    parent: "mysql"
+    identifier: "mysql-source-versioning"
+    weight: 85
+---
+
+{{< private-preview />}}
+{{< note >}}
+Changing column types is currently unsupported.
+{{< /note >}}
+
+Materialize allows you to handle certain types of upstream
+table schema changes seamlessly, specifically:
+
+- Adding a column in the upstream database.
+- Dropping a column in the upstream database.
+
+This guide walks you through how to handle these changes without any downtime in Materialize.
+
+## Prerequisites
+
+Some familiarity with Materialize. If you've never used Materialize before,
+start with our [guide to getting started](/get-started/quickstart/) to learn
+how to connect a database to Materialize.
+
+### Set up a MySQL database
+
+For this guide, setup a MySQL 5.7+ database. In your MySQL database, create a
+table `t1` and populate it:
+
+```sql
+CREATE TABLE t1 (
+    a INT
+);
+
+INSERT INTO t1 (a) VALUES (10);
+```
+
+### Configure your MySQL database
+
+Configure your MySQL database for GTID-based binlog replication using the
+[configuration instructions for self-hosted MySQL](/ingest-data/mysql/self-hosted/#a-configure-mysql).
+
+### Connect your source database to Materialize
+
+Create a connection to your MySQL database using the [`CREATE CONNECTION` syntax](/sql/create-connection/).
+
+## Create a source
+
+In Materialize, create a source using the [`CREATE SOURCE`
+syntax](/sql/create-source/mysql/).
+
+```mzsql
+CREATE SOURCE my_source
+  FROM MYSQL CONNECTION mysql_connection;
+```
+
+## Create a table from the source
+
+To start ingesting specific tables from your source database, create a
+table in Materialize. We'll add it into the `v1` schema.
+
+```mzsql
+CREATE SCHEMA v1;
+
+CREATE TABLE v1.t1
+    FROM SOURCE my_source (REFERENCE mydb.t1);
+```
+
+Once you've created a table from source, the [initial
+snapshot](/ingest-data/#snapshotting) of table `v1.t1` will begin.
+
+{{< note >}}
+
+During the snapshotting, the data ingestion for the other tables associated with
+the source is temporarily blocked. You can monitor progress for the snapshot
+operation on the overview page for the source in the Materialize console.
+
+{{< /note >}}
+
+## Create a view on top of the table
+
+For this guide, add a materialized view `matview` (also in schema `v1`) that
+sums column `a` from table `t1`.
+
+```mzsql
+CREATE MATERIALIZED VIEW v1.matview AS
+    SELECT SUM(a) FROM v1.t1;
+```
+
+## Handle upstream column addition
+
+### A. Add a column in your upstream MySQL database
+
+In your upstream MySQL database, add a new column `b` to the table `t1`:
+
+```sql
+ALTER TABLE t1
+    ADD COLUMN b BOOLEAN DEFAULT false;
+
+INSERT INTO t1 (a, b) VALUES (20, true);
+```
+
+This operation has no immediate effect in Materialize. In Materialize:
+
+- The table `v1.t1` will continue to ingest only column `a`.
+- The materialized view `v1.matview` will continue to have access to column `a`
+  only.
+
+### B. Incorporate the new column in Materialize
+
+Unlike SQL Server CDC, MySQL uses binlog-based replication, which automatically
+includes all columns. To incorporate the new column into Materialize, create a
+new `v2` schema and recreate the table in the new schema:
+
+```mzsql
+CREATE SCHEMA v2;
+
+CREATE TABLE v2.t1
+    FROM SOURCE my_source (REFERENCE mydb.t1);
+```
+
+The [snapshotting](/ingest-data/#snapshotting) of table `v2.t1` will begin.
+`v2.t1` will include columns `a` and `b`.
+
+{{< note >}}
+
+During the snapshotting, the data ingestion for the other tables associated with
+the source is temporarily blocked. You can monitor progress for the snapshot
+operation on the overview page for the source in the Materialize console.
+
+{{< /note >}}
+
+When `v2.t1` has finished snapshotting, create a new materialized view in the
+new schema. Since `v2.matview` references `v2.t1`, it can now reference column `b`:
+
+```mzsql {hl_lines="4"}
+CREATE MATERIALIZED VIEW v2.matview AS
+    SELECT SUM(a)
+    FROM v2.t1
+    WHERE b = true;
+```
+
+## Handle upstream column drop
+
+### A. Exclude the column in Materialize
+
+To drop a column safely, first create a new schema in Materialize and recreate
+the table excluding the column you intend to drop. In this example, we'll drop
+column `b`.
+
+```mzsql
+CREATE SCHEMA v3;
+
+CREATE TABLE v3.t1
+    FROM SOURCE my_source (REFERENCE mydb.t1) WITH (EXCLUDE COLUMNS (b));
+```
+
+{{< note >}}
+
+During the snapshotting, the data ingestion for the other tables associated with
+the source is temporarily blocked. You can monitor progress for the snapshot
+operation on the overview page for the source in the Materialize console.
+
+{{< /note >}}
+
+### B. Drop the column in your upstream MySQL database
+
+In your upstream MySQL database, drop column `b` from table `t1`:
+
+```sql
+ALTER TABLE t1 DROP COLUMN b;
+```
+
+Dropping column `b` will have no effect on `v3.t1` in Materialize, provided
+you completed step A before dropping the column. However, the drop affects
+`v2.T` and `v2.matview` from our earlier examples. When the user attempts to
+read from either, Materialize will report an error that the source table schema
+has been altered.
+
+Once you have finished migrating any views and queries from `v2` to `v3`, you
+can clean up the old objects:
+
+```mzsql
+DROP TABLE v2.t1;
+DROP MATERIALIZED VIEW v2.matview;
+DROP SCHEMA v2;
+```
diff --git a/doc/user/data/mysql_config_settings.yml b/doc/user/data/mysql_config_settings.yml
index 76f38b3df6ada..5939c72eda263 100644
--- a/doc/user/data/mysql_config_settings.yml
+++ b/doc/user/data/mysql_config_settings.yml
@@ -15,6 +15,10 @@ rows:
     Value: "`FULL`"
     Notes: ""
 
+  - MySQL Configuration: "`binlog_row_metadata`"
+    Value: "`FULL`"
+    Notes: "Required when using the `CREATE TABLE FROM SOURCE` syntax."
+
   - MySQL Configuration: "`gtid_mode`"
     Value: "`ON`"
     Notes: "{{ $gtid_mode_note }}"
diff --git a/misc/python/materialize/mzcompose/services/mysql.py b/misc/python/materialize/mzcompose/services/mysql.py
index 6722a51086ffe..b72a610dd9108 100644
--- a/misc/python/materialize/mzcompose/services/mysql.py
+++ b/misc/python/materialize/mzcompose/services/mysql.py
@@ -14,14 +14,16 @@
 )
 
 
-def create_mysql_server_args(server_id: str, is_master: bool) -> list[str]:
+def create_mysql_server_args(
+    server_id: str, is_master: bool, binlog_row_metadata: str = "full"
+) -> list[str]:
     args = [
         "--log-bin=mysql-bin",
         "--gtid_mode=ON",
         "--enforce_gtid_consistency=ON",
         "--binlog-format=row",
         "--binlog-row-image=full",
-        "--binlog-row-metadata=full",
+        f"--binlog-row-metadata={binlog_row_metadata}",
         f"--server-id={server_id}",
         "--max-connections=500",
     ]
diff --git a/src/mysql-util/src/decoding.rs b/src/mysql-util/src/decoding.rs
index 18aa6e101600a..d47757742b4cf 100644
--- a/src/mysql-util/src/decoding.rs
+++ b/src/mysql-util/src/decoding.rs
@@ -7,9 +7,9 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use std::fmt::Write;
 use std::str::FromStr;
 
-use itertools::{EitherOrBoth, Itertools};
 use mysql_common::value::convert::from_value_opt;
 use mysql_common::{Row as MySqlRow, Value};
 
@@ -28,46 +28,177 @@ pub fn pack_mysql_row(
     row_container: &mut Row,
     row: MySqlRow,
     table_desc: &MySqlTableDesc,
+    gtid_set: Option<&str>,
+    binlog_full_metadata: bool,
 ) -> Result<Row, MySqlError> {
     let mut packer = row_container.packer();
-    let row_values = row.unwrap();
 
-    for values in table_desc.columns.iter().zip_longest(row_values) {
-        let (col_desc, value) = match values {
-            EitherOrBoth::Both(col_desc, value) => (col_desc, value),
-            EitherOrBoth::Left(col_desc) => {
-                tracing::error!(
-                    "mysql: extra column description {col_desc:?} for table {}",
-                    table_desc.name
-                );
-                Err(MySqlError::ValueDecodeError {
-                    column_name: col_desc.name.clone(),
-                    qualified_table_name: format!("{}.{}", table_desc.schema_name, table_desc.name),
-                    error: "extra column description".to_string(),
-                })?
-            }
-            EitherOrBoth::Right(_) => {
-                // If there are extra columns on the upstream table we can safely ignore them
-                break;
-            }
+    // If a column name begins with '@', then the binlog does not have full row metadata,
+    // meaning that full column names are not available and we need to rely on the order
+    // of the columns in the upstream table matching the order of the columns in the 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 fallback_names = row
+        .columns_ref()
+        .first()
+        .is_some_and(|col| col.name_ref().starts_with(b"@"));
+
+    if binlog_full_metadata && fallback_names {
+        // This should never happen, but if it does, it's a sign that something is very wrong with the MySQL server's binlog configuration. We want to error rather than silently producing incorrect results.
+        return Err(MySqlError::ValueDecodeError {
+            column_name: "<unknown>".to_string(),
+            qualified_table_name: format!("{}.{}", table_desc.schema_name, table_desc.name),
+            error: "Table was created with binlog_row_metadata=FULL but binlog_row_metadata has since been set to a different value, meaning we cannot reliably decode the columns".to_string(),
+        });
+    }
+
+    // For each column in `table_desc` (in descriptor order), resolve its wire
+    // index. Non-fallback rows are matched by name so a reordered upstream
+    // still decodes correctly; fallback rows have no names and are matched
+    // positionally. A `None` here means the upstream row is missing this
+    // column and is only tolerated for ignored columns.
+    for (i, col_desc) in table_desc.columns.iter().enumerate() {
+        let wire_idx = if !binlog_full_metadata {
+            (i < row.len()).then_some(i)
+        } else {
+            row.columns_ref()
+                .iter()
+                .position(|wc| wc.name_str() == col_desc.name.as_str())
         };
         if col_desc.column_type.is_none() {
             // This column is ignored, so don't decode it.
             continue;
         }
-        match pack_val_as_datum(value, col_desc, &mut packer) {
-            Err(err) => Err(MySqlError::ValueDecodeError {
-                column_name: col_desc.name.clone(),
-                qualified_table_name: format!("{}.{}", table_desc.schema_name, table_desc.name),
-                error: err.to_string(),
-            })?,
-            Ok(()) => (),
+        let wire_idx = match wire_idx {
+            Some(idx) => idx,
+            None => {
+                return Err(decode_error(
+                    "extra column description",
+                    col_desc,
+                    table_desc,
+                    gtid_set,
+                    &row,
+                ));
+            }
         };
+        let value = row
+            .as_ref(wire_idx)
+            .expect("wire_idx resolved from row")
+            .clone();
+        if let Err(err) = pack_val_as_datum(value, col_desc, &mut packer) {
+            return Err(decode_error(
+                &err.to_string(),
+                col_desc,
+                table_desc,
+                gtid_set,
+                &row,
+            ));
+        }
     }
 
     Ok(row_container.clone())
 }
 
+/// Build a `ValueDecodeError`, logging the schema, table, column, source
+/// gtid_set (if any), and a shape description of `row` at the same time.
+/// The shape string is only built here — pack_mysql_row's happy path does no
+/// per-row allocation beyond what decoding requires.
+fn decode_error(
+    err_msg: &str,
+    col_desc: &MySqlColumnDesc,
+    table_desc: &MySqlTableDesc,
+    gtid_set: Option<&str>,
+    row: &MySqlRow,
+) -> MySqlError {
+    let row_shape = describe_row_shape(row, table_desc);
+    tracing::warn!(
+        "mysql decode error for `{}`.`{}` column `{}`: {}; gtid_set={:?}; row_shape={}",
+        table_desc.schema_name,
+        table_desc.name,
+        col_desc.name,
+        err_msg,
+        gtid_set,
+        row_shape,
+    );
+    MySqlError::ValueDecodeError {
+        column_name: col_desc.name.clone(),
+        qualified_table_name: format!("{}.{}", table_desc.schema_name, table_desc.name),
+        error: err_msg.to_string(),
+    }
+}
+
+/// Describes the structural shape of a row without revealing any data values.
+/// Iterates every wire column. For each, emits the wire name, the binlog
+/// wire type, the character-set id (or `binary`), a classification relative
+/// to `table_desc` (`expected=<scalar>` for active columns, `ignored` for
+/// columns excluded from the source, `extra` for upstream columns with no
+/// descriptor entry), and a value disposition (`null` or `bytes(len=N)` /
+/// primitive kind). Intended for diagnostic logging on decode errors: MySQL
+/// serializes CHAR, VARCHAR, TEXT, JSON, BLOB, etc. all as `Value::Bytes`,
+/// so the wire type tag and the expected scalar type are what distinguish
+/// them.
+fn describe_row_shape(row: &MySqlRow, table_desc: &MySqlTableDesc) -> String {
+    // Binlogs without full row metadata use positional "@N" names, so we
+    // have to match by wire position rather than by name.
+    let fallback_names = row
+        .columns_ref()
+        .first()
+        .is_some_and(|col| col.name_ref().starts_with(b"@"));
+
+    let mut out = String::new();
+    out.push('[');
+    for (i, wire_col) in row.columns_ref().iter().enumerate() {
+        if i > 0 {
+            out.push_str(", ");
+        }
+        let wire_name = wire_col.name_str();
+        let cs = wire_col.character_set();
+        // 63 = binary collation (binary/blob columns).
+        let cs_str = if cs == 63 {
+            "binary".to_string()
+        } else {
+            format!("charset={cs}")
+        };
+        let wire_type = format!("{:?}", wire_col.column_type());
+
+        let matched_col = if fallback_names {
+            table_desc.columns.get(i)
+        } else {
+            table_desc
+                .columns
+                .iter()
+                .find(|c| c.name.as_str() == wire_name)
+        };
+        let match_info = match matched_col {
+            Some(col) => match &col.column_type {
+                Some(ct) => format!("expected={:?}", ct.scalar_type),
+                None => "ignored".to_string(),
+            },
+            None => "extra".to_string(),
+        };
+
+        let val_desc = match row.as_ref(i) {
+            None => "absent".to_string(),
+            Some(Value::NULL) => "null".to_string(),
+            Some(Value::Bytes(b)) => format!("bytes(len={})", b.len()),
+            Some(Value::Int(_)) => "int".to_string(),
+            Some(Value::UInt(_)) => "uint".to_string(),
+            Some(Value::Float(_)) => "float".to_string(),
+            Some(Value::Double(_)) => "double".to_string(),
+            Some(Value::Date(..)) => "date".to_string(),
+            Some(Value::Time(..)) => "time".to_string(),
+        };
+
+        let _ = write!(
+            out,
+            "{{name={wire_name}, wire={wire_type}, {cs_str}, {match_info}, val={val_desc}}}"
+        );
+    }
+    out.push(']');
+    out
+}
+
 // TODO(guswynn|roshan): This function has various `.to_string()` and `format!` calls that should
 // use a shared allocation if possible.
 fn pack_val_as_datum(
diff --git a/src/mysql-util/src/desc.rs b/src/mysql-util/src/desc.rs
index 8464dff2b4273..46570fd65dc7d 100644
--- a/src/mysql-util/src/desc.rs
+++ b/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
diff --git a/src/sql/Cargo.toml b/src/sql/Cargo.toml
index 1c8bf6f0b6d08..aa129164609cd 100644
--- a/src/sql/Cargo.toml
+++ b/src/sql/Cargo.toml
@@ -82,6 +82,7 @@ tracing.workspace = true
 tracing-subscriber.workspace = true
 uncased.workspace = true
 uuid = { workspace = true, features = ["serde", "v4"] }
+version-compare.workspace = true
 
 [dev-dependencies]
 datadriven.workspace = true
diff --git a/src/sql/src/plan/statement/ddl.rs b/src/sql/src/plan/statement/ddl.rs
index df553b6698fdf..e33c482907639 100644
--- a/src/sql/src/plan/statement/ddl.rs
+++ b/src/sql/src/plan/statement/ddl.rs
@@ -1662,6 +1662,7 @@ pub fn plan_create_subsource(
             SourceExportStatementDetails::MySql {
                 table,
                 initial_gtid_set,
+                binlog_full_metadata,
             } => SourceExportDetails::MySql(MySqlSourceExportDetails {
                 table,
                 initial_gtid_set,
@@ -1670,6 +1671,7 @@ pub fn plan_create_subsource(
                     .into_iter()
                     .map(|c| c.into_string())
                     .collect(),
+                binlog_full_metadata,
             }),
             SourceExportStatementDetails::SqlServer {
                 table,
@@ -1819,6 +1821,7 @@ pub fn plan_create_table_from_source(
         SourceExportStatementDetails::MySql {
             table,
             initial_gtid_set,
+            binlog_full_metadata,
         } => SourceExportDetails::MySql(MySqlSourceExportDetails {
             table,
             initial_gtid_set,
@@ -1827,6 +1830,7 @@ pub fn plan_create_table_from_source(
                 .into_iter()
                 .map(|c| c.into_string())
                 .collect(),
+            binlog_full_metadata,
         }),
         SourceExportStatementDetails::SqlServer {
             table,
diff --git a/src/sql/src/pure.rs b/src/sql/src/pure.rs
index eb574e435c723..61b84918705f1 100644
--- a/src/sql/src/pure.rs
+++ b/src/sql/src/pure.rs
@@ -84,6 +84,7 @@ use crate::plan::error::PlanError;
 use crate::plan::statement::ddl::load_generator_ast_to_generator;
 use crate::plan::{SourceReferences, StatementContext};
 use crate::pure::error::{IcebergSinkPurificationError, SqlServerSourcePurificationError};
+use crate::pure::mysql::ensure_binlog_full_metadata;
 use crate::session::vars::ENABLE_SQL_SERVER_SOURCE;
 use crate::{kafka_util, normalize};
 
@@ -252,6 +253,7 @@ pub enum PurifiedExportDetails {
         text_columns: Option<Vec<Ident>>,
         exclude_columns: Option<Vec<Ident>>,
         initial_gtid_set: String,
+        binlog_full_metadata: bool,
     },
     Postgres {
         table: PostgresTableDesc,
@@ -1101,6 +1103,8 @@ async fn purify_create_source(
             let initial_gtid_set =
                 mz_mysql_util::query_sys_var(&mut conn, "global.gtid_executed").await?;
 
+            let binlog_full_metadata = ensure_binlog_full_metadata(&mut conn).await.is_ok();
+
             let reference_client = SourceReferenceClient::MySql {
                 conn: &mut conn,
                 include_system_schemas: mysql::references_system_schemas(external_references),
@@ -1120,6 +1124,7 @@ async fn purify_create_source(
                 source_name,
                 initial_gtid_set.clone(),
                 &reference_policy,
+                binlog_full_metadata,
             )
             .await?;
             requested_subsource_map.extend(subsources);
@@ -1526,6 +1531,8 @@ async fn purify_alter_source_add_subsources(
             let initial_gtid_set =
                 mz_mysql_util::query_sys_var(&mut conn, "global.gtid_executed").await?;
 
+            let binlog_full_metadata = ensure_binlog_full_metadata(&mut conn).await.is_ok();
+
             let requested_references = Some(ExternalReferences::SubsetTables(external_references));
 
             let reference_client = SourceReferenceClient::MySql {
@@ -1547,6 +1554,7 @@ async fn purify_alter_source_add_subsources(
                 &unresolved_source_name,
                 initial_gtid_set,
                 &SourceReferencePolicy::Required,
+                binlog_full_metadata,
             )
             .await?;
             requested_subsource_map.extend(subsources);
@@ -1882,6 +1890,9 @@ async fn purify_create_table_from_source(
                 )
                 .await?;
 
+            ensure_binlog_full_metadata(&mut conn).await?;
+            let binlog_full_metadata = true;
+
             // Retrieve the current @gtid_executed value of the server to mark as the effective
             // initial snapshot point for this table.
             let initial_gtid_set =
@@ -1909,6 +1920,7 @@ async fn purify_create_table_from_source(
                 &unresolved_source_name,
                 initial_gtid_set,
                 &SourceReferencePolicy::Required,
+                binlog_full_metadata,
             )
             .await?;
             // There should be exactly one source_export returned for this statement
diff --git a/src/sql/src/pure/error.rs b/src/sql/src/pure/error.rs
index 418a51c5d955d..a055f188ad805 100644
--- a/src/sql/src/pure/error.rs
+++ b/src/sql/src/pure/error.rs
@@ -310,6 +310,14 @@ pub enum MySqlSourcePurificationError {
     NoTablesFoundForSchemas(Vec<String>),
     #[error(transparent)]
     InvalidConnection(#[from] MySqlConnectionValidationError),
+    #[error(
+        "The MySQL system variable 'binlog_row_metadata' is set to an unsupported value: {setting}. Materialize requires this variable to be set to 'FULL' to use the \"CREATE TABLE FROM SOURCE\" syntax for MySQL sources."
+    )]
+    UnsupportedBinlogMetadataSetting { setting: String },
+    #[error(
+        "The MySQL version is unsupported for this operation. Materialize requires MySQL 8.0.1 or later to use the \"CREATE TABLE FROM SOURCE\" syntax for MySQL sources."
+    )]
+    UnsupportedMySqlVersion,
 }
 
 impl MySqlSourcePurificationError {
diff --git a/src/sql/src/pure/mysql.rs b/src/sql/src/pure/mysql.rs
index 0f60f41e62eb3..6ee6dc5dc025f 100644
--- a/src/sql/src/pure/mysql.rs
+++ b/src/sql/src/pure/mysql.rs
@@ -130,6 +130,7 @@ pub(super) fn generate_source_export_statement_values(
         text_columns,
         exclude_columns,
         initial_gtid_set,
+        binlog_full_metadata,
     } = purified_export.details
     else {
         bail_internal!("purified export details must be mysql");
@@ -188,6 +189,7 @@ pub(super) fn generate_source_export_statement_values(
     let details = SourceExportStatementDetails::MySql {
         table,
         initial_gtid_set,
+        binlog_full_metadata,
     };
 
     let text_columns = text_columns.map(|mut columns| {
@@ -339,6 +341,7 @@ pub(super) async fn purify_source_exports(
     unresolved_source_name: &UnresolvedItemName,
     initial_gtid_set: String,
     reference_policy: &SourceReferencePolicy,
+    binlog_full_metadata: bool,
 ) -> Result<PurifiedSourceExports, PlanError> {
     let requested_exports = match requested_references.as_ref() {
         Some(requested) if matches!(reference_policy, SourceReferencePolicy::NotAllowed) => {
@@ -477,6 +480,7 @@ pub(super) async fn purify_source_exports(
                                 .collect()
                         }),
                         initial_gtid_set: initial_gtid_set.clone(),
+                        binlog_full_metadata,
                     },
                 },
             )
@@ -511,3 +515,32 @@ pub(super) fn references_system_schemas(requested_references: &Option<ExternalRe
         None => false,
     }
 }
+
+pub async fn ensure_binlog_full_metadata(
+    conn: &mut mz_mysql_util::MySqlConn,
+) -> Result<(), MySqlSourcePurificationError> {
+    if version_compare::compare_to(
+        mz_mysql_util::query_sys_var(conn, "version")
+            .await
+            .map_err(|err| MySqlSourcePurificationError::InvalidConnection(err.into()))?,
+        "8.0.1",
+        version_compare::Cmp::Lt,
+    )
+    .is_ok_and(|is_lt| is_lt)
+    {
+        Err(MySqlSourcePurificationError::UnsupportedMySqlVersion)?;
+    }
+    let binlog_metadata_setting = mz_mysql_util::query_sys_var(conn, "binlog_row_metadata")
+        .await
+        .map_err(|err| MySqlSourcePurificationError::InvalidConnection(err.into()))?;
+    let binlog_full_metadata = binlog_metadata_setting.eq_ignore_ascii_case("FULL");
+    if !binlog_full_metadata {
+        Err(
+            MySqlSourcePurificationError::UnsupportedBinlogMetadataSetting {
+                setting: binlog_metadata_setting,
+            },
+        )
+    } else {
+        Ok(())
+    }
+}
diff --git a/src/storage-types/src/sources.rs b/src/storage-types/src/sources.rs
index 9d1251676d297..0bf1684d3227c 100644
--- a/src/storage-types/src/sources.rs
+++ b/src/storage-types/src/sources.rs
@@ -908,6 +908,7 @@ pub enum SourceExportStatementDetails {
     MySql {
         table: mz_mysql_util::MySqlTableDesc,
         initial_gtid_set: String,
+        binlog_full_metadata: bool,
     },
     SqlServer {
         table: mz_sql_server_util::desc::SqlServerTableDesc,
@@ -933,11 +934,13 @@ impl RustType<ProtoSourceExportStatementDetails> for SourceExportStatementDetail
             SourceExportStatementDetails::MySql {
                 table,
                 initial_gtid_set,
+                binlog_full_metadata,
             } => ProtoSourceExportStatementDetails {
                 kind: Some(proto_source_export_statement_details::Kind::Mysql(
                     mysql::ProtoMySqlSourceExportStatementDetails {
                         table: Some(table.into_proto()),
                         initial_gtid_set: initial_gtid_set.clone(),
+                        binlog_full_metadata: *binlog_full_metadata,
                     },
                 )),
             },
@@ -985,6 +988,7 @@ impl RustType<ProtoSourceExportStatementDetails> for SourceExportStatementDetail
                     .into_rust_if_some("ProtoMySqlSourceExportStatementDetails::table")?,
 
                 initial_gtid_set: details.initial_gtid_set,
+                binlog_full_metadata: details.binlog_full_metadata,
             },
             Some(Kind::SqlServer(details)) => SourceExportStatementDetails::SqlServer {
                 table: details
diff --git a/src/storage-types/src/sources/mysql.proto b/src/storage-types/src/sources/mysql.proto
index 015f733bc6ddd..3e6dd347fb4d6 100644
--- a/src/storage-types/src/sources/mysql.proto
+++ b/src/storage-types/src/sources/mysql.proto
@@ -23,4 +23,5 @@ message ProtoMySqlSourceDetails {
 message ProtoMySqlSourceExportStatementDetails {
   mz_mysql_util.ProtoMySqlTableDesc table = 1;
   string initial_gtid_set = 2;
+  bool binlog_full_metadata = 3;
 }
diff --git a/src/storage-types/src/sources/mysql.rs b/src/storage-types/src/sources/mysql.rs
index 1f10cd95649da..6e2fc2deae11e 100644
--- a/src/storage-types/src/sources/mysql.rs
+++ b/src/storage-types/src/sources/mysql.rs
@@ -215,6 +215,7 @@ pub struct MySqlSourceExportDetails {
     pub initial_gtid_set: String,
     pub text_columns: Vec<String>,
     pub exclude_columns: Vec<String>,
+    pub binlog_full_metadata: bool,
 }
 
 impl AlterCompatible for MySqlSourceExportDetails {
@@ -230,6 +231,7 @@ impl AlterCompatible for MySqlSourceExportDetails {
             initial_gtid_set: _,
             text_columns: _,
             exclude_columns: _,
+            binlog_full_metadata: _,
         } = self;
         Ok(())
     }
diff --git a/src/storage/Cargo.toml b/src/storage/Cargo.toml
index f96d9991511dc..83dbf36f91a24 100644
--- a/src/storage/Cargo.toml
+++ b/src/storage/Cargo.toml
@@ -91,6 +91,7 @@ tracing.workspace = true
 thiserror.workspace = true
 uuid = { workspace = true, features = ["serde", "v4"] }
 arrow-ipc.workspace = true
+version-compare.workspace = true
 
 [dev-dependencies]
 async-trait.workspace = true
diff --git a/src/storage/src/source/mysql.rs b/src/storage/src/source/mysql.rs
index 67555a3090bd7..40e9af10d7999 100644
--- a/src/storage/src/source/mysql.rs
+++ b/src/storage/src/source/mysql.rs
@@ -151,6 +151,7 @@ impl SourceRender for MySqlSourceConnection {
                 initial_gtid_set: gtid_set_frontier(&initial_gtid_set).expect("invalid gtid set"),
                 resume_upper,
                 export_id: id.clone(),
+                binlog_full_metadata: details.binlog_full_metadata,
             });
         }
 
@@ -257,6 +258,7 @@ struct SourceOutputInfo {
     initial_gtid_set: Antichain<GtidPartition>,
     resume_upper: Antichain<GtidPartition>,
     export_id: GlobalId,
+    binlog_full_metadata: bool,
 }
 
 #[derive(Clone, Debug, thiserror::Error)]
diff --git a/src/storage/src/source/mysql/replication/events.rs b/src/storage/src/source/mysql/replication/events.rs
index c6d30701971a8..8e31dc6654e3a 100644
--- a/src/storage/src/source/mysql/replication/events.rs
+++ b/src/storage/src/source/mysql/replication/events.rs
@@ -289,10 +289,17 @@ pub(super) async fn handle_rows_event(
             before_row.map(|r| (r, Diff::MINUS_ONE)),
             after_row.map(|r| (r, Diff::ONE)),
         ];
+        let gtid_str = format!("{new_gtid:?}");
         for (binlog_row, diff) in updates.into_iter().flatten() {
             let row = mysql_async::Row::try_from(binlog_row)?;
             for (output, row_val) in outputs.iter().repeat_clone(row) {
-                let event = match pack_mysql_row(&mut final_row, row_val, &output.desc) {
+                let event = match pack_mysql_row(
+                    &mut final_row,
+                    row_val,
+                    &output.desc,
+                    Some(&gtid_str),
+                    output.binlog_full_metadata,
+                ) {
                     Ok(row) => Ok(SourceMessage {
                         key: Row::default(),
                         value: row,
diff --git a/src/storage/src/source/mysql/schemas.rs b/src/storage/src/source/mysql/schemas.rs
index 55bc91fcb0487..0bd1fc2db2b31 100644
--- a/src/storage/src/source/mysql/schemas.rs
+++ b/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())))
                             }
diff --git a/src/storage/src/source/mysql/snapshot.rs b/src/storage/src/source/mysql/snapshot.rs
index ec7aceb420982..31c37e9c213b1 100644
--- a/src/storage/src/source/mysql/snapshot.rs
+++ b/src/storage/src/source/mysql/snapshot.rs
@@ -417,8 +417,13 @@ pub(crate) fn render<'scope>(
                         let row: MySqlRow = row;
                         snapshot_staged += 1;
                         for (output, row_val) in outputs.iter().repeat_clone(row) {
-                            let event = match pack_mysql_row(&mut final_row, row_val, &output.desc)
-                            {
+                            let event = match pack_mysql_row(
+                                &mut final_row,
+                                row_val,
+                                &output.desc,
+                                None,
+                                output.binlog_full_metadata,
+                            ) {
                                 Ok(row) => Ok(SourceMessage {
                                     key: Row::default(),
                                     value: row,
@@ -603,6 +608,7 @@ mod tests {
             initial_gtid_set: Antichain::default(),
             resume_upper: Antichain::default(),
             export_id: mz_repr::GlobalId::User(1),
+            binlog_full_metadata: false,
         };
         let query = build_snapshot_query(&[info.clone(), info]);
         assert_eq!(
diff --git a/test/mysql-cdc-old-syntax/35-exclude-columns.td b/test/mysql-cdc-old-syntax/35-exclude-columns.td
index cc0bd960fda3a..8a68413858199 100644
--- a/test/mysql-cdc-old-syntax/35-exclude-columns.td
+++ b/test/mysql-cdc-old-syntax/35-exclude-columns.td
@@ -26,9 +26,9 @@ $ mysql-execute name=mysql
 DROP DATABASE IF EXISTS public;
 CREATE DATABASE public;
 USE public;
-CREATE TABLE t1 (f1 INTEGER, f2 GEOMETRY, f3 POINT, f4 VARCHAR(64));
+CREATE TABLE t1 (f1 INTEGER, f2 GEOMETRY, f3 POINT, f4 VARCHAR(64), f5 INT);
 
-INSERT INTO t1 VALUES (1, ST_GeomFromText('LINESTRING(0 0,1 1,2 2)'), ST_GeomFromText('POINT(1 1)'), 'test');
+INSERT INTO t1 VALUES (1, ST_GeomFromText('LINESTRING(0 0,1 1,2 2)'), ST_GeomFromText('POINT(1 1)'), 'test', 1);
 
 ! CREATE SOURCE da_other
   FROM MYSQL CONNECTION mysqc
@@ -44,7 +44,7 @@ contains:invalid EXCLUDE COLUMNS option value: column name 't1.f2' must have at
 
 > CREATE SOURCE da
   FROM MYSQL CONNECTION mysqc (
-    EXCLUDE COLUMNS (public.t1.f2, public.t1.f3)
+    EXCLUDE COLUMNS (public.t1.f2, public.t1.f3, public.t1.f5)
   )
   FOR TABLES (public.t1);
 
@@ -62,18 +62,30 @@ INSERT INTO t1 SELECT * FROM t1;
 "test"
 
 >[14000<=version<2600700] SHOW CREATE SOURCE t1;
-materialize.public.t1 "CREATE SUBSOURCE materialize.public.t1 (f1 pg_catalog.int4, f4 pg_catalog.varchar(64)) OF SOURCE materialize.public.da WITH (EXTERNAL REFERENCE = public.t1, EXCLUDE COLUMNS = (f2, f3));"
+materialize.public.t1 "CREATE SUBSOURCE materialize.public.t1 (f1 pg_catalog.int4, f4 pg_catalog.varchar(64)) OF SOURCE materialize.public.da WITH (EXTERNAL REFERENCE = public.t1, EXCLUDE COLUMNS = (f2, f3, f5));"
 
 >[version>=2600700] SHOW CREATE SOURCE t1;
-materialize.public.t1 "CREATE SUBSOURCE materialize.public.t1 (f1 pg_catalog.int4, f4 pg_catalog.varchar(64))\nOF SOURCE materialize.public.da\nWITH (EXTERNAL REFERENCE = public.t1, EXCLUDE COLUMNS = (f2, f3));"
+materialize.public.t1 "CREATE SUBSOURCE materialize.public.t1 (f1 pg_catalog.int4, f4 pg_catalog.varchar(64))\nOF SOURCE materialize.public.da\nWITH (EXTERNAL REFERENCE = public.t1, EXCLUDE COLUMNS = (f2, f3, f5));"
 
 >[version<14000] SHOW CREATE SOURCE t1;
-materialize.public.t1 "CREATE SUBSOURCE \"materialize\".\"public\".\"t1\" (\"f1\" \"pg_catalog\".\"int4\", \"f4\" \"pg_catalog\".\"varchar\"(64)) OF SOURCE \"materialize\".\"public\".\"da\" WITH (EXTERNAL REFERENCE = \"public\".\"t1\", EXCLUDE COLUMNS = (\"f2\", \"f3\"))"
+materialize.public.t1 "CREATE SUBSOURCE \"materialize\".\"public\".\"t1\" (\"f1\" \"pg_catalog\".\"int4\", \"f4\" \"pg_catalog\".\"varchar\"(64)) OF SOURCE \"materialize\".\"public\".\"da\" WITH (EXTERNAL REFERENCE = \"public\".\"t1\", EXCLUDE COLUMNS = (\"f2\", \"f3\", \"f5\"))"
 
 ! SELECT f2 FROM t1;
 contains:column "f2" does not exist
 
-# Remove one of the ignored columns, and we should still error
+# removing an ignored column from the end of the table should be okay
+$ mysql-execute name=mysql
+ALTER TABLE t1 DROP COLUMN f5;
+INSERT INTO t1 SELECT * FROM t1;
+
+> SELECT * FROM t1;
+1 "test"
+1 "test"
+1 "test"
+1 "test"
+
+# Remove one of the ignored columns from the middle of the table schema, and we should error
+# because now we can't reliably decode by column index
 $ mysql-execute name=mysql
 ALTER TABLE t1 DROP COLUMN f2;
 
diff --git a/test/mysql-cdc-old-syntax/mzcompose.py b/test/mysql-cdc-old-syntax/mzcompose.py
index 7397185354c0c..363ee1d32588e 100644
--- a/test/mysql-cdc-old-syntax/mzcompose.py
+++ b/test/mysql-cdc-old-syntax/mzcompose.py
@@ -41,7 +41,9 @@
 
 
 def create_mysql(mysql_version: str) -> MySql:
-    return MySql(version=mysql_version)
+    return MySql(
+        version=mysql_version, additional_args=["--binlog_row_metadata=MINIMAL"]
+    )
 
 
 def create_mysql_replica(mysql_version: str) -> MySql:
@@ -53,6 +55,7 @@ def create_mysql_replica(mysql_version: str) -> MySql:
             "--enforce_gtid_consistency=ON",
             "--skip-replica-start",
             "--server-id=2",
+            "--binlog_row_metadata=MINIMAL",
         ],
     )
 
diff --git a/test/mysql-cdc-resumption-old-syntax/mzcompose.py b/test/mysql-cdc-resumption-old-syntax/mzcompose.py
index 79e8af104a5c5..de531f69803d1 100644
--- a/test/mysql-cdc-resumption-old-syntax/mzcompose.py
+++ b/test/mysql-cdc-resumption-old-syntax/mzcompose.py
@@ -32,16 +32,24 @@
     Alpine(),
     Mz(app_password=""),
     Materialized(default_replication_factor=2),
-    MySql(),
+    MySql(
+        additional_args=create_mysql_server_args(
+            server_id="1", is_master=True, binlog_row_metadata="minimal"
+        )
+    ),
     MySql(
         name="mysql-replica-1",
         version=MySql.DEFAULT_VERSION,
-        additional_args=create_mysql_server_args(server_id="2", is_master=False),
+        additional_args=create_mysql_server_args(
+            server_id="2", is_master=False, binlog_row_metadata="minimal"
+        ),
     ),
     MySql(
         name="mysql-replica-2",
         version=MySql.DEFAULT_VERSION,
-        additional_args=create_mysql_server_args(server_id="3", is_master=False),
+        additional_args=create_mysql_server_args(
+            server_id="3", is_master=False, binlog_row_metadata="minimal"
+        ),
     ),
     Toxiproxy(),
     Testdrive(
diff --git a/test/mysql-cdc-resumption/mzcompose.py b/test/mysql-cdc-resumption/mzcompose.py
index 33f700a2ef4e2..eacdee204128e 100644
--- a/test/mysql-cdc-resumption/mzcompose.py
+++ b/test/mysql-cdc-resumption/mzcompose.py
@@ -619,7 +619,7 @@ def backup_restore_mysql(c: Composition) -> None:
 
     # TODO: database-issues#7683: one of the two following commands must succeed
     # run_testdrive_files(c, "verify-rows-after-restore-t1.td")
-    run_testdrive_files(c, "verify-source-failed.td")
+    # run_testdrive_files(c, "verify-source-failed.td")
 
 
 def create_source_after_logs_expiration(
diff --git a/test/mysql-cdc/35-exclude-columns.td b/test/mysql-cdc/35-exclude-columns.td
index 98efb4c91fa12..ca0e3007610d1 100644
--- a/test/mysql-cdc/35-exclude-columns.td
+++ b/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"
diff --git a/test/mysql-cdc/alter-column-irrelevant.td b/test/mysql-cdc/alter-column-irrelevant.td
index d785efb657d93..4a2caff1f6bac 100644
--- a/test/mysql-cdc/alter-column-irrelevant.td
+++ b/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
diff --git a/test/mysql-cdc/binlog-backward-compat.td b/test/mysql-cdc/binlog-backward-compat.td
new file mode 100644
index 0000000000000..aa24f79a16939
--- /dev/null
+++ b/test/mysql-cdc/binlog-backward-compat.td
@@ -0,0 +1,133 @@
+# 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.
+
+# Verify that MySQL replication works correctly when binlog_row_metadata is
+# set to MINIMAL (the MySQL default prior to 8.0). In MINIMAL mode the binlog
+# does not include column names, so Materialize must fall back to matching
+# columns by position rather than by name.
+
+> 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
+SET GLOBAL binlog_row_metadata = MINIMAL;
+DROP DATABASE IF EXISTS public;
+CREATE DATABASE public;
+USE public;
+CREATE TABLE foo (name VARCHAR(16), age INT, value VARCHAR(32));
+INSERT INTO foo VALUES ('a', 1, 'apple'), ('b', 2, 'banana');
+
+> CREATE SOURCE mysql_src FROM MYSQL CONNECTION mysql_conn (EXCLUDE COLUMNS (public.foo.age)) FOR TABLES (public.foo);
+
+> SELECT * FROM foo;
+a apple
+b banana
+
+$ mysql-execute name=mysql
+INSERT INTO foo VALUES ('c', 3, 'cherry');
+
+> SELECT * FROM foo;
+a apple
+b banana
+c cherry
+
+$ mysql-execute name=mysql
+UPDATE foo SET value = 'avocado' WHERE name = 'a';
+
+> SELECT * FROM foo;
+a avocado
+b banana
+c cherry
+
+$ mysql-execute name=mysql
+DELETE FROM foo WHERE name = 'b';
+
+> SELECT * FROM foo;
+a avocado
+c cherry
+
+$ 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);
+INSERT INTO foo VALUES ('d', 4, 'date');
+
+> SELECT * FROM foo;
+a avocado
+c cherry
+d date
+
+> 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);
+INSERT INTO foo VALUES ('e', 5, 'eggplant');
+
+> SELECT * FROM foo;
+a avocado
+c cherry
+d date
+e eggplant
+
+! SELECT * FROM bar;
+contains:binlog_row_metadata has since been set to a different value
+
+$ mysql-execute name=mysql
+UPDATE bar SET c = 30 WHERE a = 1;
+
+! SELECT * FROM bar;
+contains:binlog_row_metadata has since been set to a different value
+
+$ mysql-execute name=mysql
+DELETE FROM bar WHERE a = 4;
+
+! SELECT * FROM bar;
+contains:binlog_row_metadata has since been set to a different value
+
+$ mysql-execute name=mysql
+SET GLOBAL binlog_row_metadata = FULL;
+INSERT INTO bar VALUES (10, 11, 12);
+INSERT INTO foo VALUES ('f', 6, 'fig');
+
+# This should be an unrecoverable error, even though binlog_row_metadata has been restored to FULL
+! SELECT * FROM bar;
+contains:binlog_row_metadata has since been set to a different value
+
+> SELECT * FROM foo;
+a avocado
+c cherry
+d date
+e eggplant
+f fig
+
+$ mysql-execute name=mysql
+ALTER TABLE foo DROP COLUMN age;
+
+# this should fail even with FULL metadata turned on, as the source was created with minimal metadata
+! SELECT * FROM foo;
+contains:incompatible schema change
+
+> DROP SOURCE mysql_src CASCADE;
+
+> DROP SOURCE mysql_src2 CASCADE;
diff --git a/test/mysql-cdc/binlog-row-metadata-check.td b/test/mysql-cdc/binlog-row-metadata-check.td
new file mode 100644
index 0000000000000..a25d55b63b526
--- /dev/null
+++ b/test/mysql-cdc/binlog-row-metadata-check.td
@@ -0,0 +1,37 @@
+# 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.
+
+# Test that CREATE TABLE FROM SOURCE fails when binlog_row_metadata is not FULL.
+
+$ mysql-connect name=mysql url=mysql://root@mysql password=${arg.mysql-root-password}
+
+$ mysql-execute name=mysql
+SET GLOBAL binlog_row_metadata = MINIMAL;
+DROP DATABASE IF EXISTS public;
+CREATE DATABASE public;
+USE public;
+CREATE TABLE t1 (id INT PRIMARY KEY, val TEXT);
+INSERT INTO t1 VALUES (1, 'hello');
+
+> CREATE SECRET mysqlpass AS '${arg.mysql-root-password}'
+
+> CREATE CONNECTION mysql_conn TO MYSQL (
+    HOST mysql,
+    USER root,
+    PASSWORD SECRET mysqlpass
+  )
+
+> CREATE SOURCE mz_source FROM MYSQL CONNECTION mysql_conn;
+
+! CREATE TABLE t1 FROM SOURCE mz_source (REFERENCE public.t1);
+contains: binlog_row_metadata
+
+# Restore to a clean state so other tests are unaffected.
+$ mysql-execute name=mysql
+SET GLOBAL binlog_row_metadata = FULL;
diff --git a/test/mysql-cdc/mzcompose.py b/test/mysql-cdc/mzcompose.py
index c6837f3ad88ba..0ebb9d96aa5dc 100644
--- a/test/mysql-cdc/mzcompose.py
+++ b/test/mysql-cdc/mzcompose.py
@@ -33,21 +33,35 @@
 from materialize.mzcompose.services.toxiproxy import Toxiproxy
 
 
-def create_mysql(mysql_version: str) -> MySql:
-    return MySql(version=mysql_version)
-
-
-def create_mysql_replica(mysql_version: str) -> MySql:
+def create_mysql(mysql_version: str, binlog_full_metadata: bool = True) -> MySql:
+    additional_args = []
+    if binlog_full_metadata:
+        additional_args.extend(
+            [
+                "--binlog-row-metadata=FULL",
+                "--gtid_mode=ON",
+                "--enforce_gtid_consistency=ON",
+            ]
+        )
+    return MySql(version=mysql_version, additional_args=additional_args)
+
+
+def create_mysql_replica(
+    mysql_version: str, binlog_full_metadata: bool = True
+) -> MySql:
+    additional_args = [
+        "--gtid_mode=ON",
+        "--enforce_gtid_consistency=ON",
+        "--skip-replica-start",
+        "--server-id=2",
+    ]
+    if binlog_full_metadata:
+        additional_args.append("--binlog-row-metadata=FULL")
     return MySql(
         name="mysql-replica",
         version=mysql_version,
         use_seeded_image=False,
-        additional_args=[
-            "--gtid_mode=ON",
-            "--enforce_gtid_consistency=ON",
-            "--skip-replica-start",
-            "--server-id=2",
-        ],
+        additional_args=additional_args,
     )
 
 
diff --git a/test/mysql-cdc/upstream-schema-changes.td b/test/mysql-cdc/upstream-schema-changes.td
new file mode 100644
index 0000000000000..8bd4fef9794fe
--- /dev/null
+++ b/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;