check optional metadata in handle_rows_event instead of decoding

Author: patrickwwbutler

src/mysql-util/src/decoding.rs

@@ -33,26 +33,6 @@ pub fn pack_mysql_row(
 ) -> Result<Row, MySqlError> {
     let mut packer = row_container.packer();
 
-    // 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

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;
@@ -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).
@@ -293,23 +299,34 @@ pub(super) async fn handle_rows_event(
         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,
-                    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 event = if !has_full_metadata && output.binlog_full_metadata {
+                    tracing::warn!(%id, "timely-{worker_id} missing full metadata for {table:?} \
+                        - this can lead to incorrect decoding of some data types. This metadata is only available on MySQL 8.0+ with binlog_version=2, and must be enabled with the binlog_row_metadata configuration option.");
+                    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);

src/storage/src/source/mysql/snapshot.rs

@@ -417,6 +417,9 @@ pub(crate) fn render<'scope>(
                         let row: MySqlRow = row;
                         snapshot_staged += 1;
                         for (output, row_val) in outputs.iter().repeat_clone(row) {
+                            // We don't need to verify if binlog_row_metadata matches the expected when snapshotting as
+                            // the snapshot query always returns rows with full metadata. If the output is configured
+                            // with binlog_full_metadata = false, then we will just ignore the metadata when decoding.
                             let event = match pack_mysql_row(
                                 &mut final_row,
                                 row_val,