xiaoxue_admin
/
dbx
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(mysql): preserve functional index expressions

This commit is contained in:
t8y2 2026-07-26 09:09:22 +08:00
parent ae9403acc4
commit 0c7e3577bd
No known key found for this signature in database
4 changed files with 218 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

114
crates/dbx-core/src/db/mysql.rs
View File

@ -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<Vec<IndexInfo>, 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<mysql_async::Row> = 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::<bool, &str>("is_unique").unwrap_or(false),
is_primary: row.get::<bool, &str>("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<Vec<IndexInfo>, 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<String, String> {
@ -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 =

84
crates/dbx-core/src/schema_diff.rs
View File

@ -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::<Vec<_>>().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::<Vec<_>>()
.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(

22
crates/dbx-core/src/table_structure_sql/indexes.rs
View File

@ -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<DatabaseType>,
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::<Vec<_>>().join(", ");
let cols = columns
.iter()
.map(|column| {
if dialect == StructureDialect::Mysql {
mysql_index_column_sql(column)
} else {
quote_ident(dialect, column)
}
})
.collect::<Vec<_>>()
.join(", ");
let idx_type = normalized_index_type(index);
let mut type_prefix = String::new();
let mut using_clause = String::new();

25
crates/dbx-core/src/table_structure_sql/tests.rs
View File

@ -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::<String>::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");