mysql: add support for invisible columns

src/mysql-util/src/lib.rs

@@ -114,10 +114,33 @@ pub enum MySqlError {
     MySql(#[from] mysql_async::Error),
 }
 
+/// Quotes MySQL identifiers. [See MySQL quote_identifier()](https://github.com/mysql/mysql-sys/blob/master/functions/quote_identifier.sql)
+pub fn quote_identifier(identifier: &str) -> String {
+    let mut escaped = identifier.replace("`", "``");
+    escaped.insert(0, '`');
+    escaped.push('`');
+    escaped
+}
+
 // NOTE: this error was renamed between MySQL 5.7 and 8.0
 // https://dev.mysql.com/doc/mysql-errors/8.0/en/server-error-reference.html#error_er_source_fatal_error_reading_binlog
 // https://dev.mysql.com/doc/mysql-errors/5.7/en/server-error-reference.html#error_er_master_fatal_error_reading_binlog
 pub const ER_SOURCE_FATAL_ERROR_READING_BINLOG_CODE: u16 = 1236;
 
 // https://dev.mysql.com/doc/mysql-errors/8.0/en/server-error-reference.html#error_er_no_such_table
 pub const ER_NO_SUCH_TABLE: u16 = 1146;
+
+#[cfg(test)]
+mod tests {
+
+    use super::quote_identifier;
+    #[mz_ore::test]
+    fn test_identifier_quoting() {
+        let expected = vec!["`a`", "`naughty``sql`", "```;naughty;sql;```"];
+        let input = ["a", "naughty`sql", "`;naughty;sql;`"]
+            .iter()
+            .map(|raw_str| quote_identifier(raw_str))
+            .collect::<Vec<_>>();
+        assert_eq!(expected, input);
+    }
+}

src/storage/src/source/mysql.rs

@@ -58,6 +58,7 @@ use std::rc::Rc;
 
 use differential_dataflow::AsCollection;
 use itertools::Itertools;
+use mz_mysql_util::quote_identifier;
 use mz_ore::cast::CastFrom;
 use mz_repr::Diff;
 use mz_repr::GlobalId;
@@ -338,7 +339,12 @@ impl MySqlTableName {
 
 impl fmt::Display for MySqlTableName {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "`{}`.`{}`", self.0, self.1)
+        write!(
+            f,
+            "{}.{}",
+            quote_identifier(&self.0),
+            quote_identifier(&self.1)
+        )
     }
 }
 

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

@@ -90,9 +90,12 @@ use std::sync::Arc;
 
 use differential_dataflow::AsCollection;
 use futures::TryStreamExt;
+use itertools::Itertools;
 use mysql_async::prelude::Queryable;
 use mysql_async::{IsolationLevel, Row as MySqlRow, TxOpts};
-use mz_mysql_util::{pack_mysql_row, query_sys_var, MySqlError, ER_NO_SUCH_TABLE};
+use mz_mysql_util::{
+    pack_mysql_row, query_sys_var, quote_identifier, MySqlError, ER_NO_SUCH_TABLE,
+};
 use mz_ore::cast::CastFrom;
 use mz_ore::future::InTask;
 use mz_ore::iter::IteratorExt;
@@ -404,9 +407,8 @@ pub(crate) fn render<G: Scope<Timestamp = GtidPartition>>(
 
                 let mut snapshot_staged = 0;
                 for (table, outputs) in &reader_snapshot_table_info {
-                    let query = format!("SELECT * FROM {}", table);
-                    trace!(%id, "timely-{worker_id} reading snapshot from \
-                                 table '{table}'");
+                    let query = build_snapshot_query(outputs);
+                    trace!(%id, "timely-{worker_id} reading snapshot query='{}'", query);
                     let mut results = tx.exec_stream(query, ()).await?;
                     let mut count = 0;
                     while let Some(row) = results.try_next().await? {
@@ -517,6 +519,31 @@ where
     Ok(total)
 }
 
+/// Builds the SQL query to be used for creating the snapshot using the first entry in outputs.
+///
+/// Expect `outputs` to contain entries for a single table, and to have at least 1 entry.
+/// Expect that each MySqlTableDesc entry contains all columns described in information_schema.columns.
+#[must_use]
+fn build_snapshot_query(outputs: &[SourceOutputInfo]) -> String {
+    let info = outputs.first().expect("MySQL table info");
+    if outputs.len() > 1
+        && !outputs
+            .iter()
+            .skip(1)
+            .map(|other_info| &other_info.desc.columns)
+            .all(|other_columns| *other_columns == info.desc.columns)
+    {
+        panic!("Mismatch in table descriptions for {}", info.table_name);
+    }
+    let columns = info
+        .desc
+        .columns
+        .iter()
+        .map(|col| quote_identifier(&col.name))
+        .join(", ");
+    format!("SELECT {} FROM {}", columns, info.table_name)
+}
+
 #[derive(Default)]
 struct TableStatistics {
     count_latency: f64,
@@ -541,3 +568,48 @@ where
 
     Ok(stats)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use mz_mysql_util::{MySqlColumnDesc, MySqlTableDesc};
+    use timely::progress::Antichain;
+
+    #[mz_ore::test]
+    fn snapshot_query_duplicate_table() {
+        let schema_name = "myschema".to_string();
+        let table_name = "mytable".to_string();
+        let table = MySqlTableName(schema_name.clone(), table_name.clone());
+        let columns = ["c1", "c2", "c3"]
+            .iter()
+            .map(|col| MySqlColumnDesc {
+                name: col.to_string(),
+                column_type: None,
+                meta: None,
+            })
+            .collect::<Vec<_>>();
+        let desc = MySqlTableDesc {
+            schema_name: schema_name.clone(),
+            name: table_name.clone(),
+            columns,
+            keys: BTreeSet::default(),
+        };
+        let info = SourceOutputInfo {
+            output_index: 1, // ignored
+            table_name: table.clone(),
+            desc,
+            text_columns: vec![],
+            exclude_columns: vec![],
+            initial_gtid_set: Antichain::default(),
+            resume_upper: Antichain::default(),
+        };
+        let query = build_snapshot_query(&[info.clone(), info]);
+        assert_eq!(
+            format!(
+                "SELECT `c1`, `c2`, `c3` FROM `{}`.`{}`",
+                &schema_name, &table_name
+            ),
+            query
+        );
+    }
+}

test/mysql-cdc/invisible-columns.td

@@ -29,20 +29,31 @@ DROP DATABASE IF EXISTS public;
 CREATE DATABASE public;
 USE public;
 
-CREATE TABLE t1 (f1 INT, f2 INT INVISIBLE, f3 INT INVISIBLE);
-INSERT INTO t1 (f1, f2, f3) VALUES (10, 20, 30);
+SET sql_generate_invisible_primary_key=ON;
+CREATE TABLE t1 (f1 INT, f2 INT INVISIBLE, f3 DATE INVISIBLE, f4 INT INVISIBLE);
+INSERT INTO t1 (f1, f2, f3, f4) VALUES (10, 20, '2025-01-28', 6);
 INSERT INTO t1 VALUES (11);
 
-# TODO: database-issues#7782 (invisible columns not supported)
-# > CREATE SOURCE mz_source FROM MYSQL CONNECTION mysql_conn;
-# > CREATE TABLE t1 FROM SOURCE mz_source (REFERENCE public.t1);
-#
-# > SELECT * FROM t1;
-# 10
-# 11
-#
-# $ mysql-execute name=mysql
-# ALTER TABLE t1 ALTER COLUMN f2 SET VISIBLE;
-#
-# ! SELECT * FROM t1;
-# contains:incompatible schema change
+> CREATE SOURCE mz_source FROM MYSQL CONNECTION mysql_conn;
+> CREATE TABLE t1 FROM SOURCE mz_source (REFERENCE public.t1) WITH (TEXT COLUMNS (f3), EXCLUDE COLUMNS (f4));
+
+> SELECT * FROM t1;
+1 10 20 "2025-01-28"
+2 11 <null> <null>
+
+$ mysql-execute name=mysql
+ALTER TABLE t1 ALTER COLUMN f2 SET VISIBLE;
+INSERT INTO t1 (f1, f2, f3, f4) VALUES (111, 222, '2025-01-29', 6);
+INSERT INTO t1 (f1, f2) VALUES (1111, 2222);
+
+> SELECT * from t1;
+1 10 20 "2025-01-28"
+2 11 <null> <null>
+3 111 222 "2025-01-29"
+4 1111 2222 <null>
+
+$ mysql-execute name=mysql
+ALTER TABLE t1 DROP COLUMN f2;
+
+! SELECT * FROM t1;
+contains:incompatible schema change

test/mysql-cdc/mzcompose.py

@@ -15,7 +15,7 @@
 import threading
 from textwrap import dedent
 
-from materialize import buildkite
+from materialize import MZ_ROOT, buildkite
 from materialize.mysql_util import (
     retrieve_invalid_ssl_context_for_mysql,
     retrieve_ssl_context_for_mysql,
@@ -99,7 +99,9 @@ def workflow_cdc(c: Composition, parser: WorkflowArgumentParser) -> None:
 
     matching_files = []
     for filter in args.filter:
-        matching_files.extend(glob.glob(filter, root_dir="test/mysql-cdc"))
+        matching_files.extend(
+            glob.glob(filter, root_dir=MZ_ROOT / "test" / "mysql-cdc")
+        )
     sharded_files: list[str] = sorted(
         buildkite.shard_list(matching_files, lambda file: file)
     )

test/mysql-cdc/quoted-identifier.td

@@ -0,0 +1,34 @@
+# 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.
+
+#
+# identifiers that need proper quoting
+#
+
+> 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 `t``1` (`f``1` INT)
+INSERT INTO `t``1` VALUES(1);
+
+> CREATE SOURCE mz_source FROM MYSQL CONNECTION mysql_conn;
+> CREATE TABLE t1 FROM SOURCE mz_source (REFERENCE public."t`1");
+
+> SELECT * FROM t1;
+1