mz_storage/source/mysql/replication/
events.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10use itertools::Itertools;
11use sqlparser::ast::{ObjectName, ObjectNamePart, ObjectType, Statement};
12use sqlparser::dialect::MySqlDialect;
13use sqlparser::parser::Parser;
14use std::collections::BTreeMap;
15
16use maplit::btreemap;
17use mysql_async::binlog::events::{OptionalMetaExtractor, QueryEvent, RowsEventData};
18use mz_mysql_util::{MySqlError, pack_mysql_row};
19use mz_ore::iter::IteratorExt;
20use mz_repr::{Diff, Row};
21use mz_storage_types::errors::DataflowError;
22use mz_storage_types::sources::mysql::GtidPartition;
23use timely::progress::Timestamp;
24use tracing::trace;
25
26use crate::source::mysql::SourceOutputInfo;
27use crate::source::types::SourceMessage;
28
29use super::super::schemas::verify_schemas;
30use super::super::{DefiniteError, MySqlTableName, TransientError};
31use super::context::ReplContext;
32
33const DIALECT: MySqlDialect = MySqlDialect {};
34
35/// Returns the MySqlTableName for the given table name referenced in a
36/// SQL statement, using the current schema if the table name is unqualified.
37fn table_ident(name: &str, current_schema: &str) -> Result<MySqlTableName, TransientError> {
38    let stripped = name.replace('`', "");
39    let mut name_iter = stripped.split('.');
40    mysql_table_name(name_iter.next(), name_iter.next(), current_schema, name)
41}
42
43fn mysql_table_name(
44    first_component: Option<&str>,
45    second_component: Option<&str>,
46    current_schema: &str,
47    name: &str,
48) -> Result<MySqlTableName, TransientError> {
49    match (first_component, second_component) {
50        (Some(t_name), None) => Ok(MySqlTableName::new(current_schema, t_name)),
51        (Some(schema_name), Some(t_name)) => Ok(MySqlTableName::new(schema_name, t_name)),
52        _ => Err(TransientError::Generic(anyhow::anyhow!(
53            "Invalid table name from QueryEvent: {}",
54            name
55        ))),
56    }
57}
58
59/// This function has the same intent as table_ident except handles the object name type parsed out of the sql by sqlparser rather than a raw string.
60/// The ObjectName type is more flexible than the constraints for an identifier in mysql. i.e. it has a function type, which appears to only be supported
61/// in snowflake. It also has a vector for identifier components, however a table identifier from the binlog should only have 1 or 2 components - never 0 or more than 2
62fn table_ident_from_object_name(
63    name: &ObjectName,
64    current_schema: &str,
65) -> Result<MySqlTableName, TransientError> {
66    let processed_name_parts: Vec<String> = name.0.iter().map(|part| match part {
67        ObjectNamePart::Identifier(ident) => Ok(ident.value.clone()),
68        // Functions for table name identifiers are a snowflake-specific concept, unexpected for mysql so we should fail hard.
69        ObjectNamePart::Function(_) => Err(TransientError::Generic(anyhow::anyhow!(
70            "Invalid table name from QueryEvent, function identifiers not supported in mysql: {}", name
71        ))),
72    }).collect::<Result<_, _>>()?;
73    if processed_name_parts.len() != 1 && processed_name_parts.len() != 2 {
74        return Err(TransientError::Generic(anyhow::anyhow!(
75            "Invalid table name from QueryEvent: {}",
76            name
77        )));
78    }
79
80    let mut name_parts_iter = processed_name_parts.iter().map(|x| x.as_str());
81    mysql_table_name(
82        name_parts_iter.next(),
83        name_parts_iter.next(),
84        current_schema,
85        &name.to_string(),
86    )
87}
88
89/// Handles QueryEvents from the MySQL replication stream. Since we only use
90/// row-based replication, we only expect to see QueryEvents for DDL changes.
91///
92/// From the MySQL docs: 'A Query_event is created for each query that modifies
93/// the database, unless the query is logged row-based.' This means that we can
94/// expect any DDL changes to be represented as QueryEvents, which we must parse
95/// to figure out if any of the tables we care about have been affected.
96///
97/// This function returns a bool to represent whether the event that was handled
98/// represents a 'complete' event that should cause the frontier to advance beyond
99/// the current GTID.
100pub(super) async fn handle_query_event(
101    event: QueryEvent<'_>,
102    ctx: &mut ReplContext<'_>,
103    new_gtid: &GtidPartition,
104) -> Result<bool, TransientError> {
105    let (id, worker_id) = (ctx.config.id, ctx.config.worker_id);
106
107    let query = event.query();
108    let current_schema = event.schema();
109    let mut is_complete_event = false;
110
111    // MySQL does not permit transactional DDL, so luckily we don't need to
112    // worry about tracking BEGIN/COMMIT query events. We only need to look
113    // for DDL changes that affect the schema of the tables we care about.
114    let mut query_iter = query.split_ascii_whitespace();
115    let first = query_iter.next();
116    let second = query_iter.next();
117    match (
118        first.map(str::to_ascii_lowercase).as_deref(),
119        second.map(str::to_ascii_lowercase).as_deref(),
120    ) {
121        // Detect `ALTER TABLE <tbl>`, `RENAME TABLE <tbl>` statements
122        (Some("alter") | Some("rename"), Some("table")) => {
123            let table = table_ident(
124                query_iter.next().ok_or_else(|| {
125                    TransientError::Generic(anyhow::anyhow!("Invalid DDL query: {}", query))
126                })?,
127                &current_schema,
128            )?;
129            is_complete_event = true;
130            if ctx.table_info.contains_key(&table) {
131                trace!(%id, "timely-{worker_id} DDL change detected \
132                       for {table:?}");
133                // The schema check below opens a connection to MySQL and waits on it. Timely
134                // only notices capability changes while we're waiting. That makes this a good
135                // spot to publish the progress we've recorded so far.
136                //
137                // It's safe because that progress only covers events we've already finished.
138                // The QUERY event's GTID isn't recorded until after we return. Any errors we send
139                // below go out at this GTID, and we still retain those capabilities.
140                // Rewinds aren't a worry either. Rows events send rewinds before they return,
141                // and none are left waiting by the time we get here.
142                ctx.downgrade_data_cap_sets();
143                let info = &ctx.table_info[&table];
144                let mut conn = ctx
145                    .connection_config
146                    .connect(
147                        &format!("timely-{worker_id} MySQL "),
148                        &ctx.config.config.connection_context.ssh_tunnel_manager,
149                    )
150                    .await?;
151                for (err_output, err) in
152                    verify_schemas(&mut *conn, btreemap! { &table => info }).await?
153                {
154                    tracing::warn!(%id, "timely-{worker_id} DDL change \
155                           verification error for {table:?}[{}]: {err:?}",
156                           err_output.output_index);
157                    let output_index = err_output.output_index;
158                    let gtid_cap = ctx.data_cap_sets[output_index].delayed(new_gtid);
159                    let update = (Err(err.into()), new_gtid.clone(), Diff::ONE);
160                    ctx.data_outputs.give(output_index, &gtid_cap, update).await;
161                    ctx.errored_outputs.insert(output_index);
162                }
163            }
164        }
165        // Detect `DROP TABLE [IF EXISTS] <tbl>, <tbl>` statements.
166        (Some("drop"), Some("table")) => {
167            let dropped_tables = drop_table_identifiers(&current_schema, &query)?;
168
169            // Sources referencing the dropped table name that were created before the table was dropped. Before is determined
170            // by looking at the initial gtid set for the source and ensuring that's before the new gtid.
171            let sources_to_drop: BTreeMap<&MySqlTableName, Vec<&SourceOutputInfo>> = dropped_tables
172                .iter()
173                .filter_map(|table_name| {
174                    ctx.table_info
175                        .get_key_value(table_name)
176                        .map(|(name, info)| {
177                            let kept = info
178                                .iter()
179                                .filter(|output| {
180                                    !ctx.errored_outputs.contains(&output.output_index)
181                                    // Only drop sources that were created before the table was dropped.
182                                    && output.initial_gtid_set.less_equal(new_gtid)
183                                })
184                                .collect();
185                            (name, kept)
186                        })
187                })
188                .collect();
189            is_complete_event = true;
190            for (table_name, outputs) in sources_to_drop {
191                tracing::info!(%id, "timely-{worker_id} DDL change dropped outputs: {outputs:?}");
192                for output in outputs {
193                    let err = DefiniteError::TableDropped(table_name.to_string());
194                    let output_index = output.output_index;
195                    let gtid_cap = ctx.data_cap_sets[output_index].delayed(new_gtid);
196                    let update = (Err(err.into()), new_gtid.clone(), Diff::ONE);
197                    ctx.data_outputs.give(output_index, &gtid_cap, update).await;
198                    ctx.errored_outputs.insert(output_index);
199                }
200            }
201        }
202        // Detect `TRUNCATE [TABLE] <tbl>` statements
203        (Some("truncate"), Some(_)) => {
204            // We need the original un-lowercased version of 'second' since it might be a table ref
205            let second = second.expect("known to be Some");
206            let table = if second.eq_ignore_ascii_case("table") {
207                table_ident(
208                    query_iter.next().ok_or_else(|| {
209                        TransientError::Generic(anyhow::anyhow!("Invalid DDL query: {}", query))
210                    })?,
211                    &current_schema,
212                )?
213            } else {
214                table_ident(second, &current_schema)?
215            };
216            is_complete_event = true;
217            if ctx.table_info.contains_key(&table) {
218                trace!(%id, "timely-{worker_id} TRUNCATE detected \
219                       for {table:?}");
220                if let Some(info) = ctx.table_info.get(&table) {
221                    for output in info {
222                        let output_index = output.output_index;
223                        let gtid_cap = ctx.data_cap_sets[output_index].delayed(new_gtid);
224                        let update = (
225                            Err(DataflowError::from(DefiniteError::TableTruncated(
226                                table.to_string(),
227                            ))),
228                            new_gtid.clone(),
229                            Diff::ONE,
230                        );
231                        ctx.data_outputs.give(output_index, &gtid_cap, update).await;
232                        ctx.errored_outputs.insert(output_index);
233                    }
234                }
235            }
236        }
237        // Detect `COMMIT` statements which signify the end of a transaction on non-XA capable
238        // storage engines
239        (Some("commit"), None) => {
240            is_complete_event = true;
241        }
242        // Detect `CREATE TABLE <tbl>` statements which don't affect existing tables but do
243        // signify a complete event (e.g. for the purposes of advancing the GTID)
244        (Some("create"), Some("table")) => {
245            // CREATE TABLE ... SELECT will have subsequent `RowEvent`s, to account for this, the statement contains the clause "START TRANSACTION".
246            // https://dev.mysql.com/worklog/task/?id=13355
247
248            let mut peek_stream = query_iter.peekable();
249            let mut ctas = false;
250            while let Some(token) = peek_stream.next() {
251                if token.eq_ignore_ascii_case("start")
252                    && peek_stream
253                        .peek()
254                        .is_some_and(|t| t.eq_ignore_ascii_case("transaction"))
255                {
256                    ctas = true;
257                    break;
258                }
259            }
260            is_complete_event = !ctas;
261        }
262        _ => {}
263    }
264
265    Ok(is_complete_event)
266}
267
268/// Handles parsing table names from "DROP TABLE [IF EXISTS] table1, table2, table3".
269fn drop_table_identifiers(
270    current_schema: &str,
271    query: &str,
272) -> Result<Vec<MySqlTableName>, TransientError> {
273    let invalid =
274        |msg| TransientError::Generic(anyhow::anyhow!("Invalid DDL query, {msg}, got: {query}"));
275
276    let parse_result = Parser::parse_sql(&DIALECT, query)
277        .map_err(|e| TransientError::Generic(anyhow::anyhow!(e)))?;
278    let stmt = parse_result
279        .iter()
280        .exactly_one()
281        .map_err(|_| invalid("expected only a single statement from the binlog"))?;
282
283    let Statement::Drop {
284        object_type,
285        temporary,
286        names,
287        ..
288    } = stmt
289    else {
290        return Err(invalid("expected DROP statement"));
291    };
292
293    if *object_type != ObjectType::Table || *temporary {
294        return Err(invalid("expected DROP TABLE statement"));
295    }
296    let table_identifiers: Vec<MySqlTableName> = names
297        .iter()
298        .map(|name| table_ident_from_object_name(name, current_schema))
299        .collect::<Result<Vec<_>, _>>()?;
300    Ok(table_identifiers)
301}
302
303/// Handles RowsEvents from the MySQL replication stream. These events contain
304/// insert/update/delete events for a single transaction or committed statement.
305///
306/// We use these events to update the dataflow with the new rows, and return a new
307/// frontier with which to advance the dataflow's progress.
308pub(super) async fn handle_rows_event(
309    event: RowsEventData<'_>,
310    ctx: &mut ReplContext<'_>,
311    new_gtid: &GtidPartition,
312    event_buffer: &mut Vec<(
313        usize,
314        (Result<SourceMessage, DataflowError>, GtidPartition, Diff),
315    )>,
316) -> Result<(), TransientError> {
317    let (id, worker_id) = (ctx.config.id, ctx.config.worker_id);
318
319    // Find the relevant table
320    let binlog_table_id = event.table_id();
321    let table_map_event = ctx
322        .stream
323        .get_ref()
324        .get_tme(binlog_table_id)
325        .ok_or_else(|| TransientError::Generic(anyhow::anyhow!("Table map event not found")))?;
326    let table = MySqlTableName::new(
327        &*table_map_event.database_name(),
328        &*table_map_event.table_name(),
329    );
330
331    let outputs = ctx.table_info.get(&table).map(|outputs| {
332        outputs
333            .into_iter()
334            .filter(|output| !ctx.errored_outputs.contains(&output.output_index))
335            .collect::<Vec<_>>()
336    });
337    let outputs = match outputs {
338        Some(outputs) => outputs,
339        None => {
340            // We don't know about this table, or there are no un-errored outputs for it.
341            return Ok(());
342        }
343    };
344
345    trace!(%id, "timely-{worker_id} handling RowsEvent for {table:?}");
346
347    // Capabilities for this event, one per output in `outputs`.
348    let gtid_caps: Vec<_> = outputs
349        .iter()
350        .map(|output| ctx.data_cap_sets[output.output_index].delayed(new_gtid))
351        .collect();
352
353    // We can check here if the binlog has full row metadata by looking at the column name optional
354    // metadata, which is only present if full metadata is enabled.
355    let optional_metadata = OptionalMetaExtractor::new(table_map_event.iter_optional_meta())?;
356    let has_full_metadata = optional_metadata.iter_column_name().next().is_some();
357
358    // Iterate over the rows in this RowsEvent. Each row is a pair of 'before_row', 'after_row',
359    // to accomodate for updates and deletes (which include a before_row),
360    // and updates and inserts (which inclued an after row).
361    let mut final_row = Row::default();
362    let mut rows_iter = event.rows(table_map_event);
363    let mut rewind_count = 0;
364    let mut additions = 0;
365    let mut retractions = 0;
366    while let Some(Ok((before_row, after_row))) = rows_iter.next() {
367        // Update metrics for updates/inserts/deletes
368        match (&before_row, &after_row) {
369            (None, None) => {}
370            (Some(_), Some(_)) => {
371                ctx.metrics.updates.inc();
372            }
373            (None, Some(_)) => {
374                ctx.metrics.inserts.inc();
375            }
376            (Some(_), None) => {
377                ctx.metrics.deletes.inc();
378            }
379        }
380
381        let updates = [
382            before_row.map(|r| (r, Diff::MINUS_ONE)),
383            after_row.map(|r| (r, Diff::ONE)),
384        ];
385        let gtid_str = format!("{new_gtid:?}");
386        for (binlog_row, diff) in updates.into_iter().flatten() {
387            let row = mysql_async::Row::try_from(binlog_row)?;
388            for ((output, row_val), gtid_cap) in outputs.iter().repeat_clone(row).zip_eq(&gtid_caps)
389            {
390                let event = if !has_full_metadata && output.binlog_full_metadata {
391                    ctx.errored_outputs.insert(output.output_index);
392                    Err(DataflowError::from(DefiniteError::ValueDecodeError(
393                        format!(
394                            "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",
395                            output.table_name
396                        ),
397                    )))
398                } else {
399                    match pack_mysql_row(
400                        &mut final_row,
401                        row_val,
402                        &output.desc,
403                        Some(&gtid_str),
404                        output.binlog_full_metadata,
405                    ) {
406                        Ok(row) => Ok(SourceMessage {
407                            key: Row::default(),
408                            value: row,
409                            metadata: Row::default(),
410                        }),
411                        // Produce a DefiniteError in the stream for any rows that fail to decode
412                        Err(err @ MySqlError::ValueDecodeError { .. }) => Err(DataflowError::from(
413                            DefiniteError::ValueDecodeError(err.to_string()),
414                        )),
415                        Err(err) => Err(err)?,
416                    }
417                };
418
419                let output_index = output.output_index;
420
421                // Rewind this update if it was already present in the snapshot
422                if let Some(rewind_req) = ctx.rewinds.get(&output_index) {
423                    if !rewind_req.snapshot_upper.less_equal(new_gtid) {
424                        rewind_count += 1;
425                        event_buffer.push((
426                            output_index,
427                            (event.clone(), GtidPartition::minimum(), -diff),
428                        ));
429                    }
430                }
431                if diff.is_positive() {
432                    additions += 1;
433                } else {
434                    retractions += 1;
435                }
436                let update = (event, new_gtid.clone(), diff);
437                ctx.data_outputs.give(output_index, gtid_cap, update).await;
438            }
439        }
440    }
441
442    // We want to emit data in individual pieces to allow timely to break large chunks of data into
443    // containers. Naively interleaving new data and rewinds in the loop above defeats a timely
444    // optimization that caches push buffers if the `.give()` time has not changed.
445    //
446    // Instead, we buffer rewind events into a reusable buffer, and emit all at once here at the end.
447
448    // A port with a pending rewind holds only the minimum capability.
449    for (output_index, update) in event_buffer.drain(..) {
450        let rewind_cap = &ctx.data_cap_sets[output_index][0];
451        ctx.data_outputs
452            .give(output_index, rewind_cap, update)
453            .await;
454    }
455
456    trace!(
457        %id,
458        "timely-{worker_id} sent updates for {new_gtid:?} \
459            with {} updates ({} additions, {} retractions) and {} \
460            rewinds",
461        additions + retractions,
462        additions,
463        retractions,
464        rewind_count,
465    );
466
467    Ok(())
468}
469
470#[cfg(test)]
471mod tests {
472    use super::*;
473
474    fn table(schema: &str, name: &str) -> MySqlTableName {
475        MySqlTableName::new(schema, name)
476    }
477
478    fn parse_drop(
479        query: &str,
480        current_schema: &str,
481    ) -> Result<Vec<MySqlTableName>, TransientError> {
482        let mut tokens = query.split_ascii_whitespace();
483        let first = tokens.next();
484        let second = tokens.next();
485        match (
486            first.map(str::to_ascii_lowercase).as_deref(),
487            second.map(str::to_ascii_lowercase).as_deref(),
488        ) {
489            (Some("drop"), Some("table")) => drop_table_identifiers(current_schema, query),
490            _ => Ok(Vec::new()),
491        }
492    }
493
494    #[mz_ore::test]
495    fn table_ident_unqualified_uses_current_schema() {
496        assert_eq!(
497            table_ident("orders", "shop").unwrap(),
498            table("shop", "orders")
499        );
500    }
501
502    #[mz_ore::test]
503    fn table_ident_qualified_overrides_current_schema() {
504        assert_eq!(
505            table_ident("inventory.orders", "shop").unwrap(),
506            table("inventory", "orders"),
507        );
508    }
509
510    #[mz_ore::test]
511    fn table_ident_strips_backtick_quoting() {
512        assert_eq!(
513            table_ident("`inventory`.`orders`", "shop").unwrap(),
514            table("inventory", "orders"),
515        );
516    }
517
518    #[mz_ore::test]
519    fn drop_parses_single_unqualified_table() {
520        assert_eq!(
521            parse_drop("DROP TABLE orders", "shop").unwrap(),
522            vec![table("shop", "orders")],
523        );
524    }
525
526    #[mz_ore::test]
527    fn drop_parses_qualified_table() {
528        assert_eq!(
529            parse_drop("DROP TABLE inventory.orders", "shop").unwrap(),
530            vec![table("inventory", "orders")],
531        );
532    }
533
534    #[mz_ore::test]
535    fn drop_parses_backtick_quoted_table() {
536        assert_eq!(
537            parse_drop("DROP TABLE `inventory`.`orders`", "shop").unwrap(),
538            vec![table("inventory", "orders")],
539        );
540    }
541
542    #[mz_ore::test]
543    fn drop_parses_if_exists_clause() {
544        assert_eq!(
545            parse_drop("DROP TABLE IF EXISTS orders", "shop").unwrap(),
546            vec![table("shop", "orders")],
547        );
548        assert_eq!(
549            parse_drop("DROP TABLE if exists orders", "shop").unwrap(),
550            vec![table("shop", "orders")],
551        );
552    }
553
554    #[mz_ore::test]
555    fn drop_rejects_if_without_exists() {
556        assert!(parse_drop("DROP TABLE IF orders", "shop").is_err());
557    }
558
559    #[mz_ore::test]
560    fn drop_rejects_missing_table_name() {
561        assert!(parse_drop("DROP TABLE", "shop").is_err());
562    }
563
564    #[mz_ore::test]
565    fn drop_parses_multiple_space_separated_tables() {
566        assert_eq!(
567            parse_drop("DROP TABLE orders, customers, items", "shop").unwrap(),
568            vec![
569                table("shop", "orders"),
570                table("shop", "customers"),
571                table("shop", "items"),
572            ],
573        );
574    }
575
576    #[mz_ore::test]
577    fn drop_parses_comma_joined_tables_without_spaces() {
578        assert_eq!(
579            parse_drop("DROP TABLE `shop`.`orders`,`shop`.`customers`", "shop").unwrap(),
580            vec![table("shop", "orders"), table("shop", "customers")],
581        );
582    }
583
584    #[mz_ore::test]
585    fn drop_does_not_treat_comment_shaped_text_in_identifier_as_comment() {
586        assert_eq!(
587            parse_drop("DROP TABLE `tbl /* not a comment */`", "shop").unwrap(),
588            vec![table("shop", "tbl /* not a comment */")],
589        );
590    }
591
592    #[mz_ore::test]
593    fn drop_parses_table_with_restrict() {
594        assert_eq!(
595            parse_drop("DROP TABLE orders RESTRICT", "shop").unwrap(),
596            vec![table("shop", "orders")],
597        );
598    }
599
600    #[mz_ore::test]
601    fn drop_parses_multiple_tables_with_cascade() {
602        assert_eq!(
603            parse_drop("DROP TABLE orders, customers CASCADE", "shop").unwrap(),
604            vec![table("shop", "orders"), table("shop", "customers")],
605        );
606    }
607
608    #[mz_ore::test]
609    fn drop_of_multiple_statements_errors() {
610        // Defensive only: a real `Query_event` never packs two top-level
611        // statements together.
612        assert!(parse_drop("DROP TABLE orders; DROP TABLE customers", "shop").is_err());
613    }
614
615    #[mz_ore::test]
616    fn drop_temporary_table_is_not_detected() {
617        assert_eq!(
618            parse_drop("DROP TEMPORARY TABLE orders", "shop").unwrap(),
619            Vec::<MySqlTableName>::new(),
620        );
621    }
622
623    #[mz_ore::test]
624    fn drop_quoted_identifier_with_dot_is_a_single_table() {
625        assert_eq!(
626            parse_drop("DROP TABLE `weird.name`", "shop").unwrap(),
627            vec![table("shop", "weird.name")],
628        );
629    }
630
631    #[mz_ore::test]
632    fn drop_escaped_backtick_identifier_is_preserved() {
633        assert_eq!(
634            parse_drop("DROP TABLE `a``b`", "shop").unwrap(),
635            vec![table("shop", "a`b")],
636        );
637    }
638
639    #[mz_ore::test]
640    fn identifier_with_multiple_dots_errors() {
641        assert!(parse_drop("DROP TABLE def.shop.customer", "shop").is_err());
642    }
643
644    #[mz_ore::test]
645    fn ansi_quotes_parsed_properly() {
646        assert_eq!(
647            parse_drop("DROP TABLE \"customer\"", "shop").unwrap(),
648            vec![table("shop", "customer")],
649        );
650    }
651}
mz_storage/source/mysql/replication/events.rs

mz_storage/source/mysql/replication/
events.rs
