chore: modify the column allocation method of auto increment expr and add auto increment check of add column

Author: KKould

src/query/service/src/interpreters/interpreter_table_add_column.rs

@@ -111,6 +111,12 @@ impl Interpreter for AddTableColumnInterpreter {
                 &self.plan.field.name, &self.plan.table
             )));
         }
+        if self.plan.is_autoincrement && num_rows > 0 {
+            return Err(ErrorCode::AlterTableError(format!(
+                "Cannot add column '{}' with `AUTOINCREMENT` to non-empty table '{}'",
+                &self.plan.field.name, &self.plan.table
+            )));
+        }
         if field.default_expr().is_some() {
             let _ = DefaultExprBinder::try_new(self.ctx.clone())?.get_scalar(&field)?;
         }

src/query/sql/src/planner/binder/ddl/table.rs

@@ -74,7 +74,6 @@ use databend_common_expression::infer_schema_type;
 use databend_common_expression::infer_table_schema;
 use databend_common_expression::types::DataType;
 use databend_common_expression::AutoIncrementExpr;
-use databend_common_expression::ColumnId;
 use databend_common_expression::ComputedExpr;
 use databend_common_expression::DataField;
 use databend_common_expression::DataSchemaRefExt;
@@ -1093,7 +1092,7 @@ impl Binder {
                     .get_table(&catalog, &database, &table)
                     .await?
                     .schema();
-                let (field, comment, is_deterministic, is_nextval) =
+                let (field, comment, is_deterministic, is_nextval, is_autoincrement) =
                     self.analyze_add_column(column, schema).await?;
                 let option = match ast_option {
                     AstAddColumnOption::First => AddColumnOption::First,
@@ -1112,6 +1111,7 @@ impl Binder {
                     option,
                     is_deterministic,
                     is_nextval,
+                    is_autoincrement,
                 })))
             }
             AlterTableAction::AddConstraint { constraint } => {
@@ -1182,7 +1182,7 @@ impl Binder {
                             .await?
                             .schema();
                         for column in column_def_vec {
-                            let (field, comment, _, _) =
+                            let (field, comment, _, _, _) =
                                 self.analyze_add_column(column, schema.clone()).await?;
                             field_and_comment.push((field, comment));
                         }
@@ -1659,12 +1659,13 @@ impl Binder {
         &self,
         column: &ColumnDefinition,
         table_schema: TableSchemaRef,
-    ) -> Result<(TableField, String, bool, bool)> {
+    ) -> Result<(TableField, String, bool, bool, bool)> {
         let name = normalize_identifier(&column.name, &self.name_resolution_ctx).name;
         let not_null = self.is_column_not_null();
         let data_type = resolve_type_name(&column.data_type, not_null)?;
         let mut is_deterministic = true;
         let mut is_nextval = false;
+        let mut is_autoincrement = false;
         let mut field = TableField::new(&name, data_type);
 
         if let Some(expr) = &column.expr {
@@ -1710,16 +1711,23 @@ impl Binder {
                         ));
                     }
                     field.auto_increment_expr = Some(AutoIncrementExpr {
-                        column_id: table_schema.fields().len() as ColumnId,
+                        column_id: table_schema.next_column_id(),
                         start: *start,
                         step: *step,
                         is_ordered: *is_ordered,
                     });
+                    is_autoincrement = true;
                 }
             }
         }
         let comment = column.comment.clone().unwrap_or_default();
-        Ok((field, comment, is_deterministic, is_nextval))
+        Ok((
+            field,
+            comment,
+            is_deterministic,
+            is_nextval,
+            is_autoincrement,
+        ))
     }
 
     #[async_backtrace::framed]
@@ -1728,11 +1736,12 @@ impl Binder {
         columns: &[ColumnDefinition],
     ) -> Result<(TableSchemaRef, Vec<String>)> {
         let mut has_computed = false;
+        let mut has_autoincrement = false;
         let mut fields = Vec::with_capacity(columns.len());
         let mut fields_comments = Vec::with_capacity(columns.len());
         let not_null = self.is_column_not_null();
         let mut default_expr_binder = DefaultExprBinder::try_new(self.ctx.clone())?;
-        for (i, column) in columns.iter().enumerate() {
+        for column in columns.iter() {
             let name = normalize_identifier(&column.name, &self.name_resolution_ctx).name;
             let schema_data_type = resolve_type_name(&column.data_type, not_null)?;
             fields_comments.push(column.comment.clone().unwrap_or_default());
@@ -1757,8 +1766,9 @@ impl Binder {
                                 "AUTO INCREMENT only supports Decimal or Numeric (e.g. INT32) types",
                             ));
                         }
+                        has_autoincrement = true;
                         field.auto_increment_expr = Some(AutoIncrementExpr {
-                            column_id: i as ColumnId,
+                            column_id: 0,
                             start: *start,
                             step: *step,
                             is_ordered: *is_ordered,
@@ -1769,6 +1779,18 @@ impl Binder {
             }
             fields.push(field);
         }
+        // update auto increment expr column id
+        if has_autoincrement {
+            let table_schema = TableSchema::new(fields.clone());
+
+            for (i, table_field) in table_schema.fields().iter().enumerate() {
+                let Some(auto_increment_expr) = fields[i].auto_increment_expr.as_mut() else {
+                    continue;
+                };
+
+                auto_increment_expr.column_id = table_field.column_id;
+            }
+        }
 
         let fields = if has_computed {
             let mut source_fields = Vec::with_capacity(fields.len());

src/query/sql/src/planner/plans/ddl/table.rs

@@ -315,6 +315,7 @@ pub struct AddTableColumnPlan {
     pub option: AddColumnOption,
     pub is_deterministic: bool,
     pub is_nextval: bool,
+    pub is_autoincrement: bool,
 }
 
 impl AddTableColumnPlan {

tests/suites/0_stateless/02_ddl/02_0001_autoincrement.result

@@ -55,29 +55,35 @@ Error: APIError: QueryFailed: [1065]error:
 >>>> 
 alter table test_autoincrement add column c7 int autoincrement;
 
+Error: APIError: QueryFailed: [1132]Cannot add column 'c7' with `AUTOINCREMENT` to non-empty table 'test_autoincrement'
+<<<<
 >>>> 
-insert into test_autoincrement (c1) values(0);
+alter table test_autoincrement drop column c3;
 
-1
 >>>> select * from test_autoincrement order by c2;
-0	0	0	1	1	0.00	NULL
-0	1	1	3	3	1.00	NULL
-0	2	2	5	5	2.00	NULL
-0	3	3	7	7	3.00	0
+0	0	1	1	0.00
+0	1	3	3	1.00
+0	2	5	5	2.00
 <<<<
 >>>> 
-alter table test_autoincrement drop column c2;
+create or replace table test_empty_autoincrement (c1 int);
 
->>>> select * from test_autoincrement order by c2;
-Error: APIError: QueryFailed: [1065]error: 
-  --> SQL:1:43
-  |
-1 | select * from test_autoincrement order by c2
-  |                                           ^^ column c2 doesn't exist
+>>>> 
+alter table test_empty_autoincrement add column c2 int autoincrement;
+
+>>>> 
+insert into test_empty_autoincrement (c1) values(0);
 
+1
+>>>> 
+insert into test_empty_autoincrement (c1) values(0);
 
+1
+>>>> select * from test_empty_autoincrement order by c2;
+0	0
+0	1
 <<<<
 >>>> 
-drop table test_autoincrement;
+drop table test_empty_autoincrement;
 
 >>>> drop database test

tests/suites/0_stateless/02_ddl/02_0001_autoincrement.sh

@@ -41,19 +41,31 @@ alter table test_autoincrement add column c7 int autoincrement;
 """
 
 stmt """
-insert into test_autoincrement (c1) values(0);
+alter table test_autoincrement drop column c3;
 """
 
 query "select * from test_autoincrement order by c2;"
 
+# test add column success on empty table
 stmt """
-alter table test_autoincrement drop column c2;
+create or replace table test_empty_autoincrement (c1 int);
 """
 
-query "select * from test_autoincrement order by c2;"
+stmt """
+alter table test_empty_autoincrement add column c2 int autoincrement;
+"""
+
+stmt """
+insert into test_empty_autoincrement (c1) values(0);
+"""
+stmt """
+insert into test_empty_autoincrement (c1) values(0);
+"""
+
+query "select * from test_empty_autoincrement order by c2;"
 
 stmt """
-drop table test_autoincrement;
+drop table test_empty_autoincrement;
 """
 
 stmt "drop database test"