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

mz_mysql_util/
lib.rs
