fix: invalid sequence step (#18800)

* fix: transform_async_function inserting multiple data in one sql causes step to be ignored

* fix: Sequence inserting multiple data in one sql causes step to be ignored

* fix: codefmt

* chore: move step check on `bind_table`

* chore: codefmt

* chore: fix stmt-error.txt

Author: KKould

src/meta/api/src/sequence_nextval_impl.rs

@@ -73,7 +73,7 @@ where KV: kvapi::KVApi<Error = MetaError> + ?Sized
         }
 
         // update meta
-        self.sequence_meta.current += count;
+        self.sequence_meta.current += count * self.sequence_meta.step as u64;
 
         let condition = vec![txn_cond_eq_seq(&self.ident, self.sequence_meta.seq)];
         let if_then = vec![

src/query/ast/src/parser/statement.rs

@@ -3432,12 +3432,6 @@ pub fn column_def(i: Input) -> IResult<ColumnDefinition> {
                         ErrorKind::Other("DEFAULT and AUTOINCREMENT cannot exist at the same time"),
                     )));
                 }
-                if !is_ordered {
-                    return Err(nom::Err::Error(Error::from_error_kind(
-                        i,
-                        ErrorKind::Other("AUTOINCREMENT only support ORDER now"),
-                    )));
-                }
                 def.expr = Some(ColumnExpr::AutoIncrement {
                     start,
                     step,

src/query/ast/tests/it/parser.rs

@@ -1018,7 +1018,6 @@ fn test_statement_error() {
         r#"CREATE TABLE t(c1 NULLABLE(int) NOT NULL);"#,
         r#"create table a (c1 decimal(38), c2 int) partition by ();"#,
         r#"CREATE TABLE t(c1 int, c2 int) partition by (c1, c2) PROPERTIES ("read.split.target-size"='134217728', "read.split.metadata-target-size"=33554432);"#,
-        r#"create table a.b (c integer autoincrement (10, 20) NOORDER)"#,
         r#"drop table if a.b"#,
         r#"show table"#,
         r#"show column"#,

src/query/ast/tests/it/testdata/stmt-error.txt

@@ -111,18 +111,6 @@ error:
   | while parsing `CREATE [OR REPLACE] TABLE [IF NOT EXISTS] [<database>.]<table> [<source>] [<table_options>]`
 
 
----------- Input ----------
-create table a.b (c integer autoincrement (10, 20) NOORDER)
----------- Output ---------
-error: 
-  --> SQL:1:59
-  |
-1 | create table a.b (c integer autoincrement (10, 20) NOORDER)
-  | ------                                                    ^ AUTOINCREMENT only support ORDER now
-  | |                                                          
-  | while parsing `CREATE [OR REPLACE] TABLE [IF NOT EXISTS] [<database>.]<table> [<source>] [<table_options>]`
-
-
 ---------- Input ----------
 drop table if a.b
 ---------- Output ---------

src/query/service/src/pipelines/processors/transforms/transform_async_function.rs

@@ -27,11 +27,13 @@ use databend_common_expression::FromData;
 use databend_common_meta_app::principal::AutoIncrementKey;
 use databend_common_meta_app::schema::GetAutoIncrementNextValueReq;
 use databend_common_meta_app::schema::GetSequenceNextValueReq;
+use databend_common_meta_app::schema::GetSequenceReply;
 use databend_common_meta_app::schema::GetSequenceReq;
 use databend_common_meta_app::schema::SequenceIdent;
 use databend_common_pipeline_transforms::processors::AsyncTransform;
 use databend_common_sql::binder::AsyncFunctionDesc;
 use databend_common_storages_fuse::TableContext;
+use databend_common_users::GrantObjectVisibilityChecker;
 use databend_common_users::Object;
 
 use crate::pipelines::processors::transforms::transform_dictionary::DictionaryOperator;
@@ -135,26 +137,30 @@ impl TransformAsyncFunction {
         } else {
             // Get or create the sequence counter
             let counter = counter_lock.read().await;
+            let fn_range_collect = |start: u64, end: u64, step: i64| {
+                (0..end - start)
+                    .map(|num| start + num * step as u64)
+                    .collect::<Vec<_>>()
+            };
+            // We need to fetch more sequence numbers
+            let catalog = ctx.get_default_catalog()?;
 
             // Try to reserve sequence numbers from the counter
             if let Some((start, _end)) = counter.try_reserve(count) {
+                let step = fetcher.step(&ctx, &catalog).await?;
                 // We have enough sequence numbers in the current batch
-                let range = start..start + count;
-                UInt64Type::from_data(range.collect::<Vec<u64>>())
+                UInt64Type::from_data(fn_range_collect(start, start + count, step))
             } else {
                 // drop the read lock and get the write lock
                 drop(counter);
                 let counter = counter_lock.write().await;
                 {
                     // try reserve again
                     if let Some((start, _end)) = counter.try_reserve(count) {
+                        let step = fetcher.step(&ctx, &catalog).await?;
                         // We have enough sequence numbers in the current batch
-                        let range = start..start + count;
-                        UInt64Type::from_data(range.collect::<Vec<u64>>())
+                        UInt64Type::from_data(fn_range_collect(start, count, step))
                     } else {
-                        // We need to fetch more sequence numbers
-                        let catalog = ctx.get_default_catalog()?;
-
                         // Get current state of the counter
                         let current = counter.current.load(Ordering::Relaxed);
                         let max = counter.max.load(Ordering::Relaxed);
@@ -163,7 +169,11 @@ impl TransformAsyncFunction {
                         let remaining = max.saturating_sub(current);
                         let to_fetch = count.saturating_sub(remaining);
 
-                        let (start, batch_size) = fetcher.fetch(&ctx, &catalog, to_fetch).await?;
+                        let NextValFetchResult {
+                            start,
+                            batch_size,
+                            step,
+                        } = fetcher.fetch(&ctx, &catalog, to_fetch).await?;
 
                         // If we have remaining numbers, use them first
                         if remaining > 0 {
@@ -175,14 +185,16 @@ impl TransformAsyncFunction {
 
                             // Add the remaining numbers
                             let remaining_to_use = remaining.min(count);
-                            numbers.extend(
-                                (current..current + remaining_to_use).collect::<Vec<u64>>(),
-                            );
+                            numbers.extend(fn_range_collect(
+                                current,
+                                current + remaining_to_use,
+                                step,
+                            ));
 
                             // Add numbers from the new batch if needed
                             if remaining_to_use < count {
                                 let new_needed = count - remaining_to_use;
-                                numbers.extend((start..start + new_needed).collect::<Vec<u64>>());
+                                numbers.extend(fn_range_collect(start, start + new_needed, step));
                                 // Update the counter to reflect that we've used some of the new batch
                                 counter.current.store(start + new_needed, Ordering::SeqCst);
                             }
@@ -192,8 +204,7 @@ impl TransformAsyncFunction {
                             // No remaining numbers, just use the new batch
                             counter.update_batch(start + count, batch_size - count);
                             // Return the sequence numbers needed for this request
-                            let range = start..start + count;
-                            UInt64Type::from_data(range.collect::<Vec<u64>>())
+                            UInt64Type::from_data(fn_range_collect(start, start + count, step))
                         }
                     }
                 }
@@ -211,7 +222,15 @@ pub trait NextValFetcher {
         ctx: &QueryContext,
         catalog: &Arc<dyn Catalog>,
         to_fetch: u64,
-    ) -> Result<(u64 /* start */, u64 /* batch */)>;
+    ) -> Result<NextValFetchResult>;
+
+    async fn step(&self, ctx: &QueryContext, catalog: &Arc<dyn Catalog>) -> Result<i64>;
+}
+
+pub struct NextValFetchResult {
+    start: u64,
+    batch_size: u64,
+    step: i64,
 }
 
 pub struct SequenceNextValFetcher {
@@ -224,20 +243,8 @@ impl NextValFetcher for SequenceNextValFetcher {
         ctx: &QueryContext,
         catalog: &Arc<dyn Catalog>,
         to_fetch: u64,
-    ) -> Result<(u64, u64)> {
-        let visibility_checker = if ctx
-            .get_settings()
-            .get_enable_experimental_sequence_privilege_check()?
-        {
-            Some(ctx.get_visibility_checker(false, Object::Sequence).await?)
-        } else {
-            None
-        };
-
-        let req = GetSequenceReq {
-            ident: self.sequence_ident.clone(),
-        };
-        let resp = catalog.get_sequence(req, &visibility_checker).await?;
+    ) -> Result<NextValFetchResult> {
+        let (resp, visibility_checker) = self.get_sequence(ctx, catalog).await?;
         let step_size = resp.meta.step as u64;
 
         // Calculate batch size - take the larger of count or step_size
@@ -252,7 +259,42 @@ impl NextValFetcher for SequenceNextValFetcher {
         let resp = catalog
             .get_sequence_next_value(req, &visibility_checker)
             .await?;
-        Ok((resp.start, batch_size))
+        Ok(NextValFetchResult {
+            start: resp.start,
+            batch_size,
+            step: resp.step,
+        })
+    }
+
+    async fn step(&self, ctx: &QueryContext, catalog: &Arc<dyn Catalog>) -> Result<i64> {
+        self.get_sequence(ctx, catalog)
+            .await
+            .map(|(resp, _)| resp.meta.step)
+    }
+}
+
+impl SequenceNextValFetcher {
+    async fn get_sequence(
+        &self,
+        ctx: &QueryContext,
+        catalog: &Arc<dyn Catalog>,
+    ) -> Result<(GetSequenceReply, Option<GrantObjectVisibilityChecker>)> {
+        let visibility_checker = if ctx
+            .get_settings()
+            .get_enable_experimental_sequence_privilege_check()?
+        {
+            Some(ctx.get_visibility_checker(false, Object::Sequence).await?)
+        } else {
+            None
+        };
+
+        let req = GetSequenceReq {
+            ident: self.sequence_ident.clone(),
+        };
+        catalog
+            .get_sequence(req, &visibility_checker)
+            .await
+            .map(|reply| (reply, visibility_checker))
     }
 }
 
@@ -267,24 +309,31 @@ impl NextValFetcher for AutoIncrementNextValFetcher {
         ctx: &QueryContext,
         catalog: &Arc<dyn Catalog>,
         to_fetch: u64,
-    ) -> Result<(u64, u64)> {
+    ) -> Result<NextValFetchResult> {
         let step_size = self.expr.step as u64;
 
         // Calculate batch size - take the larger of count or step_size
         let batch_size = to_fetch.max(step_size);
+        let step = self.expr.step;
 
         // Calculate batch size - take the larger of count or step_size
         let req = GetAutoIncrementNextValueReq {
             tenant: ctx.get_tenant(),
             key: self.key,
             expr: self.expr,
-            // FIXME: count * step
-            // count: batch_size,
-            count: 1,
+            count: batch_size,
         };
 
         let resp = catalog.get_autoincrement_next_value(req).await?;
-        Ok((resp.start, batch_size))
+        Ok(NextValFetchResult {
+            start: resp.start,
+            batch_size,
+            step,
+        })
+    }
+
+    async fn step(&self, _ctx: &QueryContext, _catalog: &Arc<dyn Catalog>) -> Result<i64> {
+        Ok(self.expr.step)
     }
 }
 

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

@@ -1714,14 +1714,7 @@ impl Binder {
                     step,
                     is_ordered,
                 } => {
-                    if !matches!(
-                        field.data_type().remove_nullable(),
-                        TableDataType::Number(_) | TableDataType::Decimal(_)
-                    ) {
-                        return Err(ErrorCode::SemanticError(
-                            "AUTO INCREMENT only supports Decimal or Numeric (e.g. INT32) types",
-                        ));
-                    }
+                    Self::check_autoincrement_expr(&field, step, is_ordered)?;
                     field.auto_increment_expr = Some(AutoIncrementExpr {
                         column_id: table_schema.next_column_id(),
                         start: *start,
@@ -1742,6 +1735,31 @@ impl Binder {
         ))
     }
 
+    fn check_autoincrement_expr(field: &TableField, step: &i64, is_ordered: &bool) -> Result<()> {
+        if !matches!(
+            field.data_type().remove_nullable(),
+            TableDataType::Number(_) | TableDataType::Decimal(_)
+        ) {
+            return Err(ErrorCode::SemanticError(
+                "AUTOINCREMENT only supports Decimal or Numeric (e.g. INT32) types",
+            ));
+        }
+        if !is_ordered {
+            return Err(ErrorCode::SemanticError(
+                "AUTOINCREMENT only support ORDER now",
+            ));
+        }
+        if step == &0 {
+            return Err(ErrorCode::SemanticError("INCREMENT cannot be 0"));
+        }
+        if step < &0 {
+            return Err(ErrorCode::SemanticError(
+                "INCREMENT does not currently support negative numbers",
+            ));
+        }
+        Ok(())
+    }
+
     #[async_backtrace::framed]
     pub async fn analyze_create_table_schema_by_columns(
         &self,
@@ -1770,14 +1788,7 @@ impl Binder {
                         step,
                         is_ordered,
                     } => {
-                        if !matches!(
-                            field.data_type().remove_nullable(),
-                            TableDataType::Number(_) | TableDataType::Decimal(_)
-                        ) {
-                            return Err(ErrorCode::SemanticError(
-                                "AUTO INCREMENT only supports Decimal or Numeric (e.g. INT32) types",
-                            ));
-                        }
+                        Self::check_autoincrement_expr(&field, step, is_ordered)?;
                         has_autoincrement = true;
                         field.auto_increment_expr = Some(AutoIncrementExpr {
                             column_id: 0,

tests/sqllogictests/suites/base/05_ddl/05_0036_sequence.test

@@ -222,6 +222,31 @@ select * from t order by id;
 statement ok
 drop table t;
 
+statement ok
+CREATE OR REPLACE SEQUENCE seq start = 1 increment = 2;
+
+statement ok
+CREATE OR REPLACE table t2(id int default nextval(seq), name string);
+
+statement ok
+INSERT INTO t2(name) values('x'),('x'),('x');
+
+statement ok
+INSERT INTO t2(name) values('x'),('x'),('x');
+
+query IT
+select * from t2 order by id;
+----
+1 x
+3 x
+5 x
+7 x
+9 x
+11 x
+
+statement ok
+DROP TABLE t2
+
 statement ok
 DROP SEQUENCE seq;