Merge pull request #167 from altmannmarcelo/metadata_for_mrbr

Optional Metadata fix for Minimal Row Based Replication

Author: blackbeam

src/binlog/mod.rs

@@ -1157,6 +1157,39 @@ mod tests {
                     assert_eq!(output, ev.data());
                 }
 
+                // https://github.com/blackbeam/rust_mysql_common/issues/162
+                if file_path.file_name().unwrap() == "minimal_row_metadata.000001" {
+                    let event_data = ev.read_data().unwrap();
+                    match event_data {
+                        Some(EventData::RowsEvent(ev)) => {
+                            let table_map_event =
+                                binlog_file.reader().get_tme(ev.table_id()).unwrap();
+                            let row = ev.rows(table_map_event).next().unwrap().unwrap();
+                            let after_image = row.1.unwrap();
+                            after_image
+                                .columns()
+                                .iter()
+                                .enumerate()
+                                .for_each(|(i, column)| {
+                                    if column.name_str() == "@2" {
+                                        assert_eq!(
+                                            column.character_set(),
+                                            CollationId::UTF8MB4_0900_AI_CI as u16
+                                        );
+                                    } else if column.name_str() == "@4" {
+                                        match after_image.as_ref(i).unwrap() {
+                                            BinlogValue::Value(val) => {
+                                                assert_eq!(val, &Value::Int(3230202323));
+                                            }
+                                            _ => panic!("Expected a value"),
+                                        }
+                                    }
+                                });
+                        }
+                        _ => (),
+                    }
+                }
+
                 ev_pos = ev_end;
             }
         }

src/binlog/row.rs

@@ -176,67 +176,67 @@ impl<'de> MyDeserialize<'de> for BinlogRow {
         let mut column_name_iter = opt_meta_extractor.iter_column_name();
 
         for i in 0..(num_columns as usize) {
-            // check if column is in columns list
-            if cols.get(i).as_deref().copied().unwrap_or(false) {
-                let column_type = table_info.get_column_type(i);
-
-                // TableMapEvent must define column type for the current column.
-                let column_type = match column_type {
-                    Ok(Some(ty)) => ty,
-                    Ok(None) => {
-                        return Err(io::Error::new(io::ErrorKind::InvalidData, "No column type"));
-                    }
-                    Err(e) => return Err(io::Error::new(io::ErrorKind::InvalidData, e)),
-                };
-
-                let column_meta = table_info.get_column_metadata(i).unwrap_or(&[]);
-
-                let is_partial = column_type == ColumnType::MYSQL_TYPE_JSON
-                    && partial_cols
-                        .as_mut()
-                        .and_then(|bits| bits.next().as_deref().copied())
-                        .unwrap_or(false);
-
-                let is_unsigned = column_type
-                    .is_numeric_type()
-                    .then(|| signedness_iterator.next())
-                    .flatten()
-                    .unwrap_or_default();
-
-                let charset = if column_type.is_character_type() {
-                    charset_iter.next().transpose()?.unwrap_or_default()
-                } else if column_type.is_enum_or_set_type() {
-                    enum_and_set_charset_iter
-                        .next()
-                        .transpose()?
-                        .unwrap_or_default()
-                } else {
-                    Default::default()
-                };
-
-                let column_name_raw = column_name_iter.next().transpose()?;
-                let column_name = column_name_raw
-                    .as_ref()
-                    .map(|x| Cow::Borrowed(x.name_raw()))
-                    .unwrap_or_else(|| {
-                        // default column name is `@<i>` where i is a column offset in a table
-                        Cow::Owned(format!("@{}", i).into())
-                    });
-
-                let mut column_flags = ColumnFlags::empty();
-
-                if is_unsigned {
-                    column_flags |= ColumnFlags::UNSIGNED_FLAG;
-                }
+            let column_type = table_info.get_column_type(i);
 
-                if primary_key_iter
-                    .next_if(|next| next.is_err() || next.as_ref().ok() == Some(&(i as u64)))
-                    .transpose()?
-                    .is_some()
-                {
-                    column_flags |= ColumnFlags::PRI_KEY_FLAG;
+            // TableMapEvent must define column type for the current column.
+            let column_type = match column_type {
+                Ok(Some(ty)) => ty,
+                Ok(None) => {
+                    return Err(io::Error::new(io::ErrorKind::InvalidData, "No column type"));
                 }
+                Err(e) => return Err(io::Error::new(io::ErrorKind::InvalidData, e)),
+            };
+
+            let column_meta = table_info.get_column_metadata(i).unwrap_or(&[]);
+
+            let is_partial = column_type == ColumnType::MYSQL_TYPE_JSON
+                && partial_cols
+                    .as_mut()
+                    .and_then(|bits| bits.next().as_deref().copied())
+                    .unwrap_or(false);
+
+            let is_unsigned = column_type
+                .is_numeric_type()
+                .then(|| signedness_iterator.next())
+                .flatten()
+                .unwrap_or_default();
+
+            let charset = if column_type.is_character_type() {
+                charset_iter.next().transpose()?.unwrap_or_default()
+            } else if column_type.is_enum_or_set_type() {
+                enum_and_set_charset_iter
+                    .next()
+                    .transpose()?
+                    .unwrap_or_default()
+            } else {
+                Default::default()
+            };
+
+            let column_name_raw = column_name_iter.next().transpose()?;
+            let column_name = column_name_raw
+                .as_ref()
+                .map(|x| Cow::Borrowed(x.name_raw()))
+                .unwrap_or_else(|| {
+                    // default column name is `@<i>` where i is a column offset in a table
+                    Cow::Owned(format!("@{}", i).into())
+                });
+
+            let mut column_flags = ColumnFlags::empty();
+
+            if is_unsigned {
+                column_flags |= ColumnFlags::UNSIGNED_FLAG;
+            }
+
+            if primary_key_iter
+                .next_if(|next| next.is_err() || next.as_ref().ok() == Some(&(i as u64)))
+                .transpose()?
+                .is_some()
+            {
+                column_flags |= ColumnFlags::PRI_KEY_FLAG;
+            }
 
+            // check if column is in columns list
+            if cols.get(i).as_deref().copied().unwrap_or(false) {
                 let column = Column::new(column_type)
                     .with_schema(table_info.database_name_raw())
                     .with_table(table_info.table_name_raw())