diff --git a/crates/dbx-core/src/transfer.rs b/crates/dbx-core/src/transfer.rs index 0c8ab66398..efc48364cd 100644 --- a/crates/dbx-core/src/transfer.rs +++ b/crates/dbx-core/src/transfer.rs @@ -18,6 +18,9 @@ static OCEANBASE_MYSQL_TABLE_OPTION_RE: std::sync::LazyLock = 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 = 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(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 { + 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 { + let supported = supported.iter().map(|name| name.to_ascii_lowercase()).collect::>(); + map_mysql_ddl_code_spans(sql, |code| { + MYSQL_COLLATE_CLAUSE_RE + .replace_all(code, |captures: ®ex::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> { + 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, String> { + let names = names.iter().map(|name| quote_string_literal(name)).collect::>().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::>(); + 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::()); 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` ( diff --git a/crates/dbx-core/tests/live_mysql_transfer.rs b/crates/dbx-core/tests/live_mysql_transfer.rs index 59ef648fb5..9482d224e7 100644 --- a/crates/dbx-core/tests/live_mysql_transfer.rs +++ b/crates/dbx-core/tests/live_mysql_transfer.rs @@ -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::().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() {