MySQL Source Versioning V2 - reorganized boogaloo (#36333)

This PR effectively re-implements source versioning for mysql after we
reverted the initial implementation last week due to decoding problems
with exclude columns resulting in an incident. In doing so, it fixes a
number of issues that existed with the first implementation, including:
MaterializeInc/database-issues#11312
MaterializeInc/database-issues#11313
MaterializeInc/database-issues#11315

And provides a safer mechanism for handling schema changes and changes
to the `binlog_row_metadata` MySQL system variable.

The first commit is roughly the changes in
#36253 which should be
merged first, and is required for the other changes.

Second commit updates the decoding logic based on the binlog metadata
setting at source creation

Third commit updates the logic to verify mysql schemas with the schemas
in the upstream, allowing for certain types of schema changes when
binlog_row_metadata is FULL.

Fourth commit contains docs for how to make schema changes to your mysql
source without downtime in materialize.

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: patrickwwbutler

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",
     ]

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,164 @@ 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;
-            }
-        };
+    // For each column in `table_desc` (in descriptor order), resolve its wire
+    // index. With binlog_full_metadata=true, columns are matched by name so a reordered upstream
+    // still decodes correctly; without binlog_full_metadata, rows have no column names and must be
+    // matched positionally. A `None` here means the upstream row is missing this column and is
+    // only tolerated for ignored columns, and for binlog_full_metadata = false, is only tolerated
+    // for ignored columns at the end of the table.
+    for (i, col_desc) in table_desc.columns.iter().enumerate() {
         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 = if !binlog_full_metadata {
+            // No column name metadata, so we match by index.
+            (i < row.len()).then_some(i)
+        } else {
+            // This means the row from the binlog has column name included in the metadata,
+            // so we can match on that instead of position.
+            row.columns_ref()
+                .iter()
+                .position(|wc| wc.name_str() == col_desc.name.as_str())
+        };
+
+        let wire_idx = match wire_idx {
+            Some(idx) => idx,
+            None => {
+                // We could not find a column in the incoming row that matches this descriptor column.
+                // This is an error as the column is not ignored (ignored columns have already been skipped).
+                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(

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,
+        binlog_full_metadata: bool,
+    ) -> Result<(), anyhow::Error> {
         if self == other {
             return Ok(());
         }
@@ -90,12 +94,42 @@ 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`.
-        let mut other_columns = other.columns.iter();
-        for self_column in &self.columns {
-            let other_column = other_columns.next().ok_or_else(|| {
+        // 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`.
+        for (i, self_column) in self.columns.iter().enumerate() {
+            if self_column.column_type.is_none() {
+                // This is an excluded column and can be ignored.
+                continue;
+            }
+            let wire_idx = if !binlog_full_metadata {
+                // No column name metadata, so we match by index.
+                (i < other.columns.len()).then_some(i)
+            } else {
+                // This means the row from the binlog has column name included in the metadata,
+                // so we can match on that instead of position.
+                other
+                    .columns
+                    .iter()
+                    .position(|oc| oc.name.as_str() == self_column.name.as_str())
+            };
+
+            let wire_idx = match wire_idx {
+                Some(idx) => idx,
+                None => {
+                    // We could not find a column in the incoming row that matches this descriptor column.
+                    // This is an error as the column is not ignored (ignored columns have already been skipped).
+                    return Err(anyhow::anyhow!(
+                        "column {} no longer present in table {}",
+                        self_column.name,
+                        self.name
+                    ));
+                }
+            };
+            let other_column = other.columns.get(wire_idx).ok_or_else(|| {
                 anyhow::anyhow!(
                     "column {} no longer present in table {}",
                     self_column.name,
@@ -110,7 +144,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.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)]

src/storage/src/source/mysql/replication.rs

@@ -394,7 +394,7 @@ pub(crate) fn render<'scope>(
 
                         events::handle_rows_event(
                             data,
-                            &repl_context,
+                            &mut repl_context,
                             &cur_gtid,
                             &mut row_event_buffer,
                         )

src/storage/src/source/mysql/replication/events.rs

@@ -8,6 +8,7 @@
 // by the Apache License, Version 2.0.
 
 use maplit::btreemap;
+use mysql_async::binlog::events::OptionalMetaExtractor;
 use mysql_common::binlog::events::{QueryEvent, RowsEventData};
 use mz_mysql_util::{MySqlError, pack_mysql_row};
 use mz_ore::iter::IteratorExt;
@@ -221,7 +222,7 @@ pub(super) async fn handle_query_event(
 /// frontier with which to advance the dataflow's progress.
 pub(super) async fn handle_rows_event(
     event: RowsEventData<'_>,
-    ctx: &ReplContext<'_>,
+    ctx: &mut ReplContext<'_>,
     new_gtid: &GtidPartition,
     event_buffer: &mut Vec<(
         (usize, Result<SourceMessage, DataflowError>),
@@ -262,6 +263,11 @@ pub(super) async fn handle_rows_event(
     // Capability for this event.
     let gtid_cap = ctx.data_cap_set.delayed(new_gtid);
 
+    // We can check here if the binlog has full row metadata by looking at the column name optional
+    // metadata, which is only present if full metadata is enabled.
+    let optional_metadata = OptionalMetaExtractor::new(table_map_event.iter_optional_meta())?;
+    let has_full_metadata = optional_metadata.iter_column_name().next().is_some();
+
     // Iterate over the rows in this RowsEvent. Each row is a pair of 'before_row', 'after_row',
     // to accomodate for updates and deletes (which include a before_row),
     // and updates and inserts (which inclued an after row).
@@ -289,20 +295,37 @@ 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) {
-                    Ok(row) => Ok(SourceMessage {
-                        key: Row::default(),
-                        value: row,
-                        metadata: Row::default(),
-                    }),
-                    // Produce a DefiniteError in the stream for any rows that fail to decode
-                    Err(err @ MySqlError::ValueDecodeError { .. }) => Err(DataflowError::from(
-                        DefiniteError::ValueDecodeError(err.to_string()),
-                    )),
-                    Err(err) => Err(err)?,
+                let event = if !has_full_metadata && output.binlog_full_metadata {
+                    ctx.errored_outputs.insert(output.output_index);
+                    Err(DataflowError::from(DefiniteError::ValueDecodeError(
+                        format!(
+                            "Table {0} 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",
+                            output.table_name
+                        ),
+                    )))
+                } else {
+                    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,
+                            metadata: Row::default(),
+                        }),
+                        // Produce a DefiniteError in the stream for any rows that fail to decode
+                        Err(err @ MySqlError::ValueDecodeError { .. }) => Err(DataflowError::from(
+                            DefiniteError::ValueDecodeError(err.to_string()),
+                        )),
+                        Err(err) => Err(err)?,
+                    }
                 };
 
                 let data = (output.output_index, event);