mysql: add support for invisible columns

ci/nightly/pipeline.template.yml

@@ -927,7 +927,7 @@ steps:
       - id: checks-0dt-restart-entire-mz-forced-migrations-azurite
         label: "Checks 0dt restart of the entire Mz with forced migrations with :azure: blob store"
         depends_on: build-aarch64
-        timeout_in_minutes: 120
+        timeout_in_minutes: 180
         parallelism: 2
         agents:
           queue: hetzner-aarch64-16cpu-32gb

ci/test/pipeline.template.yml

@@ -410,7 +410,7 @@ steps:
 
       - id: mysql-cdc-resumption
         label: "MySQL CDC resumption tests"
-        parallelism: 4
+        parallelism: 6
         depends_on: build-aarch64
         timeout_in_minutes: 30
         inputs: [test/mysql-cdc-resumption]

misc/python/materialize/checks/all_checks/mysql_cdc.py

@@ -480,5 +480,106 @@ def validate(self) -> Testdrive:
         )
 
 
+@externally_idempotent(False)
+class MySqlInvisibleColumn(Check):
+    def _can_run(self, e: Executor) -> bool:
+        return self.base_version > MzVersion.parse_mz("v0.132.0-dev")
+
+    def initialize(self) -> Testdrive:
+        return Testdrive(
+            dedent(
+                f"""
+                $ mysql-connect name=mysql url=mysql://root@mysql password={MySql.DEFAULT_ROOT_PASSWORD}
+
+                $ mysql-execute name=mysql
+                # create the database if it does not exist yet but do not drop it
+                CREATE DATABASE IF NOT EXISTS public;
+                USE public;
+
+                CREATE USER mysql4 IDENTIFIED BY 'mysql';
+                GRANT REPLICATION SLAVE ON *.* TO mysql4;
+                GRANT ALL ON public.* TO mysql4;
+
+                DROP TABLE IF EXISTS mysql_invisible_table;
+
+                CREATE TABLE mysql_invisible_table (f1 INT, f2 FLOAT INVISIBLE, f3 DATE INVISIBLE, f4 TEXT INVISIBLE);
+
+                INSERT INTO mysql_invisible_table (f1, f2, f3, f4) VALUES (1, 0.1, '2025-01-01', 'one');
+
+                > CREATE SECRET mysql_invisible_pass AS 'mysql';
+                > CREATE CONNECTION mysql_invisible_conn TO MYSQL (
+                    HOST 'mysql',
+                    USER mysql4,
+                    PASSWORD SECRET mysql_invisible_pass
+                  )
+
+                > CREATE SOURCE mysql_invisible_source
+                  FROM MYSQL CONNECTION mysql_invisible_conn;
+                > CREATE TABLE mysql_invisible_table FROM SOURCE mysql_invisible_source (REFERENCE public.mysql_invisible_table);
+
+                # Return all rows
+                > CREATE MATERIALIZED VIEW mysql_invisible_view AS
+                  SELECT * FROM mysql_invisible_table
+                """
+            )
+        )
+
+    def manipulate(self) -> list[Testdrive]:
+        return [
+            Testdrive(dedent(s))
+            for s in [
+                f"""
+                $ mysql-connect name=mysql url=mysql://root@mysql password={MySql.DEFAULT_ROOT_PASSWORD}
+
+                $ mysql-execute name=mysql
+                USE public;
+                INSERT INTO mysql_invisible_table (f1, f2, f3, f4) VALUES (2, 0.2, '2025-02-02', 'two');
+                """,
+                f"""
+                $ mysql-connect name=mysql url=mysql://root@mysql password={MySql.DEFAULT_ROOT_PASSWORD}
+
+                $ mysql-execute name=mysql
+                USE public;
+                INSERT INTO mysql_invisible_table (f1, f2, f3, f4) VALUES (3, 0.3, '2025-03-03', 'three');
+                """,
+            ]
+        ]
+
+    def validate(self) -> Testdrive:
+        return Testdrive(
+            dedent(
+                f"""
+                > SELECT * FROM mysql_invisible_table;
+                1 0.1 2025-01-01 one
+                2 0.2 2025-02-02 two
+                3 0.3 2025-03-03 three
+
+                $ mysql-connect name=mysql url=mysql://root@mysql password={MySql.DEFAULT_ROOT_PASSWORD}
+
+                $ mysql-execute name=mysql
+                USE public;
+                ALTER TABLE mysql_invisible_table ALTER COLUMN f2 SET VISIBLE;
+                INSERT INTO mysql_invisible_table (f1, f2, f3, f4) VALUES (4, 0.4, '2025-04-04', 'four');
+
+                > SELECT * FROM mysql_invisible_table;
+                1 0.1 2025-01-01 one
+                2 0.2 2025-02-02 two
+                3 0.3 2025-03-03 three
+                4 0.4 2025-04-04 four
+
+                # Rollback the last INSERTs so that validate() can be called multiple times
+                $ mysql-execute name=mysql
+                DELETE FROM mysql_invisible_table WHERE f1 = 4;
+                ALTER TABLE mysql_invisible_table ALTER COLUMN f2 SET INVISIBLE;
+
+                > SELECT * FROM mysql_invisible_table;
+                1 0.1 2025-01-01 one
+                2 0.2 2025-02-02 two
+                3 0.3 2025-03-03 three
+                """
+            )
+        )
+
+
 def remove_target_cluster_from_explain(sql: str) -> str:
     return re.sub(r"\n\s*Target cluster: \w+\n", "", sql)

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,29 @@ 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");
+    for output in &outputs[1..] {
+        assert!(
+            info.desc.columns == output.desc.columns,
+            "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 +566,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