fix(table-editor): coalesce MySQL auto-increment key changes

Closes #6873

crates/dbx-core/src/table_structure_sql/column_alter.rs

@@ -277,6 +277,11 @@ pub(super) fn build_mysql_existing_column_sql(
     column: &EditableStructureColumn,
     position_clause: &str,
 ) -> Vec<String> {
+    let operation = build_mysql_existing_column_clause(column, position_clause);
+    vec![format!("ALTER TABLE {table} {operation};")]
+}
+
+pub(super) fn build_mysql_existing_column_clause(column: &EditableStructureColumn, position_clause: &str) -> String {
     let original_name = column.original.as_ref().map(|original| original.name.as_str()).unwrap_or(&column.name);
     let operation = if column.name == original_name {
         format!("MODIFY COLUMN {}", column_definition(StructureDialect::Mysql, column))
@@ -287,7 +292,7 @@ pub(super) fn build_mysql_existing_column_sql(
             column_definition(StructureDialect::Mysql, column)
         )
     };
-    vec![format!("ALTER TABLE {table} {operation}{position_clause};")]
+    format!("{operation}{position_clause}")
 }
 
 pub(super) fn build_doris_existing_column_sql(

crates/dbx-core/src/table_structure_sql/columns.rs

@@ -1,6 +1,6 @@
 use super::column_alter::{
     build_clickhouse_existing_column_sql, build_doris_existing_column_sql, build_h2_existing_column_sql,
-    build_informix_existing_column_sql, build_iris_existing_column_sql, build_mysql_existing_column_sql,
+    build_informix_existing_column_sql, build_iris_existing_column_sql, build_mysql_existing_column_clause,
     build_oracle_like_existing_column_sql, build_oscar_existing_column_sql, build_postgres_existing_column_sql,
     build_questdb_existing_column_sql, build_sqlite_existing_column_sql, build_sqlserver_existing_column_sql,
     build_xugu_existing_column_sql, has_column_extra_change, has_existing_column_attribute_change,
@@ -44,6 +44,17 @@ pub(super) fn build_column_sql(options: &TableStructureSqlOptions, warnings: &mu
         } else {
             HashSet::new()
         };
+    let mysql_primary_key_change = if options.database_type == Some(DatabaseType::Mysql) && !options.is_gaussdb_m_mode {
+        primary_key_change(options)
+    } else {
+        None
+    };
+    // MySQL validates AUTO_INCREMENT against the statement's final index layout. Keep the
+    // dependent column and key clauses together so no implicit commit exposes an invalid middle state.
+    let mysql_coalesced_primary_key_change = mysql_primary_key_change
+        .as_ref()
+        .filter(|change| options.columns.iter().any(|column| mysql_auto_increment_touches_primary_key(column, change)));
+    let mut mysql_primary_key_column_clauses = Vec::new();
     let mut statements = Vec::new();
 
     for column in &options.columns {
@@ -114,16 +125,25 @@ pub(super) fn build_column_sql(options: &TableStructureSqlOptions, warnings: &mu
                     column.name
                 ));
             }
-            statements.extend(build_add_column_sql(
-                dialect,
-                options.database_type,
-                capabilities.comment,
-                &table,
-                column,
-                &position_clause,
-                options.schema.as_deref(),
-                &options.table_name,
-            ));
+            if mysql_coalesced_primary_key_change
+                .is_some_and(|change| mysql_auto_increment_touches_primary_key(column, change))
+            {
+                mysql_primary_key_column_clauses.push(format!(
+                    "ADD COLUMN {}{position_clause}",
+                    column_definition(StructureDialect::Mysql, column)
+                ));
+            } else {
+                statements.extend(build_add_column_sql(
+                    dialect,
+                    options.database_type,
+                    capabilities.comment,
+                    &table,
+                    column,
+                    &position_clause,
+                    options.schema.as_deref(),
+                    &options.table_name,
+                ));
+            }
             if has_original_column_positions
                 && matches!(dialect, StructureDialect::Mysql | StructureDialect::ClickHouse)
             {
@@ -182,11 +202,17 @@ pub(super) fn build_column_sql(options: &TableStructureSqlOptions, warnings: &mu
         }
 
         match dialect {
-            StructureDialect::Mysql => statements.extend(build_mysql_existing_column_sql(
-                &table,
-                column,
-                if has_position_change { &position_clause } else { "" },
-            )),
+            StructureDialect::Mysql => {
+                let clause =
+                    build_mysql_existing_column_clause(column, if has_position_change { &position_clause } else { "" });
+                if mysql_coalesced_primary_key_change
+                    .is_some_and(|change| mysql_auto_increment_touches_primary_key(column, change))
+                {
+                    mysql_primary_key_column_clauses.push(clause);
+                } else {
+                    statements.push(format!("ALTER TABLE {table} {clause};"));
+                }
+            }
             StructureDialect::Doris => statements.extend(build_doris_existing_column_sql(&table, column, "")),
             StructureDialect::Postgres => statements.extend(build_postgres_existing_column_sql(&table, column)),
             StructureDialect::Oracle | StructureDialect::Dameng => {
@@ -223,9 +249,27 @@ pub(super) fn build_column_sql(options: &TableStructureSqlOptions, warnings: &mu
         }
     }
 
-    // Keep the existing key while column DDL validates. This avoids leaving a table
-    // without a key when an incoming key column cannot be made valid.
-    statements.extend(build_primary_key_sql(options, dialect, &table, warnings));
+    if let Some(change) = mysql_coalesced_primary_key_change {
+        let mut clauses = Vec::new();
+        if !change.old_ids.is_empty() {
+            clauses.push("DROP PRIMARY KEY".to_string());
+        }
+        clauses.append(&mut mysql_primary_key_column_clauses);
+        if !change.new_names.is_empty() {
+            let pk_list = change
+                .new_names
+                .iter()
+                .map(|name| quote_ident(StructureDialect::Mysql, name))
+                .collect::<Vec<_>>()
+                .join(", ");
+            clauses.push(format!("ADD PRIMARY KEY ({pk_list})"));
+        }
+        statements.push(format!("ALTER TABLE {table} {};", clauses.join(", ")));
+    } else {
+        // Keep the existing key while column DDL validates. This avoids leaving a table
+        // without a key when an incoming key column cannot be made valid.
+        statements.extend(build_primary_key_sql(options, dialect, &table, warnings));
+    }
 
     statements
 }
@@ -239,6 +283,47 @@ fn appears_in_add_primary_key(column: &EditableStructureColumn) -> bool {
     column.is_primary_key && !column.marked_for_drop
 }
 
+struct PrimaryKeyChange<'a> {
+    old_ids: HashSet<&'a str>,
+    new_ids: HashSet<&'a str>,
+    new_names: Vec<&'a str>,
+}
+
+fn primary_key_change(options: &TableStructureSqlOptions) -> Option<PrimaryKeyChange<'_>> {
+    if options.columns.iter().any(|column| column.marked_for_drop && was_primary_key_column(column)) {
+        return None;
+    }
+
+    let old_ids: HashSet<&str> = options
+        .columns
+        .iter()
+        .filter(|column| was_primary_key_column(column))
+        .map(|column| column.id.as_str())
+        .collect();
+    let new_ids: HashSet<&str> = options
+        .columns
+        .iter()
+        .filter(|column| appears_in_add_primary_key(column))
+        .map(|column| column.id.as_str())
+        .collect();
+    if old_ids == new_ids {
+        return None;
+    }
+
+    let new_names = options
+        .columns
+        .iter()
+        .filter(|column| appears_in_add_primary_key(column))
+        .map(|column| column.name.as_str())
+        .collect();
+    Some(PrimaryKeyChange { old_ids, new_ids, new_names })
+}
+
+fn mysql_auto_increment_touches_primary_key(column: &EditableStructureColumn, change: &PrimaryKeyChange<'_>) -> bool {
+    column.extra.as_ref().is_some_and(|extra| extra.auto_increment.unwrap_or(false))
+        && (change.old_ids.contains(column.id.as_str()) || change.new_ids.contains(column.id.as_str()))
+}
+
 pub(super) fn build_primary_key_sql(
     options: &TableStructureSqlOptions,
     dialect: StructureDialect,
@@ -247,23 +332,9 @@ pub(super) fn build_primary_key_sql(
 ) -> Vec<String> {
     let capabilities = capabilities_for(options.database_type);
 
-    // A draft cannot drop a primary-key column. The column pass emits the user-facing
-    // warning; keep the entire PK change empty so another checked column cannot
-    // turn that invalid draft into a partial DROP/ADD constraint change.
-    if options.columns.iter().any(|column| column.marked_for_drop && was_primary_key_column(column)) {
-        return Vec::new();
-    }
-
     // Membership by draft id (set equality): pure rename / local reorder of the same key
-    // columns is not a PK change. ADD still lists columns in table order.
-    let old_pk_ids: HashSet<&str> =
-        options.columns.iter().filter(|c| was_primary_key_column(c)).map(|c| c.id.as_str()).collect();
-    let new_pk_ids: HashSet<&str> =
-        options.columns.iter().filter(|c| appears_in_add_primary_key(c)).map(|c| c.id.as_str()).collect();
-
-    if old_pk_ids == new_pk_ids {
-        return Vec::new();
-    }
+    // columns is not a PK change. A draft that drops a PK column is also rejected here.
+    let Some(change) = primary_key_change(options) else { return Vec::new() };
 
     if !capabilities.alter_primary_key {
         warnings.push(format!(
@@ -274,7 +345,7 @@ pub(super) fn build_primary_key_sql(
     }
 
     let persisted_postgres_primary_key_name = if options.database_type == Some(DatabaseType::Postgres)
-        && !old_pk_ids.is_empty()
+        && !change.old_ids.is_empty()
     {
         let mut primary_indexes =
             options.indexes.iter().filter_map(|index| index.original.as_ref().filter(|original| original.is_primary));
@@ -293,7 +364,7 @@ pub(super) fn build_primary_key_sql(
     };
 
     let mut statements = Vec::new();
-    if !old_pk_ids.is_empty() {
+    if !change.old_ids.is_empty() {
         let Some(drop_sql) = drop_primary_key_statement(dialect, table, options, persisted_postgres_primary_key_name)
         else {
             warnings.push(format!(
@@ -305,10 +376,8 @@ pub(super) fn build_primary_key_sql(
         statements.push(drop_sql);
     }
 
-    let new_pk_names: Vec<&str> =
-        options.columns.iter().filter(|c| appears_in_add_primary_key(c)).map(|c| c.name.as_str()).collect();
-    if !new_pk_names.is_empty() {
-        let pk_list = new_pk_names.iter().map(|n| quote_ident(dialect, n)).collect::<Vec<_>>().join(", ");
+    if !change.new_names.is_empty() {
+        let pk_list = change.new_names.iter().map(|name| quote_ident(dialect, name)).collect::<Vec<_>>().join(", ");
         // DM8: ADD [CONSTRAINT name] PRIMARY KEY; anonymous form matches Navicat/DBeaver/MySQL editors.
         let constraint = persisted_postgres_primary_key_name
             .map(|name| format!("CONSTRAINT {} ", quote_ident(dialect, name)))

crates/dbx-core/src/table_structure_sql/tests.rs

@@ -3464,6 +3464,209 @@ fn builds_mysql_alter_table_change_primary_key() {
     );
 }
 
+#[test]
+fn mysql_coalesces_auto_increment_primary_key_migration() {
+    let mut old_pk = existing_pk_column("campaign_rel_id", "bigint(20)", true, false);
+    old_pk.id = "old_campaign_rel_id".to_string();
+
+    let mut id = existing_pk_column("id", "bigint(20)", false, true);
+    id.id = "new_id".to_string();
+    id.comment = "自增主键".to_string();
+    id.extra = Some(ColumnExtra { auto_increment: Some(true), ..Default::default() });
+
+    let mut options = structure_change_options(DatabaseType::Mysql, None, "tbl_gy_campaign_rel", vec![old_pk, id]);
+    options.indexes = vec![existing_index("campaign_rel_id_UNIQUE", &["campaign_rel_id"], true)];
+
+    let result = build_table_structure_change_sql(options);
+
+    assert_eq!(result.warnings, Vec::<String>::new());
+    assert_eq!(
+        result.statements,
+        vec![
+            "ALTER TABLE `tbl_gy_campaign_rel` DROP PRIMARY KEY, MODIFY COLUMN `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '自增主键', ADD PRIMARY KEY (`id`);"
+        ]
+    );
+}
+
+#[test]
+fn mysql_coalesces_new_auto_increment_primary_key_column() {
+    let mut old_pk = existing_pk_column("legacy_id", "bigint", true, false);
+    old_pk.id = "old_legacy_id".to_string();
+
+    let mut id = column("id");
+    id.data_type = "bigint".to_string();
+    id.is_nullable = false;
+    id.is_primary_key = true;
+    id.extra = Some(ColumnExtra { auto_increment: Some(true), ..Default::default() });
+
+    let result = build_table_structure_change_sql(structure_change_options(
+        DatabaseType::Mysql,
+        None,
+        "users",
+        vec![old_pk, id],
+    ));
+
+    assert_eq!(result.warnings, Vec::<String>::new());
+    assert_eq!(
+        result.statements,
+        vec![
+            "ALTER TABLE `users` DROP PRIMARY KEY, ADD COLUMN `id` bigint NOT NULL AUTO_INCREMENT, ADD PRIMARY KEY (`id`);"
+        ]
+    );
+}
+
+#[test]
+fn mysql_stable_primary_key_auto_increment_changes_keep_column_only_alter() {
+    let mut enable = existing_pk_column("id", "bigint", true, true);
+    enable.extra = Some(ColumnExtra { auto_increment: Some(true), ..Default::default() });
+
+    let enabled =
+        build_table_structure_change_sql(structure_change_options(DatabaseType::Mysql, None, "users", vec![enable]));
+    assert_eq!(enabled.warnings, Vec::<String>::new());
+    assert_eq!(enabled.statements, vec!["ALTER TABLE `users` MODIFY COLUMN `id` bigint NOT NULL AUTO_INCREMENT;"]);
+
+    let mut disable = existing_pk_column("id", "bigint", true, true);
+    disable.extra = Some(ColumnExtra::default());
+    disable.original.as_mut().unwrap().extra = Some("auto_increment".to_string());
+
+    let disabled =
+        build_table_structure_change_sql(structure_change_options(DatabaseType::Mysql, None, "users", vec![disable]));
+    assert_eq!(disabled.warnings, Vec::<String>::new());
+    assert_eq!(disabled.statements, vec!["ALTER TABLE `users` MODIFY COLUMN `id` bigint NOT NULL;"]);
+}
+
+#[test]
+fn mysql_coalesces_migration_away_from_existing_auto_increment_primary_key() {
+    let mut old_pk = existing_pk_column("id", "bigint", true, false);
+    old_pk.id = "old_id".to_string();
+    old_pk.extra = Some(ColumnExtra { auto_increment: Some(true), ..Default::default() });
+    old_pk.original.as_mut().unwrap().extra = Some("auto_increment".to_string());
+
+    let mut new_pk = existing_pk_column("external_id", "varchar(64)", false, true);
+    new_pk.id = "new_external_id".to_string();
+
+    let result = build_table_structure_change_sql(structure_change_options(
+        DatabaseType::Mysql,
+        None,
+        "users",
+        vec![old_pk, new_pk],
+    ));
+
+    assert_eq!(result.warnings, Vec::<String>::new());
+    assert_eq!(result.statements, vec!["ALTER TABLE `users` DROP PRIMARY KEY, ADD PRIMARY KEY (`external_id`);"]);
+}
+
+#[test]
+fn mysql_coalesces_renamed_auto_increment_primary_key_column() {
+    let mut old_pk = existing_pk_column("campaign_rel_id", "bigint", true, false);
+    old_pk.id = "old_campaign_rel_id".to_string();
+
+    let mut id = existing_pk_column("id", "bigint", false, true);
+    id.id = "new_id".to_string();
+    id.original.as_mut().unwrap().name = "legacy_id".to_string();
+    id.extra = Some(ColumnExtra { auto_increment: Some(true), ..Default::default() });
+
+    let result = build_table_structure_change_sql(structure_change_options(
+        DatabaseType::Mysql,
+        None,
+        "users",
+        vec![old_pk, id],
+    ));
+
+    assert_eq!(result.warnings, Vec::<String>::new());
+    assert_eq!(
+        result.statements,
+        vec![
+            "ALTER TABLE `users` DROP PRIMARY KEY, CHANGE COLUMN `legacy_id` `id` bigint NOT NULL AUTO_INCREMENT, ADD PRIMARY KEY (`id`);"
+        ]
+    );
+}
+
+#[test]
+fn mysql_coalesces_composite_primary_key_around_auto_increment_column() {
+    for (auto_first, has_supporting_index, expected_primary_key) in
+        [(true, false, "`id`, `tenant_id`"), (false, false, "`tenant_id`, `id`"), (false, true, "`tenant_id`, `id`")]
+    {
+        let mut old_pk = existing_pk_column("legacy_id", "bigint", true, false);
+        old_pk.id = "old_legacy_id".to_string();
+
+        let mut id = existing_pk_column("id", "bigint", false, true);
+        id.id = "new_id".to_string();
+        id.extra = Some(ColumnExtra { auto_increment: Some(true), ..Default::default() });
+
+        let mut tenant_id = existing_pk_column("tenant_id", "bigint", false, true);
+        tenant_id.id = "new_tenant_id".to_string();
+
+        let columns = if auto_first { vec![old_pk, id, tenant_id] } else { vec![old_pk, tenant_id, id] };
+        let mut options = structure_change_options(DatabaseType::Mysql, None, "users", columns);
+        if has_supporting_index {
+            options.indexes = vec![existing_index("uniq_users_id", &["id"], true)];
+        }
+
+        let result = build_table_structure_change_sql(options);
+
+        assert_eq!(result.warnings, Vec::<String>::new());
+        assert_eq!(result.statements.len(), 1);
+        assert_eq!(
+            result.statements[0],
+            format!(
+                "ALTER TABLE `users` DROP PRIMARY KEY, MODIFY COLUMN `id` bigint NOT NULL AUTO_INCREMENT, ADD PRIMARY KEY ({expected_primary_key});"
+            )
+        );
+    }
+}
+
+#[test]
+fn mysql_compatible_database_keeps_existing_primary_key_statement_sequence() {
+    let mut old_pk = existing_pk_column("legacy_id", "bigint", true, false);
+    old_pk.id = "old_legacy_id".to_string();
+
+    let mut id = existing_pk_column("id", "bigint", false, true);
+    id.id = "new_id".to_string();
+    id.extra = Some(ColumnExtra { auto_increment: Some(true), ..Default::default() });
+
+    let result = build_table_structure_change_sql(structure_change_options(
+        DatabaseType::StarRocks,
+        None,
+        "users",
+        vec![old_pk, id],
+    ));
+
+    assert_eq!(result.warnings, Vec::<String>::new());
+    assert_eq!(
+        result.statements,
+        vec![
+            "ALTER TABLE `users` MODIFY COLUMN `id` bigint NOT NULL AUTO_INCREMENT;",
+            "ALTER TABLE `users` DROP PRIMARY KEY;",
+            "ALTER TABLE `users` ADD PRIMARY KEY (`id`);",
+        ]
+    );
+}
+
+#[test]
+fn gaussdb_m_mode_keeps_existing_primary_key_statement_sequence() {
+    let mut old_pk = existing_pk_column("legacy_id", "bigint", true, false);
+    old_pk.id = "old_legacy_id".to_string();
+
+    let mut id = existing_pk_column("id", "bigint", false, true);
+    id.id = "new_id".to_string();
+    id.extra = Some(ColumnExtra { auto_increment: Some(true), ..Default::default() });
+
+    let mut options = structure_change_options(DatabaseType::Gaussdb, None, "users", vec![old_pk, id]);
+    options.is_gaussdb_m_mode = true;
+    let result = build_table_structure_change_sql(options);
+
+    assert_eq!(result.warnings, Vec::<String>::new());
+    assert_eq!(
+        result.statements,
+        vec![
+            "ALTER TABLE `users` MODIFY COLUMN `id` bigint NOT NULL AUTO_INCREMENT;",
+            "ALTER TABLE `users` DROP PRIMARY KEY;",
+            "ALTER TABLE `users` ADD PRIMARY KEY (`id`);",
+        ]
+    );
+}
+
 #[test]
 fn builds_no_statements_when_primary_key_unchanged() {
     let result = build_table_structure_change_sql(structure_change_options(