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..2c73bac879737 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,164 @@ pub fn pack_mysql_row( row_container: &mut Row, row: MySqlRow, table_desc: &MySqlTableDesc, + gtid_set: Option<&str>, + binlog_full_metadata: bool, ) -> Result { 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=` 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..051d821a866ca 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, + 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 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, resume_upper: Antichain, export_id: GlobalId, + binlog_full_metadata: bool, } #[derive(Clone, Debug, thiserror::Error)] diff --git a/src/storage/src/source/mysql/replication.rs b/src/storage/src/source/mysql/replication.rs index a0d7af34667e5..0030b1978492e 100644 --- a/src/storage/src/source/mysql/replication.rs +++ b/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, ) diff --git a/src/storage/src/source/mysql/replication/events.rs b/src/storage/src/source/mysql/replication/events.rs index c6d30701971a8..2664fc052b72e 100644 --- a/src/storage/src/source/mysql/replication/events.rs +++ b/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), @@ -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(>id_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); 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..7344155a21406 100644 --- a/src/storage/src/source/mysql/snapshot.rs +++ b/src/storage/src/source/mysql/snapshot.rs @@ -417,8 +417,16 @@ 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) - { + // 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, + &output.desc, + None, + output.binlog_full_metadata, + ) { Ok(row) => Ok(SourceMessage { key: Row::default(), value: row, @@ -527,8 +535,8 @@ where fn build_snapshot_query(outputs: &[SourceOutputInfo]) -> String { let info = outputs.first().expect("MySQL table info"); for output in &outputs[1..] { - // the columns are decoded solely based on position, so we just need to ensure that - // all columns are accounted for. + // the columns may be decoded based on position, and different outputs may replicate + // different columns, so we need to ensure that all columns are accounted for. assert!( info.desc.columns.len() == output.desc.columns.len(), "Mismatch in table descriptions for {}", @@ -603,6 +611,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..ad24dca97ab25 100644 --- a/test/mysql-cdc-old-syntax/mzcompose.py +++ b/test/mysql-cdc-old-syntax/mzcompose.py @@ -31,7 +31,7 @@ CockroachOrPostgresMetadata, ) from materialize.mzcompose.services.minio import Minio -from materialize.mzcompose.services.mysql import MySql +from materialize.mzcompose.services.mysql import MySql, create_mysql_server_args from materialize.mzcompose.services.mz import Mz from materialize.mzcompose.services.test_certs import TestCerts from materialize.mzcompose.services.testdrive import Testdrive @@ -41,19 +41,21 @@ def create_mysql(mysql_version: str) -> MySql: - return MySql(version=mysql_version) + return MySql( + version=mysql_version, + additional_args=create_mysql_server_args( + server_id="1", is_master=True, binlog_row_metadata="minimal" + ), + ) def create_mysql_replica(mysql_version: str) -> MySql: return MySql( name="mysql-replica", version=mysql_version, - additional_args=[ - "--gtid_mode=ON", - "--enforce_gtid_consistency=ON", - "--skip-replica-start", - "--server-id=2", - ], + additional_args=create_mysql_server_args( + server_id="2", is_master=False, 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..ba30f2805a4af 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( @@ -190,12 +198,16 @@ def workflow_master_changes(c: Composition) -> None: 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" + ), ), ): initialize(c, create_source=False) @@ -277,7 +289,9 @@ def workflow_switch_to_replica_and_kill_master(c: Composition) -> None: 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" + ), ), ): initialize(c) 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/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 +b banana +c cherry wild +d date ajwa + +$ mysql-execute name=mysql +INSERT INTO foo VALUES ('e', 'elderberry', 'montypython'); + +> SELECT * FROM foo2; +a apple +b banana +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;