mz_mysql_util/
lib.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
10//! MySQL utility library.
11
12mod tunnel;
13use std::time::Duration;
14
15use aws_rds::RdsTokenError;
16pub use tunnel::{
17    Config, DEFAULT_CONNECT_TIMEOUT, DEFAULT_SNAPSHOT_LOCK_WAIT_TIMEOUT,
18    DEFAULT_SNAPSHOT_MAX_EXECUTION_TIME, DEFAULT_SNAPSHOT_WAIT_TIMEOUT, DEFAULT_TCP_KEEPALIVE,
19    MySqlConn, TimeoutConfig, TunnelConfig,
20};
21
22mod desc;
23pub use desc::{
24    MySqlColumnDesc, MySqlKeyDesc, MySqlTableDesc, ProtoMySqlColumnDesc, ProtoMySqlKeyDesc,
25    ProtoMySqlTableDesc,
26};
27
28mod replication;
29pub use replication::{
30    ensure_full_row_binlog_format, ensure_gtid_consistency, ensure_replication_commit_order,
31    query_sys_var,
32};
33
34pub mod schemas;
35pub use schemas::{
36    MySqlTableSchema, QualifiedTableRef, SYSTEM_SCHEMAS, SchemaRequest, schema_info,
37};
38
39pub mod privileges;
40pub use privileges::validate_source_privileges;
41
42pub mod decoding;
43pub use decoding::pack_mysql_row;
44
45pub mod probe;
46pub use probe::{KeyProber, MAX_KEY_LENGTH};
47
48pub mod partition;
49pub use partition::{PartitionParams, partition_table};
50
51mod aws_rds;
52
53#[derive(Debug, Clone)]
54pub struct UnsupportedDataType {
55    pub column_type: String,
56    pub qualified_table_name: String,
57    pub column_name: String,
58    pub intended_type: Option<String>,
59}
60
61impl std::fmt::Display for UnsupportedDataType {
62    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
63        match &self.intended_type {
64            Some(intended_type) => write!(
65                f,
66                "'{}.{}' of type '{}' represented as: '{}'",
67                self.qualified_table_name, self.column_name, self.column_type, intended_type
68            ),
69            None => write!(
70                f,
71                "'{}.{}' of type '{}'",
72                self.qualified_table_name, self.column_name, self.column_type
73            ),
74        }
75    }
76}
77
78#[derive(Debug, Clone)]
79pub struct MissingPrivilege {
80    pub privilege: String,
81    pub qualified_table_name: String,
82}
83
84impl std::fmt::Display for MissingPrivilege {
85    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
86        write!(
87            f,
88            "Missing privilege '{}' for '{}'",
89            self.privilege, self.qualified_table_name
90        )
91    }
92}
93
94#[derive(Debug, thiserror::Error)]
95pub enum MySqlError {
96    #[error("error validating privileges: {0:?}")]
97    MissingPrivileges(Vec<MissingPrivilege>),
98    #[error("error creating mysql connection with config: {0}")]
99    InvalidClientConfig(String),
100    #[error("error setting up ssh: {0}")]
101    Ssh(#[source] anyhow::Error),
102    #[error("error decoding value for '{qualified_table_name}' column '{column_name}': {error}")]
103    ValueDecodeError {
104        column_name: String,
105        qualified_table_name: String,
106        error: String,
107    },
108    #[error("non-UTF-8 key value in '{qualified_table_name}' column '{column_name}': {error}")]
109    NonUtf8KeyValue {
110        qualified_table_name: String,
111        column_name: String,
112        error: String,
113    },
114    #[error(
115        "missing row estimate in '{qualified_table_name}' for key range ({lower_bound}, {upper_bound})"
116    )]
117    MissingRowEstimate {
118        qualified_table_name: String,
119        /// Redacted at construction, safe to log.
120        lower_bound: String,
121        /// Redacted at construction, safe to log.
122        upper_bound: String,
123    },
124    #[error("unsupported data types: {columns:?}")]
125    UnsupportedDataTypes { columns: Vec<UnsupportedDataType> },
126    #[error("duplicated column names in table '{qualified_table_name}': {columns:?}")]
127    DuplicatedColumnNames {
128        qualified_table_name: String,
129        columns: Vec<String>,
130    },
131    #[error("invalid mysql system setting '{setting}'. Expected '{expected}'. Got '{actual}'.")]
132    InvalidSystemSetting {
133        setting: String,
134        expected: String,
135        actual: String,
136    },
137    /// Any other error we bail on.
138    #[error(transparent)]
139    Generic(#[from] anyhow::Error),
140    /// A mysql_async error.
141    #[error(transparent)]
142    MySql(#[from] mysql_async::Error),
143    #[error("connection attempt timed out after {0:?}")]
144    ConnectionTimeout(Duration),
145    /// Error retrieving AWS authorization token
146    #[error(transparent)]
147    AwsTokenError(#[from] RdsTokenError),
148}
149
150/// Quotes MySQL identifiers. [See MySQL quote_identifier()](https://github.com/mysql/mysql-sys/blob/master/functions/quote_identifier.sql)
151pub fn quote_identifier(identifier: &str) -> String {
152    let mut escaped = identifier.replace("`", "``");
153    escaped.insert(0, '`');
154    escaped.push('`');
155    escaped
156}
157
158// NOTE: this error was renamed between MySQL 5.7 and 8.0
159// https://dev.mysql.com/doc/mysql-errors/8.0/en/server-error-reference.html#error_er_source_fatal_error_reading_binlog
160// https://dev.mysql.com/doc/mysql-errors/5.7/en/server-error-reference.html#error_er_master_fatal_error_reading_binlog
161pub const ER_SOURCE_FATAL_ERROR_READING_BINLOG_CODE: u16 = 1236;
162
163// https://dev.mysql.com/doc/mysql-errors/8.0/en/server-error-reference.html#error_er_no_such_table
164pub const ER_NO_SUCH_TABLE: u16 = 1146;
165
166#[cfg(test)]
167mod tests {
168
169    use super::quote_identifier;
170    #[mz_ore::test]
171    fn test_identifier_quoting() {
172        let expected = vec!["`a`", "`naughty``sql`", "```;naughty;sql;```"];
173        let input = ["a", "naughty`sql", "`;naughty;sql;`"]
174            .iter()
175            .map(|raw_str| quote_identifier(raw_str))
176            .collect::<Vec<_>>();
177        assert_eq!(expected, input);
178    }
179}
mz_mysql_util/lib.rs

mz_mysql_util/
lib.rs
