From 0c7e3577bdfdfeee23c967e53f768081ff564164 Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Sun, 26 Jul 2026 09:09:22 +0800 Subject: [PATCH] fix(mysql): preserve functional index expressions --- crates/dbx-core/src/db/mysql.rs | 114 ++++++++++++++---- crates/dbx-core/src/schema_diff.rs | 84 ++++++++++++- .../src/table_structure_sql/indexes.rs | 22 +++- .../dbx-core/src/table_structure_sql/tests.rs | 25 ++++ 4 files changed, 218 insertions(+), 27 deletions(-) diff --git a/crates/dbx-core/src/db/mysql.rs b/crates/dbx-core/src/db/mysql.rs index 2ae625f809..f1883d27c0 100644 --- a/crates/dbx-core/src/db/mysql.rs +++ b/crates/dbx-core/src/db/mysql.rs @@ -3799,38 +3799,75 @@ fn skip_sql_whitespace_and_comments(bytes: &[u8], mut i: usize) -> usize { } } -pub async fn list_indexes(pool: &MySqlPool, database: &str, table: &str) -> Result, String> { - let sql = format!( - "SELECT INDEX_NAME, GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX) AS columns, \ - MIN(NON_UNIQUE) = 0 AS is_unique, INDEX_NAME = 'PRIMARY' AS is_primary, \ - INDEX_TYPE, MAX(NULLIF(INDEX_COMMENT, '')) AS INDEX_COMMENT \ +fn mysql_list_indexes_sql(database: &str, table: &str, include_expression: bool) -> String { + let expression_column = if include_expression { "EXPRESSION, " } else { "" }; + format!( + "SELECT INDEX_NAME, COLUMN_NAME, {expression_column}SEQ_IN_INDEX, NON_UNIQUE, INDEX_TYPE, INDEX_COMMENT \ FROM information_schema.STATISTICS \ WHERE TABLE_SCHEMA = {} AND TABLE_NAME = {} \ - GROUP BY INDEX_NAME, INDEX_TYPE \ - ORDER BY INDEX_NAME", + ORDER BY INDEX_NAME, SEQ_IN_INDEX", quote_value(database), quote_value(table), - ); - let mut conn = get_conn_with_timeout(pool, super::connection_timeout()).await?; - let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?; - let rows: Vec = result.collect_and_drop().await.map_err(|e| e.to_string())?; + ) +} - Ok(rows - .iter() - .map(|row| { - let cols_str = get_str_by_name(row, "columns"); - IndexInfo { - name: get_str_by_name(row, "INDEX_NAME"), - columns: cols_str.split(',').filter(|s| !s.is_empty()).map(|s| s.to_string()).collect(), - is_unique: row.get::("is_unique").unwrap_or(false), - is_primary: row.get::("is_primary").unwrap_or(false), - filter: None, - index_type: Some(get_str_by_name(row, "INDEX_TYPE")), - included_columns: None, - comment: get_opt_str(row, "INDEX_COMMENT").filter(|value| !value.is_empty()), +fn mysql_statistics_expression_is_unsupported(error: &mysql_async::Error) -> bool { + matches!(error, mysql_async::Error::Server(server_error) if server_error.code == 1054) +} + +pub async fn list_indexes(pool: &MySqlPool, database: &str, table: &str) -> Result, String> { + let mut conn = get_conn_with_timeout(pool, super::connection_timeout()).await?; + let expression_sql = mysql_list_indexes_sql(database, table, true); + let legacy_sql = mysql_list_indexes_sql(database, table, false); + let (result, include_expression) = match conn.query_iter(&expression_sql).await { + Ok(result) => (result, true), + Err(error) if mysql_statistics_expression_is_unsupported(&error) => { + // MySQL 5.7 and older compatible servers do not expose EXPRESSION; keep the legacy metadata path. + log::debug!("MySQL index expressions are unavailable, retrying without EXPRESSION: {error}"); + (conn.query_iter(&legacy_sql).await.map_err(|e| e.to_string())?, false) + } + Err(error) => return Err(error.to_string()), + }; + let mut indexes = Vec::new(); + let mut index_positions = HashMap::new(); + + result + .for_each_and_drop(|row| { + let name = get_str_by_name(&row, "INDEX_NAME"); + let index_position = if let Some(index_position) = index_positions.get(&name) { + *index_position + } else { + let index_position = indexes.len(); + index_positions.insert(name.clone(), index_position); + indexes.push(IndexInfo { + name: name.clone(), + columns: Vec::new(), + is_unique: get_opt_i32(&row, "NON_UNIQUE").unwrap_or(1) == 0, + is_primary: name == "PRIMARY", + filter: None, + index_type: Some(get_str_by_name(&row, "INDEX_TYPE")), + included_columns: None, + comment: get_opt_str(&row, "INDEX_COMMENT").filter(|value| !value.is_empty()), + }); + index_position + }; + + let index_part = if include_expression { + get_opt_str(&row, "EXPRESSION") + .filter(|value| !value.trim().is_empty()) + .map(|expression| format!("({})", expression.trim())) + .or_else(|| get_opt_str(&row, "COLUMN_NAME").filter(|value| !value.is_empty())) + } else { + get_opt_str(&row, "COLUMN_NAME").filter(|value| !value.is_empty()) + }; + if let Some(index_part) = index_part { + indexes[index_position].columns.push(index_part); } }) - .collect()) + .await + .map_err(|e| e.to_string())?; + + Ok(indexes) } pub async fn show_create_table_ddl(pool: &MySqlPool, database: &str, table: &str) -> Result { @@ -4909,6 +4946,33 @@ mod tests { ); } + #[test] + fn mysql_index_metadata_query_has_expression_compatibility_fallback() { + let with_expression = mysql_list_indexes_sql("db", "users", true); + assert!(with_expression.contains("EXPRESSION, SEQ_IN_INDEX")); + + let without_expression = mysql_list_indexes_sql("db", "users", false); + assert!(!without_expression.contains("EXPRESSION")); + assert!(without_expression.contains("ORDER BY INDEX_NAME, SEQ_IN_INDEX")); + } + + #[test] + fn mysql_index_metadata_falls_back_only_for_unknown_expression_column() { + let unsupported = mysql_async::Error::Server(mysql_async::ServerError { + code: 1054, + message: "Unknown column 'EXPRESSION'".to_string(), + state: "42S22".to_string(), + }); + let permission_denied = mysql_async::Error::Server(mysql_async::ServerError { + code: 1044, + message: "Access denied".to_string(), + state: "42000".to_string(), + }); + + assert!(mysql_statistics_expression_is_unsupported(&unsupported)); + assert!(!mysql_statistics_expression_is_unsupported(&permission_denied)); + } + #[test] fn mysql_group_concat_not_supported_error_retries_without_session_variable() { let error = diff --git a/crates/dbx-core/src/schema_diff.rs b/crates/dbx-core/src/schema_diff.rs index 71795dd4d1..65aa52e47b 100644 --- a/crates/dbx-core/src/schema_diff.rs +++ b/crates/dbx-core/src/schema_diff.rs @@ -1116,9 +1116,31 @@ fn drop_index_sql(table_name: &str, index_name: &str, db_type: DatabaseType, sch } } +fn mysql_index_column_sql(column: &str) -> String { + let trimmed = column.trim(); + // MySQL metadata represents a functional key part as an expression wrapped for CREATE INDEX. + if trimmed.starts_with("((") && trimmed.ends_with("))") { + trimmed.to_string() + } else { + quote_id(column, DatabaseType::Mysql) + } +} + fn create_index_sql(table_name: &str, index: &IndexInfo, db_type: DatabaseType, schema: Option<&str>) -> String { let table = qualified_name(table_name, db_type, schema); - let columns = index.columns.iter().map(|column| quote_id(column, db_type)).collect::>().join(", "); + let columns = + index + .columns + .iter() + .map(|column| { + if db_type == DatabaseType::Mysql { + mysql_index_column_sql(column) + } else { + quote_id(column, db_type) + } + }) + .collect::>() + .join(", "); let unique = if index.is_unique { "UNIQUE " } else { "" }; let index_type = index.index_type.as_deref().unwrap_or_default(); let using_clause = if !index_type.is_empty() && db_type == DatabaseType::Postgres { @@ -1754,6 +1776,66 @@ mod tests { assert_eq!(diffs[0].changes, vec!["unique: YES → NO", "columns: status → status, created_at"]); } + #[test] + fn detects_mysql_functional_index_changes_and_preserves_expression_ddl() { + let functional_key_part = "((case when (`STATUS` = _utf8mb4'online') then _utf8mb4'online' else NULL end))"; + let source_index = index(IndexInfo { + name: "test_UNIQUE".to_string(), + columns: vec!["attr".to_string(), "attr2".to_string(), functional_key_part.to_string()], + is_unique: true, + is_primary: false, + filter: None, + index_type: None, + included_columns: None, + comment: None, + }); + let target_index = index(IndexInfo { + name: "test_UNIQUE".to_string(), + columns: vec!["attr".to_string(), "attr2".to_string()], + is_unique: true, + is_primary: false, + filter: None, + index_type: None, + included_columns: None, + comment: None, + }); + + let diffs = diff_indexes(&[source_index.clone()], &[target_index]); + assert_eq!(diffs.len(), 1); + assert_eq!(diffs[0].diff_type, "modified"); + assert_eq!(diffs[0].changes, vec![format!("columns: attr, attr2 → attr, attr2, {functional_key_part}")]); + + let sql = generate_schema_sync_sql( + &[TableDiff { + diff_type: "modified".to_string(), + object_type: Some("table".to_string()), + name: "test".to_string(), + columns: None, + indexes: Some(diffs), + foreign_keys: None, + triggers: None, + ddl: None, + target_ddl: None, + source_table_comment: None, + target_table_comment: None, + sync_sql: None, + }], + &[], + &[], + &[], + &[], + DatabaseType::Mysql, + Some("dbx_issue_4114"), + false, + ); + + assert!(sql.contains("DROP INDEX `test_UNIQUE` ON `dbx_issue_4114`.`test`;")); + assert!(sql.contains(&format!( + "CREATE UNIQUE INDEX `test_UNIQUE` ON `dbx_issue_4114`.`test` (`attr`, `attr2`, {functional_key_part});" + ))); + assert!(!sql.contains("`((case")); + } + #[test] fn detects_foreign_key_additions_removals_and_target_changes() { let diffs = diff_foreign_keys( diff --git a/crates/dbx-core/src/table_structure_sql/indexes.rs b/crates/dbx-core/src/table_structure_sql/indexes.rs index f60bd53272..9525d98d05 100644 --- a/crates/dbx-core/src/table_structure_sql/indexes.rs +++ b/crates/dbx-core/src/table_structure_sql/indexes.rs @@ -116,6 +116,16 @@ pub(super) fn mysql_index_parts(index_type: &str) -> (String, String) { } } +fn mysql_index_column_sql(column: &str) -> String { + let trimmed = column.trim(); + // Keep the wrapped expression from MySQL metadata instead of quoting it as a column identifier. + if trimmed.starts_with("((") && trimmed.ends_with("))") { + trimmed.to_string() + } else { + quote_ident(StructureDialect::Mysql, column) + } +} + pub(super) fn build_drop_index_sql( database_type: Option, dialect: StructureDialect, @@ -160,7 +170,17 @@ pub(super) fn build_create_index_statements( } let unique = if index.is_unique { "UNIQUE " } else { "" }; - let cols = columns.iter().map(|column| quote_ident(dialect, column)).collect::>().join(", "); + let cols = columns + .iter() + .map(|column| { + if dialect == StructureDialect::Mysql { + mysql_index_column_sql(column) + } else { + quote_ident(dialect, column) + } + }) + .collect::>() + .join(", "); let idx_type = normalized_index_type(index); let mut type_prefix = String::new(); let mut using_clause = String::new(); diff --git a/crates/dbx-core/src/table_structure_sql/tests.rs b/crates/dbx-core/src/table_structure_sql/tests.rs index 2fbee20565..84964e3805 100644 --- a/crates/dbx-core/src/table_structure_sql/tests.rs +++ b/crates/dbx-core/src/table_structure_sql/tests.rs @@ -1240,6 +1240,31 @@ fn mysql_create_unique_index_with_comment_and_btree() { ); } +#[test] +fn mysql_create_functional_index_preserves_key_part_syntax() { + let functional_key_part = "((case when (`STATUS` = _utf8mb4'online') then _utf8mb4'online' else NULL end))"; + let mut idx = index("test_UNIQUE", &["attr", "attr2", functional_key_part]); + idx.is_unique = true; + + let result = build_table_structure_change_sql(TableStructureSqlOptions { + database_type: Some(DatabaseType::Mysql), + schema: None, + table_name: "test".to_string(), + columns: Vec::new(), + indexes: vec![idx], + foreign_keys: Vec::new(), + triggers: Vec::new(), + table_comment: None, + original_table_comment: None, + }); + + assert_eq!(result.warnings, Vec::::new()); + assert_eq!( + result.statements, + vec![format!("CREATE UNIQUE INDEX `test_UNIQUE` ON `test` (`attr`, `attr2`, {functional_key_part});")] + ); +} + #[test] fn mysql_add_timestamp_column_drops_invalid_precision() { let mut created_at = column("created_at");