MySQL Source Versioning V2 - reorganized boogaloo

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);