feat(dameng): support toggling identity on existing columns

Closes #6906

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

@@ -1,6 +1,7 @@
 use super::column_format::{
-    clickhouse_column_type, column_data_type, column_definition, is_mysql_character_data_type,
-    original_is_mysql_generated_column, original_mysql_generated_clause,
+    clickhouse_column_type, column_data_type, column_definition, has_dameng_identity,
+    is_dameng_identity_compatible_type, is_mysql_character_data_type, original_is_mysql_generated_column,
+    original_mysql_generated_clause,
 };
 use super::columns::build_drop_column_sql;
 use super::comments::build_sqlserver_column_comment_sql;
@@ -100,7 +101,7 @@ pub fn build_single_column_alter_sql(options: SingleColumnAlterSqlOptions) -> Ta
         StructureDialect::Mysql => statements.extend(build_mysql_existing_column_sql(&table, &options.column, "")),
         StructureDialect::Doris => statements.extend(build_doris_existing_column_sql(&table, &options.column, "")),
         StructureDialect::Postgres => statements.extend(build_postgres_existing_column_sql(&table, &options.column)),
-        StructureDialect::Oracle | StructureDialect::Dameng => {
+        StructureDialect::Oracle => {
             if options.database_type == Some(crate::models::connection::DatabaseType::Iris) {
                 statements.extend(build_iris_existing_column_sql(&table, &options.column));
             } else if options.database_type == Some(crate::models::connection::DatabaseType::Xugu) {
@@ -109,6 +110,9 @@ pub fn build_single_column_alter_sql(options: SingleColumnAlterSqlOptions) -> Ta
                 statements.extend(build_oracle_like_existing_column_sql(dialect, &table, &options.column))
             }
         }
+        StructureDialect::Dameng => {
+            statements.extend(build_dameng_existing_column_sql(&table, &options.column, true, &mut warnings))
+        }
         StructureDialect::Oscar => statements.extend(build_oscar_existing_column_sql(dialect, &table, &options.column)),
         StructureDialect::H2 => statements.extend(build_h2_existing_column_sql(&table, &options.column)),
         StructureDialect::ClickHouse => {
@@ -221,6 +225,78 @@ fn identity_matches_original(identity: &super::types::ColumnIdentity, original_e
     sqlserver_identity_matches_original(identity, original_extra)
 }
 
+#[derive(Clone, Copy)]
+enum DamengIdentityTransition {
+    None,
+    Add { seed: i64, increment: i64 },
+    Drop,
+    ParametersChanged,
+}
+
+fn dameng_identity_transition(column: &EditableStructureColumn) -> DamengIdentityTransition {
+    let Some(original) = &column.original else {
+        return DamengIdentityTransition::None;
+    };
+    let original_extra = original.extra.as_deref().unwrap_or("");
+    let original_identity = original_has_identity(original_extra);
+    let current_identity = has_dameng_identity(column);
+    match (original_identity, current_identity) {
+        (false, false) => DamengIdentityTransition::None,
+        (false, true) => {
+            let identity = column.extra.as_ref().and_then(|extra| extra.identity.as_ref());
+            DamengIdentityTransition::Add {
+                seed: identity.and_then(|identity| identity.seed).unwrap_or(1),
+                increment: identity.and_then(|identity| identity.increment).unwrap_or(1),
+            }
+        }
+        (true, false) => DamengIdentityTransition::Drop,
+        (true, true) => match column.extra.as_ref().and_then(|extra| extra.identity.as_ref()) {
+            Some(identity) if !identity_matches_original(identity, original_extra) => {
+                DamengIdentityTransition::ParametersChanged
+            }
+            _ => DamengIdentityTransition::None,
+        },
+    }
+}
+
+pub(super) fn dameng_drops_identity(column: &EditableStructureColumn) -> bool {
+    matches!(dameng_identity_transition(column), DamengIdentityTransition::Drop)
+}
+
+pub(super) fn validate_dameng_existing_identity_change(
+    column: &EditableStructureColumn,
+    warnings: &mut Vec<String>,
+) -> bool {
+    match dameng_identity_transition(column) {
+        DamengIdentityTransition::Add { increment, .. } => {
+            if !is_dameng_identity_compatible_type(&column.data_type) {
+                warnings.push(format!(
+                    "Dameng identity column \"{}\" must use tinyint, smallint, int, integer, bigint, number, numeric, or decimal/dec with scale 0.",
+                    column.name
+                ));
+            } else if column.is_nullable {
+                warnings.push(format!(
+                    "Dameng identity column \"{}\" must be NOT NULL before identity can be enabled.",
+                    column.name
+                ));
+            } else if increment == 0 {
+                warnings.push(format!("Dameng identity column \"{}\" increment cannot be 0.", column.name));
+            } else {
+                return true;
+            }
+            false
+        }
+        DamengIdentityTransition::ParametersChanged => {
+            warnings.push(format!(
+                "Changing Dameng IDENTITY seed or increment for existing column \"{}\" is not supported from this editor.",
+                column.name
+            ));
+            false
+        }
+        DamengIdentityTransition::None | DamengIdentityTransition::Drop => true,
+    }
+}
+
 pub(super) fn has_column_extra_change(column: &EditableStructureColumn) -> bool {
     let Some(original) = &column.original else { return false };
     let current_extra = column.extra.as_ref();
@@ -502,6 +578,32 @@ pub(super) fn build_oracle_like_existing_column_sql(
     statements
 }
 
+pub(super) fn build_dameng_existing_column_sql(
+    table: &str,
+    column: &EditableStructureColumn,
+    emit_identity_drop: bool,
+    warnings: &mut Vec<String>,
+) -> Vec<String> {
+    let transition = dameng_identity_transition(column);
+    if !validate_dameng_existing_identity_change(column, warnings) {
+        return Vec::new();
+    }
+    let mut statements = Vec::new();
+    if emit_identity_drop && matches!(transition, DamengIdentityTransition::Drop) {
+        statements.push(format!("ALTER TABLE {table} DROP IDENTITY;"));
+    }
+
+    statements.extend(build_oracle_like_existing_column_sql(StructureDialect::Dameng, table, column));
+
+    if let DamengIdentityTransition::Add { seed, increment } = transition {
+        statements.push(format!(
+            "ALTER TABLE {table} ADD COLUMN {} IDENTITY({seed}, {increment});",
+            quote_ident(StructureDialect::Dameng, &column.name)
+        ));
+    }
+    statements
+}
+
 pub(super) fn build_iris_existing_column_sql(table: &str, column: &EditableStructureColumn) -> Vec<String> {
     let Some(original) = &column.original else {
         return Vec::new();

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

@@ -1,9 +1,10 @@
 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_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,
+    build_clickhouse_existing_column_sql, build_dameng_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_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, dameng_drops_identity,
+    has_column_extra_change, has_existing_column_attribute_change, validate_dameng_existing_identity_change,
 };
 use super::column_format::{
     column_definition, has_dameng_identity, is_dameng_identity_compatible_type, is_mysql_character_data_type,
@@ -25,6 +26,32 @@ pub(super) fn build_column_sql(options: &TableStructureSqlOptions, warnings: &mu
     let table = qualified_table(dialect, options.schema.as_deref(), &options.table_name);
     let database_label = database_label(options.database_type);
     let active_columns: Vec<_> = options.columns.iter().filter(|column| !column.marked_for_drop).collect();
+    if dialect == StructureDialect::Dameng {
+        let identity_columns: Vec<_> = active_columns.iter().filter(|column| has_dameng_identity(column)).collect();
+        if identity_columns.len() > 1
+            || identity_columns.iter().any(|column| {
+                column.extra.as_ref().and_then(|extra| extra.identity.as_ref()).and_then(|identity| identity.increment)
+                    == Some(0)
+            })
+        {
+            return Vec::new();
+        }
+        let mut valid_identity_changes = true;
+        for column in &active_columns {
+            if has_dameng_identity(column) && !is_dameng_identity_compatible_type(&column.data_type) {
+                warnings.push(format!(
+                    "Dameng identity column \"{}\" must use tinyint, smallint, int, integer, bigint, number, numeric, or decimal/dec with scale 0.",
+                    column.name
+                ));
+                valid_identity_changes = false;
+            } else if column.original.is_some() && !validate_dameng_existing_identity_change(column, warnings) {
+                valid_identity_changes = false;
+            }
+        }
+        if !valid_identity_changes {
+            return Vec::new();
+        }
+    }
     if is_oracle_like(dialect)
         && active_columns.is_empty()
         && options.columns.iter().any(|column| column.marked_for_drop)
@@ -56,6 +83,13 @@ pub(super) fn build_column_sql(options: &TableStructureSqlOptions, warnings: &mu
         .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();
+    // DM8 owns identity at table level, so remove the old identity before any per-column ADD,
+    // even when the target column appears earlier in the submitted draft.
+    if dialect == StructureDialect::Dameng
+        && options.columns.iter().any(|column| !column.marked_for_drop && dameng_drops_identity(column))
+    {
+        statements.push(format!("ALTER TABLE {table} DROP IDENTITY;"));
+    }
 
     for column in &options.columns {
         if column.marked_for_drop {
@@ -109,16 +143,6 @@ pub(super) fn build_column_sql(options: &TableStructureSqlOptions, warnings: &mu
                 ));
                 continue;
             }
-            if dialect == StructureDialect::Dameng
-                && has_dameng_identity(column)
-                && !is_dameng_identity_compatible_type(&column.data_type)
-            {
-                warnings.push(format!(
-                    "Dameng identity column \"{}\" must use tinyint, smallint, int, integer, bigint, number, numeric, or decimal/dec with scale 0.",
-                    column.name
-                ));
-                continue;
-            }
             if !capabilities.comment && !clean(&column.comment).is_empty() {
                 warnings.push(format!(
                     "Column comments are not supported for {database_label} from this editor; the comment for \"{}\" was ignored.",
@@ -215,7 +239,7 @@ pub(super) fn build_column_sql(options: &TableStructureSqlOptions, warnings: &mu
             }
             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 => {
+            StructureDialect::Oracle => {
                 if options.database_type == Some(crate::models::connection::DatabaseType::Iris) {
                     statements.extend(build_iris_existing_column_sql(&table, column));
                 } else if options.database_type == Some(crate::models::connection::DatabaseType::Xugu) {
@@ -224,6 +248,9 @@ pub(super) fn build_column_sql(options: &TableStructureSqlOptions, warnings: &mu
                     statements.extend(build_oracle_like_existing_column_sql(dialect, &table, column))
                 }
             }
+            StructureDialect::Dameng => {
+                statements.extend(build_dameng_existing_column_sql(&table, column, false, warnings))
+            }
             // 神通 MODIFY 语法与 Oracle 有差异（NULL/NOT NULL 须单独一条），用专属实现。
             StructureDialect::Oscar => statements.extend(build_oscar_existing_column_sql(dialect, &table, column)),
             StructureDialect::H2 => statements.extend(build_h2_existing_column_sql(&table, column)),

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

@@ -2833,6 +2833,212 @@ fn dameng_unchanged_identity_extra_does_not_mark_existing_column_changed() {
     assert_eq!(result.statements, Vec::<String>::new());
 }
 
+#[test]
+fn dameng_enables_identity_on_existing_not_null_column() {
+    let mut id = existing_pk_column("ID", "INT", false, false);
+    id.extra = Some(ColumnExtra {
+        auto_increment: Some(true),
+        identity: Some(ColumnIdentity { generation: None, seed: Some(10), increment: Some(2) }),
+        ..Default::default()
+    });
+
+    let result = build_table_structure_change_sql(structure_change_options(
+        DatabaseType::Dameng,
+        Some("SYSDBA"),
+        "TEST",
+        vec![id],
+    ));
+
+    assert_eq!(result.warnings, Vec::<String>::new());
+    assert_eq!(result.statements, vec!["ALTER TABLE \"SYSDBA\".\"TEST\" ADD COLUMN \"ID\" IDENTITY(10, 2);"]);
+}
+
+#[test]
+fn dameng_makes_existing_column_not_null_before_enabling_identity() {
+    let mut id = column("ID");
+    id.data_type = "INT".to_string();
+    id.is_nullable = false;
+    id.extra = Some(ColumnExtra {
+        auto_increment: Some(true),
+        identity: Some(ColumnIdentity { generation: None, seed: Some(1), increment: Some(1) }),
+        ..Default::default()
+    });
+    id.original = Some(ColumnInfo {
+        name: "ID".to_string(),
+        data_type: "INT".to_string(),
+        is_nullable: true,
+        column_default: None,
+        is_primary_key: false,
+        extra: None,
+        comment: None,
+        ..Default::default()
+    });
+
+    let result = build_table_structure_change_sql(structure_change_options(
+        DatabaseType::Dameng,
+        Some("SYSDBA"),
+        "TEST",
+        vec![id],
+    ));
+
+    assert_eq!(result.warnings, Vec::<String>::new());
+    assert_eq!(
+        result.statements,
+        vec![
+            "ALTER TABLE \"SYSDBA\".\"TEST\" MODIFY (\"ID\" INT NOT NULL);",
+            "ALTER TABLE \"SYSDBA\".\"TEST\" ADD COLUMN \"ID\" IDENTITY(1, 1);",
+        ]
+    );
+}
+
+#[test]
+fn dameng_disables_identity_on_existing_column() {
+    let mut id = existing_pk_column("ID", "INT", false, false);
+    id.extra = Some(ColumnExtra::default());
+    id.original.as_mut().unwrap().extra = Some("IDENTITY(10, 2)".to_string());
+
+    let result = build_table_structure_change_sql(structure_change_options(
+        DatabaseType::Dameng,
+        Some("SYSDBA"),
+        "TEST",
+        vec![id],
+    ));
+
+    assert_eq!(result.warnings, Vec::<String>::new());
+    assert_eq!(result.statements, vec!["ALTER TABLE \"SYSDBA\".\"TEST\" DROP IDENTITY;"]);
+}
+
+#[test]
+fn dameng_moves_identity_with_drop_before_add_regardless_of_column_order() {
+    let mut target = existing_pk_column("TARGET_ID", "BIGINT", false, false);
+    target.extra = Some(ColumnExtra {
+        auto_increment: Some(true),
+        identity: Some(ColumnIdentity { generation: None, seed: Some(100), increment: Some(5) }),
+        ..Default::default()
+    });
+
+    let mut source = existing_pk_column("SOURCE_ID", "INT", false, false);
+    source.extra = Some(ColumnExtra::default());
+    source.original.as_mut().unwrap().extra = Some("IDENTITY(1, 1)".to_string());
+
+    let result = build_table_structure_change_sql(structure_change_options(
+        DatabaseType::Dameng,
+        Some("SYSDBA"),
+        "TEST",
+        vec![target, source],
+    ));
+
+    assert_eq!(result.warnings, Vec::<String>::new());
+    assert_eq!(
+        result.statements,
+        vec![
+            "ALTER TABLE \"SYSDBA\".\"TEST\" DROP IDENTITY;",
+            "ALTER TABLE \"SYSDBA\".\"TEST\" ADD COLUMN \"TARGET_ID\" IDENTITY(100, 5);",
+        ]
+    );
+}
+
+#[test]
+fn dameng_rejects_identity_on_incompatible_existing_column() {
+    let mut code = column("CODE");
+    code.data_type = "VARCHAR(255)".to_string();
+    code.is_nullable = false;
+    code.extra = Some(ColumnExtra {
+        auto_increment: Some(true),
+        identity: Some(ColumnIdentity { generation: None, seed: Some(1), increment: Some(1) }),
+        ..Default::default()
+    });
+    code.original = Some(ColumnInfo {
+        name: "CODE".to_string(),
+        data_type: "VARCHAR(255)".to_string(),
+        is_nullable: false,
+        column_default: None,
+        is_primary_key: false,
+        extra: None,
+        comment: None,
+        ..Default::default()
+    });
+
+    let result = build_table_structure_change_sql(structure_change_options(
+        DatabaseType::Dameng,
+        Some("SYSDBA"),
+        "TEST",
+        vec![code],
+    ));
+
+    assert!(result.statements.is_empty());
+    assert_eq!(
+        result.warnings,
+        vec!["Dameng identity column \"CODE\" must use tinyint, smallint, int, integer, bigint, number, numeric, or decimal/dec with scale 0."]
+    );
+}
+
+#[test]
+fn dameng_rejects_zero_increment_when_enabling_existing_identity() {
+    let mut id = existing_pk_column("ID", "INT", false, false);
+    id.extra = Some(ColumnExtra {
+        auto_increment: Some(true),
+        identity: Some(ColumnIdentity { generation: None, seed: Some(1), increment: Some(0) }),
+        ..Default::default()
+    });
+
+    let result = build_table_structure_change_sql(structure_change_options(
+        DatabaseType::Dameng,
+        Some("SYSDBA"),
+        "TEST",
+        vec![id],
+    ));
+
+    assert!(result.statements.is_empty());
+    assert_eq!(result.warnings, vec!["Dameng identity column \"ID\" increment cannot be 0."]);
+}
+
+#[test]
+fn dameng_rejects_changing_existing_identity_parameters() {
+    let mut id = existing_pk_column("ID", "INT", false, false);
+    id.extra = Some(ColumnExtra {
+        auto_increment: Some(true),
+        identity: Some(ColumnIdentity { generation: None, seed: Some(10), increment: Some(3) }),
+        ..Default::default()
+    });
+    id.original.as_mut().unwrap().extra = Some("IDENTITY(10, 2)".to_string());
+
+    let result = build_table_structure_change_sql(structure_change_options(
+        DatabaseType::Dameng,
+        Some("SYSDBA"),
+        "TEST",
+        vec![id],
+    ));
+
+    assert!(result.statements.is_empty());
+    assert_eq!(
+        result.warnings,
+        vec![
+            "Changing Dameng IDENTITY seed or increment for existing column \"ID\" is not supported from this editor."
+        ]
+    );
+}
+
+#[test]
+fn oracle_does_not_adopt_dameng_existing_identity_ddl() {
+    let mut id = existing_pk_column("ID", "NUMBER(10)", false, false);
+    id.extra = Some(ColumnExtra {
+        auto_increment: Some(true),
+        identity: Some(ColumnIdentity { generation: None, seed: Some(1), increment: Some(1) }),
+        ..Default::default()
+    });
+
+    let result = build_table_structure_change_sql(structure_change_options(
+        DatabaseType::Oracle,
+        Some("APP"),
+        "USERS",
+        vec![id],
+    ));
+
+    assert_eq!(result.warnings, Vec::<String>::new());
+    assert_eq!(result.statements, Vec::<String>::new());
+}
+
 #[test]
 fn dameng_rejects_adding_second_identity_column() {
     let mut existing = column("ID");
@@ -2876,6 +3082,7 @@ fn dameng_rejects_adding_second_identity_column() {
         is_gaussdb_m_mode: false,
     });
 
+    assert_eq!(result.statements, Vec::<String>::new());
     assert_eq!(result.warnings, vec!["Dameng tables can have only one identity column."]);
 }