github/dbx
4
0
Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-10-21 06:00:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(transfer): recover unsupported MySQL collations

Browse Source 
Closes #6023
This commit is contained in:
t8y2
2026-08-12 12:20:45 +00:00
parent 6429334549
commit c570517fe0
2 changed files with 450 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/dbx-core/src/transfer.rs
+244 -30
View File
@@ -18,6 +18,9 @@ static OCEANBASE_MYSQL_TABLE_OPTION_RE: std::sync::LazyLock<Regex> = std::sync::
Regex::new(r"(?i)\b(?:AUTO_INCREMENT_MODE|REPLICA_NUM|USE_BLOOM_FILTER|TABLET_SIZE|PCTFREE)\s*=")
.expect("valid OceanBase MySQL table option regex")
});
static MYSQL_COLLATE_CLAUSE_RE: std::sync::LazyLock<Regex> = std::sync::LazyLock::new(|| {
Regex::new(r"(?i)\bCOLLATE\s*=?\s*([A-Za-z0-9_]+)\b").expect("valid MySQL COLLATE clause regex")
});
const MAX_TRANSFER_WRITE_SQL_BYTES: usize = 512 * 1024;
const MAX_SQLSERVER_INSERT_ROWS: usize = 1000;
@@ -445,7 +448,11 @@ pub fn convert_cross_family_object_ddl(
/// (with `''` and backslash escapes), MySQL double-quoted strings when
/// `double_quote_is_string` is set, `--`/`#` line comments and `/* */`
/// block comments. Returns byte ranges `(start, end)` of those spans.
fn sql_non_code_spans(sql: &str, double_quote_is_string: bool) -> Vec<(usize, usize)> {
fn sql_non_code_spans_with_mysql_identifiers(
sql: &str,
double_quote_is_string: bool,
backtick_is_identifier: bool,
) -> Vec<(usize, usize)> {
let bytes = sql.as_bytes();
let mut spans = Vec::new();
let mut i = 0;
@@ -454,6 +461,7 @@ fn sql_non_code_spans(sql: &str, double_quote_is_string: bool) -> Vec<(usize, us
let starts_non_code = match b {
b'\'' => true,
b'"' if double_quote_is_string => true,
b'`' if backtick_is_identifier => true,
b'-' if i + 1 < bytes.len() && bytes[i + 1] == b'-' => true,
b'#' => true, // MySQL line comment
b'/' if i + 1 < bytes.len() && bytes[i + 1] == b'*' => true,
@@ -465,7 +473,7 @@ fn sql_non_code_spans(sql: &str, double_quote_is_string: bool) -> Vec<(usize, us
}
let start = i;
i = match b {
b'\'' | b'"' => {
b'\'' | b'"' | b'`' => {
i += 1;
while i < bytes.len() {
if bytes[i] == b'\\' && i + 1 < bytes.len() {
@@ -503,6 +511,10 @@ fn sql_non_code_spans(sql: &str, double_quote_is_string: bool) -> Vec<(usize, us
spans
}
fn sql_non_code_spans(sql: &str, double_quote_is_string: bool) -> Vec<(usize, usize)> {
sql_non_code_spans_with_mysql_identifiers(sql, double_quote_is_string, false)
}
/// Applies `f` to every code span of `sql`; string literals and comments
/// (see `sql_non_code_spans`) pass through verbatim so rewrites never touch
/// text inside them.
@@ -526,6 +538,26 @@ where
out
}
fn map_mysql_ddl_code_spans<F>(sql: &str, mut f: F) -> String
where
F: FnMut(&str) -> String,
{
let spans = sql_non_code_spans_with_mysql_identifiers(sql, true, true);
let mut out = String::with_capacity(sql.len());
let mut prev = 0;
for (start, end) in spans {
if start > prev {
out.push_str(&f(&sql[prev..start]));
}
out.push_str(&sql[start..end]);
prev = end;
}
if prev < sql.len() {
out.push_str(&f(&sql[prev..]));
}
out
}
/// Masks string literals and comments with placeholders so subsequent regex
/// rewrites cannot touch them, and returns the restore map to swap the
/// original text back afterwards.
@@ -3290,6 +3322,52 @@ fn contains_oceanbase_mysql_table_options(sql: &str) -> bool {
OCEANBASE_MYSQL_TABLE_OPTION_RE.is_match(&sql_without_literals_or_comments)
}
fn mysql_ddl_collation_names(sql: &str) -> Vec<String> {
let mut names = Vec::new();
map_mysql_ddl_code_spans(sql, |code| {
for captures in MYSQL_COLLATE_CLAUSE_RE.captures_iter(code) {
let name = captures[1].to_string();
if !names.iter().any(|existing: &String| existing.eq_ignore_ascii_case(&name)) {
names.push(name);
}
}
String::new()
});
names
}
fn remove_unsupported_mysql_collations(sql: &str, supported: &HashSet<String>) -> String {
let supported = supported.iter().map(|name| name.to_ascii_lowercase()).collect::<HashSet<_>>();
map_mysql_ddl_code_spans(sql, |code| {
MYSQL_COLLATE_CLAUSE_RE
.replace_all(code, |captures: &regex::Captures| {
if supported.contains(&captures[1].to_ascii_lowercase()) {
captures[0].to_string()
} else {
String::new()
}
})
.to_string()
})
}
fn mysql_collations_for_transfer_ddl_recovery(
sql: &str,
error: &str,
target_db_type: &DatabaseType,
reused_source_ddl: bool,
) -> Option<Vec<String>> {
if !reused_source_ddl
|| !matches!(target_db_type, DatabaseType::Mysql)
|| !error.to_ascii_lowercase().contains("unknown collation")
|| !sql.trim_start().to_ascii_uppercase().starts_with("CREATE TABLE ")
{
return None;
}
let names = mysql_ddl_collation_names(sql);
(!names.is_empty()).then_some(names)
}
fn can_reuse_source_table_ddl(
source_db_type: &DatabaseType,
target_db_type: &DatabaseType,
@@ -4067,6 +4145,48 @@ async fn execute_transfer_ddl_on_pool(
Ok(())
}
async fn supported_mysql_transfer_collations(
state: &AppState,
pool_key: &str,
names: &[String],
) -> Result<HashSet<String>, String> {
let names = names.iter().map(|name| quote_string_literal(name)).collect::<Vec<_>>().join(", ");
let sql = format!("SELECT COLLATION_NAME FROM information_schema.COLLATIONS WHERE COLLATION_NAME IN ({names})");
let result = execute_on_pool(state, pool_key, &sql).await?;
Ok(result.rows.iter().filter_map(|row| json_string_cell(row, 0)).map(|name| name.to_ascii_lowercase()).collect())
}
async fn execute_transfer_create_table_ddl_on_pool(
state: &AppState,
pool_key: &str,
sql: &str,
db_type: &DatabaseType,
reused_source_ddl: bool,
) -> Result<(), String> {
let original_error = match execute_transfer_ddl_on_pool(state, pool_key, sql, db_type).await {
Ok(()) => return Ok(()),
Err(error) => error,
};
let Some(collations) = mysql_collations_for_transfer_ddl_recovery(sql, &original_error, db_type, reused_source_ddl)
else {
return Err(original_error);
};
let supported = supported_mysql_transfer_collations(state, pool_key, &collations)
.await
.map_err(|error| format!("{original_error}; failed to inspect target MySQL collations: {error}"))?;
let rewritten = remove_unsupported_mysql_collations(sql, &supported);
if rewritten == sql {
return Err(format!("{original_error}; target MySQL reports all referenced collations as supported"));
}
let unsupported =
collations.iter().filter(|name| !supported.contains(&name.to_ascii_lowercase())).cloned().collect::<Vec<_>>();
log::warn!("[transfer] retrying target table DDL without unsupported MySQL collations: {}", unsupported.join(", "));
execute_transfer_ddl_on_pool(state, pool_key, &rewritten, db_type)
.await
.map_err(|error| format!("{original_error}; retry without unsupported MySQL collations failed: {error}"))
}
fn transfer_table_already_exists_error(error: &str) -> bool {
let lower = error.to_ascii_lowercase();
lower.contains("already exists")
@@ -6279,8 +6399,9 @@ where
target_driver_profile.as_deref(),
preserves_target_table_name,
);
let mut reused_source_ddl = false;
let ddl = if can_reuse_source_ddl {
let source_ddl = if let Some(catalog) =
let (source_ddl, source_ddl_was_read) = if let Some(catalog) =
resolve_external_transfer_catalog(request.source_catalog.as_deref(), source_db_type)
{
// Doris/StarRocks external catalog: read DDL directly via
@@ -6295,9 +6416,38 @@ where
};
p.clone()
};
db::doris::get_catalog_table_ddl(&pool, catalog, &request.source_database, table).await
.unwrap_or_else(|err| {
match db::doris::get_catalog_table_ddl(&pool, catalog, &request.source_database, table).await {
Ok(ddl) => (ddl, true),
Err(err) => {
log::warn!("[transfer] catalog DDL read failed for {table} in catalog '{catalog}': {err}; falling back to generated DDL");
(
generate_create_table_ddl(
&columns,
&target_table,
&request.source_schema,
&request.target_schema,
target_db_type,
source_db_type,
table_comment.as_deref(),
request.target_catalog.as_deref(),
),
false,
)
}
}
} else {
match crate::schema::get_table_ddl_core(
state,
&request.source_connection_id,
&request.source_database,
&request.source_schema,
table,
None,
)
.await
{
Ok(ddl) => (ddl, true),
Err(_) => (
generate_create_table_ddl(
&columns,
&target_table,
@@ -6307,30 +6457,10 @@ where
source_db_type,
table_comment.as_deref(),
request.target_catalog.as_deref(),
)
})
} else {
crate::schema::get_table_ddl_core(
state,
&request.source_connection_id,
&request.source_database,
&request.source_schema,
table,
None,
)
.await
.unwrap_or_else(|_| {
generate_create_table_ddl(
&columns,
&target_table,
&request.source_schema,
&request.target_schema,
target_db_type,
source_db_type,
table_comment.as_deref(),
request.target_catalog.as_deref(),
)
})
),
false,
),
}
};
if contains_oceanbase_mysql_table_options(&source_ddl)
&& !is_oceanbase_mysql_profile(target_db_type, target_driver_profile.as_deref())
@@ -6346,6 +6476,7 @@ where
request.target_catalog.as_deref(),
)
} else {
reused_source_ddl = source_ddl_was_read;
rewrite_transfer_source_table_ddl(
&source_ddl,
&request.source_schema,
@@ -6368,7 +6499,14 @@ where
};
log::info!("[transfer] creating target table: {}", ddl.chars().take(200).collect::<String>());
let target_table_created = transfer_create_table_created(
execute_transfer_ddl_on_pool(state, target_pool_key, &ddl, target_db_type).await,
execute_transfer_create_table_ddl_on_pool(
state,
target_pool_key,
&ddl,
target_db_type,
reused_source_ddl,
)
.await,
"Failed to create table",
)?;
if target_table_created {
@@ -8723,6 +8861,82 @@ mod tests {
assert!(can_reuse_source_table_ddl(&DatabaseType::Mysql, &DatabaseType::Mysql, None, None, true,));
}
#[test]
fn mysql_transfer_collation_recovery_only_reads_ddl_code() {
let ddl = r#"CREATE TABLE `COLLATE utf8mb4_identifier_ci` (
`id` bigint NOT NULL,
`note` varchar(255) COMMENT 'COLLATE utf8mb4_literal_ci',
`name` varchar(64) COLLATE utf8mb4_0900_ai_ci,
`legacy` varchar(64) collate = utf8mb4_unicode_ci
) DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci
/* COLLATE utf8mb4_comment_ci */"#;
assert_eq!(
mysql_ddl_collation_names(ddl),
vec!["utf8mb4_0900_ai_ci".to_string(), "utf8mb4_unicode_ci".to_string()]
);
}
#[test]
fn mysql_transfer_collation_recovery_removes_only_unsupported_clauses() {
let ddl = r#"CREATE TABLE `items` (
`name` varchar(64) COLLATE utf8mb4_0900_ai_ci COMMENT 'COLLATE utf8mb4_0900_ai_ci',
`legacy` varchar(64) COLLATE utf8mb4_unicode_ci
) DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='items'"#;
let supported = HashSet::from(["utf8mb4_unicode_ci".to_string()]);
let rewritten = remove_unsupported_mysql_collations(ddl, &supported);
assert!(!rewritten.contains("varchar(64) COLLATE utf8mb4_0900_ai_ci"));
assert!(!rewritten.contains("utf8mb4 COLLATE=utf8mb4_0900_ai_ci"));
assert!(rewritten.contains("COLLATE utf8mb4_unicode_ci"));
assert!(rewritten.contains("COMMENT 'COLLATE utf8mb4_0900_ai_ci'"));
assert!(rewritten.contains("DEFAULT CHARSET=utf8mb4"));
assert!(rewritten.contains("COMMENT='items'"));
}
#[test]
fn mysql_transfer_collation_recovery_has_a_narrow_error_gate() {
let ddl = "CREATE TABLE `items` (`name` varchar(64) COLLATE utf8mb4_0900_ai_ci)";
assert_eq!(
mysql_collations_for_transfer_ddl_recovery(
ddl,
"ERROR 1273 (HY000): Unknown collation: 'utf8mb4_0900_ai_ci'",
&DatabaseType::Mysql,
true,
),
Some(vec!["utf8mb4_0900_ai_ci".to_string()])
);
assert_eq!(
mysql_collations_for_transfer_ddl_recovery(
ddl,
"ERROR 1064 (42000): syntax error",
&DatabaseType::Mysql,
true,
),
None
);
assert_eq!(
mysql_collations_for_transfer_ddl_recovery(
ddl,
"ERROR 1273 (HY000): Unknown collation",
&DatabaseType::Mysql,
false,
),
None
);
assert_eq!(
mysql_collations_for_transfer_ddl_recovery(
ddl,
"ERROR 1273 (HY000): Unknown collation",
&DatabaseType::Postgres,
true,
),
None
);
}
#[test]
fn detects_oceanbase_mysql_table_options_outside_literals_and_comments() {
let ddl = r#"CREATE TABLE `items` (
crates/dbx-core/tests/live_mysql_transfer.rs
+206
View File
@@ -32,6 +32,35 @@ fn live_mysql_config(id: &str) -> ConnectionConfig {
.expect("live MySQL transfer config should deserialize")
}
fn live_cross_version_mysql_config(id: &str, prefix: &str) -> ConnectionConfig {
let host = std::env::var(format!("DBX_LIVE_MYSQL_TRANSFER_{prefix}_HOST"))
.unwrap_or_else(|_| panic!("DBX_LIVE_MYSQL_TRANSFER_{prefix}_HOST"));
let port = std::env::var(format!("DBX_LIVE_MYSQL_TRANSFER_{prefix}_PORT"))
.ok()
.and_then(|value| value.parse::<u16>().ok())
.unwrap_or(3306);
let username =
std::env::var(format!("DBX_LIVE_MYSQL_TRANSFER_{prefix}_USER")).unwrap_or_else(|_| "root".to_string());
let password = std::env::var(format!("DBX_LIVE_MYSQL_TRANSFER_{prefix}_PASSWORD"))
.unwrap_or_else(|_| panic!("DBX_LIVE_MYSQL_TRANSFER_{prefix}_PASSWORD"));
serde_json::from_value(json!({
"id": id,
"name": id,
"db_type": DatabaseType::Mysql,
"host": host,
"port": port,
"username": username,
"password": password,
"database": null,
"connect_timeout_secs": 10,
"query_timeout_secs": 30,
"idle_timeout_secs": 60,
"keepalive_interval_secs": 0
}))
.expect("live cross-version MySQL transfer config should deserialize")
}
fn mysql_url(config: &ConnectionConfig) -> String {
format!("mysql://{}:{}@{}:{}", config.username, config.password, config.host, config.port)
}
@@ -68,6 +97,183 @@ async fn query_text(pool: &mysql::MySqlPool, sql: &str) -> String {
mysql::execute_query(pool, sql, false).await.unwrap().rows[0][0].as_str().unwrap().to_string()
}
#[tokio::test]
#[ignore = "requires disposable MySQL 8 source and 5.7 target endpoints via DBX_LIVE_MYSQL_TRANSFER_SOURCE_*/TARGET_* variables"]
async fn live_mysql_transfer_downgrades_unsupported_source_collations() {
let suffix = uuid::Uuid::new_v4().simple().to_string();
let source_connection_id = format!("live-mysql8-transfer-{suffix}");
let target_connection_id = format!("live-mysql57-transfer-{suffix}");
let source_database = format!("dbx_6023_src_{}", &suffix[..12]);
let target_database = format!("dbx_6023_dst_{}", &suffix[..12]);
let modern_target_database = format!("dbx_6023_modern_{}", &suffix[..12]);
let source_config = live_cross_version_mysql_config(&source_connection_id, "SOURCE");
let target_config = live_cross_version_mysql_config(&target_connection_id, "TARGET");
let source_setup_pool = mysql::connect(&mysql_url(&source_config), Duration::from_secs(10)).await.unwrap();
let target_setup_pool = mysql::connect(&mysql_url(&target_config), Duration::from_secs(10)).await.unwrap();
mysql::execute_query(
&source_setup_pool,
&format!(
"CREATE DATABASE `{source_database}` CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci;\
CREATE DATABASE `{modern_target_database}` CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci;\
CREATE TABLE `{source_database}`.`transfer_probe` (\
id BIGINT NOT NULL AUTO_INCREMENT,\
code VARCHAR(64) COLLATE utf8mb4_0900_ai_ci NOT NULL DEFAULT 'x' COMMENT 'probe',\
legacy VARCHAR(64) COLLATE utf8mb4_unicode_ci NOT NULL,\
normalized VARCHAR(64) GENERATED ALWAYS AS (lower(code)) STORED,\
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,\
PRIMARY KEY (id), UNIQUE KEY uk_code (code), KEY idx_legacy (legacy)\
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='issue 6023';\
INSERT INTO `{source_database}`.`transfer_probe` (code, legacy) VALUES ('AbC', 'legacy')"
),
true,
)
.await
.unwrap();
mysql::execute_query(
&target_setup_pool,
&format!("CREATE DATABASE `{target_database}` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"),
false,
)
.await
.unwrap();
let task_tmp = std::path::PathBuf::from(
std::env::var("DBX_LIVE_MYSQL_TRANSFER_TMP_DIR").expect("DBX_LIVE_MYSQL_TRANSFER_TMP_DIR"),
);
let dir = task_tmp.join(format!("live-mysql-collation-transfer-{suffix}"));
std::fs::create_dir_all(&dir).unwrap();
let storage = Storage::open(&dir.join("storage.db")).await.unwrap();
let state = AppState::new(storage);
state.configs.write().await.insert(source_connection_id.clone(), source_config);
state.configs.write().await.insert(target_connection_id.clone(), target_config);
let source_pool_key = state.get_or_create_pool(&source_connection_id, Some(&source_database)).await.unwrap();
let target_pool_key = state.get_or_create_pool(&target_connection_id, Some(&target_database)).await.unwrap();
let modern_target_pool_key =
state.get_or_create_pool(&source_connection_id, Some(&modern_target_database)).await.unwrap();
let request = TransferRequest {
transfer_id: format!("live-mysql-collation-transfer-{suffix}"),
source_connection_id,
source_database: source_database.clone(),
source_schema: source_database.clone(),
source_catalog: None,
target_connection_id,
target_database: target_database.clone(),
target_schema: target_database.clone(),
target_catalog: None,
tables: vec!["transfer_probe".to_string()],
create_table: true,
content: TransferContent::default(),
objects: Vec::new(),
mode: TransferMode::Append,
target_table_name_case: TransferTableNameCase::Preserve,
ownership_policy: TransferOwnershipPolicy::Preserve,
batch_size: 10,
};
let test_result = async {
let transferred = transfer_table(
&state,
&request,
"transfer_probe",
0,
&DatabaseType::Mysql,
&DatabaseType::Mysql,
&source_pool_key,
&target_pool_key,
|_| {},
)
.await?;
assert_eq!(transferred, 1);
let ddl_result = mysql::execute_query(
&target_setup_pool,
&format!("SHOW CREATE TABLE `{target_database}`.`transfer_probe`"),
false,
)
.await?;
let target_ddl = ddl_result.rows[0][1].as_str().unwrap();
assert!(!target_ddl.contains("utf8mb4_0900_ai_ci"), "ddl: {target_ddl}");
assert!(target_ddl.contains("COLLATE utf8mb4_unicode_ci"), "ddl: {target_ddl}");
assert!(target_ddl.contains("AUTO_INCREMENT"), "ddl: {target_ddl}");
assert!(target_ddl.contains("GENERATED ALWAYS AS"), "ddl: {target_ddl}");
assert!(target_ddl.contains("ON UPDATE CURRENT_TIMESTAMP"), "ddl: {target_ddl}");
assert!(target_ddl.contains("UNIQUE KEY `uk_code`"), "ddl: {target_ddl}");
assert!(target_ddl.contains("KEY `idx_legacy`"), "ddl: {target_ddl}");
assert!(target_ddl.contains("COMMENT='issue 6023'"), "ddl: {target_ddl}");
assert_eq!(
query_text(
&target_setup_pool,
&format!("SELECT CONCAT(code, ':', legacy, ':', normalized) FROM `{target_database}`.`transfer_probe`"),
)
.await,
"AbC:legacy:abc"
);
let modern_request = TransferRequest {
transfer_id: format!("live-mysql-modern-collation-transfer-{suffix}"),
source_connection_id: request.source_connection_id.clone(),
source_database: source_database.clone(),
source_schema: source_database.clone(),
source_catalog: None,
target_connection_id: request.source_connection_id.clone(),
target_database: modern_target_database.clone(),
target_schema: modern_target_database.clone(),
target_catalog: None,
tables: vec!["transfer_probe".to_string()],
create_table: true,
content: TransferContent::default(),
objects: Vec::new(),
mode: TransferMode::Append,
target_table_name_case: TransferTableNameCase::Preserve,
ownership_policy: TransferOwnershipPolicy::Preserve,
batch_size: 10,
};
assert_eq!(
transfer_table(
&state,
&modern_request,
"transfer_probe",
0,
&DatabaseType::Mysql,
&DatabaseType::Mysql,
&source_pool_key,
&modern_target_pool_key,
|_| {},
)
.await?,
1
);
let modern_ddl = mysql::execute_query(
&source_setup_pool,
&format!("SHOW CREATE TABLE `{modern_target_database}`.`transfer_probe`"),
false,
)
.await?
.rows[0][1]
.as_str()
.unwrap()
.to_string();
assert!(modern_ddl.contains("utf8mb4_0900_ai_ci"), "ddl: {modern_ddl}");
Ok::<_, String>(())
}
.await;
let source_cleanup = mysql::execute_query(
&source_setup_pool,
&format!("DROP DATABASE `{source_database}`; DROP DATABASE `{modern_target_database}`"),
true,
)
.await;
let target_cleanup =
mysql::execute_query(&target_setup_pool, &format!("DROP DATABASE `{target_database}`"), false).await;
source_setup_pool.disconnect().await.unwrap();
target_setup_pool.disconnect().await.unwrap();
let _ = std::fs::remove_dir_all(dir);
source_cleanup.unwrap();
target_cleanup.unwrap();
test_result.unwrap();
}
#[tokio::test]
#[ignore = "requires a disposable MySQL 5.7+ endpoint via DBX_LIVE_MYSQL_TRANSFER_* variables"]
async fn live_mysql_transfer_preserves_spatial_values_and_modes() {