diff --git a/contrib/shark/expected/identity.out b/contrib/shark/expected/identity.out
new file mode 100644
index 0000000000000000000000000000000000000000..dcaeff8016d2db1793781f485475175e2c6dd18b
--- /dev/null
+++ b/contrib/shark/expected/identity.out
@@ -0,0 +1,283 @@
+CREATE TABLE book(bookId int IDENTITY, bookname NVARCHAR(50), author NVARCHAR(50));
+NOTICE:  CREATE TABLE will create implicit sequence "book_bookid_seq_identity" for serial column "book.bookid"
+INSERT INTO book VALUES('book1','author1'),('book2','author2');
+INSERT INTO book(bookname,author) VALUES('book3','author3'),('book4','author4');
+INSERT INTO book VALUES(3,'book5','author5');
+ERROR:  INSERT has more expressions than target columns
+LINE 1: INSERT INTO book VALUES(3,'book5','author5');
+                                          ^
+select * from book;
+ bookid | bookname | author  
+--------+----------+---------
+      1 | book1    | author1
+      2 | book2    | author2
+      3 | book3    | author3
+      4 | book4    | author4
+(4 rows)
+
+ALTER table book add column id int identity;
+ERROR:  multiple identity specifications for column "bookid" of table "book"
+drop table if exists book;
+CREATE TABLE book(id int identity, bookId int IDENTITY, bookname NVARCHAR(50), author NVARCHAR(50));
+NOTICE:  CREATE TABLE will create implicit sequence "book_id_seq_identity" for serial column "book.id"
+ERROR:  table can only have one identity column
+CREATE TABLE book
+(
+    bookId int NOT NULL PRIMARY KEY IDENTITY, 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+NOTICE:  CREATE TABLE will create implicit sequence "book_bookid_seq_identity" for serial column "book.bookid"
+NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "book_pkey" for table "book"
+INSERT INTO book VALUES('book1','author1'),('book2','author2');
+select * from book;
+ bookid | bookname | author  
+--------+----------+---------
+      1 | book1    | author1
+      2 | book2    | author2
+(2 rows)
+
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId int NOT NULL PRIMARY KEY IDENTITY(3,5), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+NOTICE:  CREATE TABLE will create implicit sequence "book_bookid_seq_identity" for serial column "book.bookid"
+NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "book_pkey" for table "book"
+INSERT INTO book VALUES('book1','author1'),('book2','author2');
+select * from book;
+ bookid | bookname | author  
+--------+----------+---------
+      3 | book1    | author1
+      8 | book2    | author2
+(2 rows)
+
+delete book where bookId=8;
+select * from book;
+ bookid | bookname | author  
+--------+----------+---------
+      3 | book1    | author1
+(1 row)
+
+INSERT INTO book(bookname,author) VALUES('book3','author3'),('book4','author4');
+select * from book;
+ bookid | bookname | author  
+--------+----------+---------
+      3 | book1    | author1
+     13 | book3    | author3
+     18 | book4    | author4
+(3 rows)
+
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId tinyint NOT NULL PRIMARY KEY IDENTITY(3,5), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+NOTICE:  CREATE TABLE will create implicit sequence "book_bookid_seq_identity" for serial column "book.bookid"
+NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "book_pkey" for table "book"
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId smallint NOT NULL PRIMARY KEY IDENTITY(3,5), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+NOTICE:  CREATE TABLE will create implicit sequence "book_bookid_seq_identity" for serial column "book.bookid"
+NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "book_pkey" for table "book"
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId bigint NOT NULL PRIMARY KEY IDENTITY(3,5), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+NOTICE:  CREATE TABLE will create implicit sequence "book_bookid_seq_identity" for serial column "book.bookid"
+NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "book_pkey" for table "book"
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId numeric NOT NULL PRIMARY KEY IDENTITY(3,5), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+NOTICE:  CREATE TABLE will create implicit sequence "book_bookid_seq_identity" for serial column "book.bookid"
+NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "book_pkey" for table "book"
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId decimal NOT NULL PRIMARY KEY IDENTITY(3,5), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+NOTICE:  CREATE TABLE will create implicit sequence "book_bookid_seq_identity" for serial column "book.bookid"
+NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "book_pkey" for table "book"
+drop table if exists book;
+CREATE TABLE book
+(
+    bookname NVARCHAR(50),
+    author NVARCHAR(50)
+);
+ALTER TABLE book add bookId int not null primary key identity;
+NOTICE:  ALTER TABLE will create implicit sequence "book_bookid_seq_identity" for serial column "book.bookid"
+NOTICE:  ALTER TABLE / ADD PRIMARY KEY will create implicit index "book_pkey" for table "book"
+INSERT INTO book VALUES('book1','author1'),('book2','author2');
+select * from book;
+ bookname | author  | bookid 
+----------+---------+--------
+ book1    | author1 |      1
+ book2    | author2 |      2
+(2 rows)
+
+delete book where bookId=8;
+select * from book;
+ bookname | author  | bookid 
+----------+---------+--------
+ book1    | author1 |      1
+ book2    | author2 |      2
+(2 rows)
+
+INSERT INTO book(bookname,author) VALUES('book3','author3'),('book4','author4');
+select * from book;
+ bookname | author  | bookid 
+----------+---------+--------
+ book1    | author1 |      1
+ book2    | author2 |      2
+ book3    | author3 |      3
+ book4    | author4 |      4
+(4 rows)
+
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId int NOT NULL PRIMARY KEY IDENTITY(100,100),
+    bookname NVARCHAR(50),
+    author NVARCHAR(50)
+);
+NOTICE:  CREATE TABLE will create implicit sequence "book_bookid_seq_identity" for serial column "book.bookid"
+NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "book_pkey" for table "book"
+INSERT INTO book VALUES('book1','author1'),('book2','author2');
+INSERT INTO book(bookname,author) VALUES('book3','author3'),('book4','author4');
+select * from book;
+ bookid | bookname | author  
+--------+----------+---------
+    100 | book1    | author1
+    200 | book2    | author2
+    300 | book3    | author3
+    400 | book4    | author4
+(4 rows)
+
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId tinyint NOT NULL PRIMARY KEY IDENTITY(-1,1), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+NOTICE:  CREATE TABLE will create implicit sequence "book_bookid_seq_identity" for serial column "book.bookid"
+ERROR:  Identity column contains invalid SEED.
+drop table if exists book;
+NOTICE:  table "book" does not exist, skipping
+CREATE TABLE book
+(
+    bookId tinyint NOT NULL PRIMARY KEY IDENTITY(0,-1), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+NOTICE:  CREATE TABLE will create implicit sequence "book_bookid_seq_identity" for serial column "book.bookid"
+ERROR:  Identity column contains invalid INCREMENT.
+drop table if exists book;
+NOTICE:  table "book" does not exist, skipping
+CREATE TABLE book
+(
+    bookId tinyint NOT NULL PRIMARY KEY IDENTITY(255,256), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+NOTICE:  CREATE TABLE will create implicit sequence "book_bookid_seq_identity" for serial column "book.bookid"
+ERROR:  Identity column contains invalid INCREMENT.
+drop table if exists book;
+NOTICE:  table "book" does not exist, skipping
+CREATE TABLE book
+(
+    bookId smallint NOT NULL PRIMARY KEY IDENTITY(32767,32768), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+NOTICE:  CREATE TABLE will create implicit sequence "book_bookid_seq_identity" for serial column "book.bookid"
+ERROR:  Identity column contains invalid INCREMENT.
+drop table if exists book;
+NOTICE:  table "book" does not exist, skipping
+CREATE TABLE book
+(
+    bookId text NOT NULL PRIMARY KEY IDENTITY, 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+ERROR:  Identity column 'bookid' must be of data type int, bigint, smallint, tinyint, or decimal or numeric with a scale of 0, unencrypted, and constrained to be nonnullable.
+drop table if exists book;
+NOTICE:  table "book" does not exist, skipping
+CREATE TABLE book
+(
+    bookId numeric(4,2) NOT NULL PRIMARY KEY IDENTITY, 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+ERROR:  Identity column 'bookid' must be of data type int, bigint, smallint, tinyint, or decimal or numeric with a scale of 0, unencrypted, and constrained to be nonnullable.
+CREATE TABLE book(id serial, bookId int identity default 5, bookname char(20));
+NOTICE:  CREATE TABLE will create implicit sequence "book_id_seq" for serial column "book.id"
+NOTICE:  CREATE TABLE will create implicit sequence "book_bookid_seq_identity" for serial column "book.bookid"
+ERROR:  multiple default values specified for column "bookid" of table "book"
+CREATE TABLE t_identity(id serial, col int identity(5,10), col2 text)
+partition by range(col)(
+partition p1 VALUES LESS THAN (10),
+partition p2 VALUES LESS THAN (20),
+partition p3 VALUES LESS THAN (30),
+partition p4 VALUES LESS THAN (MAXVALUE)
+);
+NOTICE:  CREATE TABLE will create implicit sequence "t_identity_id_seq" for serial column "t_identity.id"
+NOTICE:  CREATE TABLE will create implicit sequence "t_identity_col_seq_identity" for serial column "t_identity.col"
+insert into t_identity(col2) values('abc');
+insert into t_identity(col2) values('def');
+insert into t_identity(col2) values('ghi');
+insert into t_identity(col2) values('jkl');
+select * from t_identity partition(p1);
+ id | col | col2 
+----+-----+------
+  1 |   5 | abc
+(1 row)
+
+select * from t_identity partition(p2);
+ id | col | col2 
+----+-----+------
+  2 |  15 | def
+(1 row)
+
+select * from t_identity partition(p3);
+ id | col | col2 
+----+-----+------
+  3 |  25 | ghi
+(1 row)
+
+select * from t_identity partition(p4);
+ id | col | col2 
+----+-----+------
+  4 |  35 | jkl
+(1 row)
+
+drop table t_identity;
+CREATE TABLE t_identity(id decimal(12) not null identity(1,1), col text);
+NOTICE:  CREATE TABLE will create implicit sequence "t_identity_id_seq_identity" for serial column "t_identity.id"
+insert into t_identity values('aaa');
+select * from t_identity;
+ id | col 
+----+-----
+  1 | aaa
+(1 row)
+
+drop table if exists t_identity;
+CREATE TABLE t_identity(id decimal(12,2) not null identity(1,1), col text);
+ERROR:  Identity column 'id' must be of data type int, bigint, smallint, tinyint, or decimal or numeric with a scale of 0, unencrypted, and constrained to be nonnullable.
diff --git a/contrib/shark/input/gs_dump_d_format.source b/contrib/shark/input/gs_dump_d_format.source
index e0cfa3095dbcadd04130e0732b01c82de341ae8a..e7a531da30bf7eadf444c99439423e4c69fad2a1 100644
--- a/contrib/shark/input/gs_dump_d_format.source
+++ b/contrib/shark/input/gs_dump_d_format.source
@@ -9,6 +9,8 @@ CREATE TABLE mysche.[t2](a int);
 CREATE TABLE [mysche].[t3](a int);
 INSERT INTO mysche.t1 (a) VALUES (1), (2), (3);
 CREATE VIEW [t4] AS SELECT * FROM mysche.t1;
+CREATE TABLE table_ident(id int identity, name char(20));
+INSERT INTO table_ident values('aaa'),('bbb'),('ccc');
 
 \! @abs_bindir@/gs_dump dump_d_database -p @portstring@ -f @abs_bindir@/dump_d.tar -F t >/dev/null 2>&1; echo $?
 \! @abs_bindir@/gs_restore -d restore_d_database -p @portstring@ @abs_bindir@/dump_d.tar >/dev/null 2>&1; echo $?
@@ -19,6 +21,7 @@ SELECT * from mysche.t1;
 SELECT * from mysche.t2;
 SELECT * from mysche.t3;
 SELECT * from t4;
+SELECT * from table_ident;
 
 \c contrib_regression
 drop database dump_d_database;
diff --git a/contrib/shark/output/gs_dump_d_format.source b/contrib/shark/output/gs_dump_d_format.source
index 866407378d62fae5915a5146e4eee73c30150017..62d2317521bc1f4ab9b4091eec6bded3e63eec33 100644
--- a/contrib/shark/output/gs_dump_d_format.source
+++ b/contrib/shark/output/gs_dump_d_format.source
@@ -9,6 +9,9 @@ CREATE TABLE mysche.[t2](a int);
 CREATE TABLE [mysche].[t3](a int);
 INSERT INTO mysche.t1 (a) VALUES (1), (2), (3);
 CREATE VIEW [t4] AS SELECT * FROM mysche.t1;
+CREATE TABLE table_ident(id int identity, name char(20));
+NOTICE:  CREATE TABLE will create implicit sequence "table_ident_id_seq_identity" for serial column "table_ident.id"
+INSERT INTO table_ident values('aaa'),('bbb'),('ccc');
 \! @abs_bindir@/gs_dump dump_d_database -p @portstring@ -f @abs_bindir@/dump_d.tar -F t >/dev/null 2>&1; echo $?
 0
 \! @abs_bindir@/gs_restore -d restore_d_database -p @portstring@ @abs_bindir@/dump_d.tar >/dev/null 2>&1; echo $?
@@ -43,6 +46,14 @@ SELECT * from t4;
  3
 (3 rows)
 
+SELECT * from table_ident;
+ id |         name         
+----+----------------------
+  1 | aaa                 
+  2 | bbb                 
+  3 | ccc                 
+(3 rows)
+
 \c contrib_regression
 drop database dump_d_database;
 drop database restore_d_database;
diff --git a/contrib/shark/parallel_schedule b/contrib/shark/parallel_schedule
index 23bb475c995fdc460a34f49f90a2295807cec5b7..e892ad30ce8b1f7bbb382c572a7765f530f37d50 100644
--- a/contrib/shark/parallel_schedule
+++ b/contrib/shark/parallel_schedule
@@ -4,6 +4,7 @@ test: basetest
 test: day_test rand_test
 test: objectproperty
 test: clustered_index insert_stmt sbr_test
+test: identity
 test: test_ddl_and_dml
 test: rotate-test-part2 rotate-test-part2_ci
 test: functions
diff --git a/contrib/shark/sql/identity.sql b/contrib/shark/sql/identity.sql
new file mode 100644
index 0000000000000000000000000000000000000000..77e785e9eed59a06295dad43f75e559bce6bf676
--- /dev/null
+++ b/contrib/shark/sql/identity.sql
@@ -0,0 +1,163 @@
+CREATE TABLE book(bookId int IDENTITY, bookname NVARCHAR(50), author NVARCHAR(50));
+INSERT INTO book VALUES('book1','author1'),('book2','author2');
+INSERT INTO book(bookname,author) VALUES('book3','author3'),('book4','author4');
+INSERT INTO book VALUES(3,'book5','author5');
+select * from book;
+ALTER table book add column id int identity;
+drop table if exists book;
+
+CREATE TABLE book(id int identity, bookId int IDENTITY, bookname NVARCHAR(50), author NVARCHAR(50));
+
+CREATE TABLE book
+(
+    bookId int NOT NULL PRIMARY KEY IDENTITY, 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+INSERT INTO book VALUES('book1','author1'),('book2','author2');
+select * from book;
+drop table if exists book;
+
+CREATE TABLE book
+(
+    bookId int NOT NULL PRIMARY KEY IDENTITY(3,5), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+INSERT INTO book VALUES('book1','author1'),('book2','author2');
+select * from book;
+delete book where bookId=8;
+select * from book;
+INSERT INTO book(bookname,author) VALUES('book3','author3'),('book4','author4');
+select * from book;
+drop table if exists book;
+
+CREATE TABLE book
+(
+    bookId tinyint NOT NULL PRIMARY KEY IDENTITY(3,5), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId smallint NOT NULL PRIMARY KEY IDENTITY(3,5), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId bigint NOT NULL PRIMARY KEY IDENTITY(3,5), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId numeric NOT NULL PRIMARY KEY IDENTITY(3,5), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId decimal NOT NULL PRIMARY KEY IDENTITY(3,5), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+drop table if exists book;
+CREATE TABLE book
+(
+    bookname NVARCHAR(50),
+    author NVARCHAR(50)
+);
+ALTER TABLE book add bookId int not null primary key identity;
+INSERT INTO book VALUES('book1','author1'),('book2','author2');
+select * from book;
+delete book where bookId=8;
+select * from book;
+INSERT INTO book(bookname,author) VALUES('book3','author3'),('book4','author4');
+select * from book;
+drop table if exists book;
+
+CREATE TABLE book
+(
+    bookId int NOT NULL PRIMARY KEY IDENTITY(100,100),
+    bookname NVARCHAR(50),
+    author NVARCHAR(50)
+);
+INSERT INTO book VALUES('book1','author1'),('book2','author2');
+INSERT INTO book(bookname,author) VALUES('book3','author3'),('book4','author4');
+select * from book;
+drop table if exists book;
+
+
+CREATE TABLE book
+(
+    bookId tinyint NOT NULL PRIMARY KEY IDENTITY(-1,1), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId tinyint NOT NULL PRIMARY KEY IDENTITY(0,-1), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId tinyint NOT NULL PRIMARY KEY IDENTITY(255,256), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId smallint NOT NULL PRIMARY KEY IDENTITY(32767,32768), 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId text NOT NULL PRIMARY KEY IDENTITY, 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+drop table if exists book;
+CREATE TABLE book
+(
+    bookId numeric(4,2) NOT NULL PRIMARY KEY IDENTITY, 
+    bookname NVARCHAR(50), 
+    author NVARCHAR(50)
+);
+
+CREATE TABLE book(id serial, bookId int identity default 5, bookname char(20));
+
+CREATE TABLE t_identity(id serial, col int identity(5,10), col2 text)
+partition by range(col)(
+partition p1 VALUES LESS THAN (10),
+partition p2 VALUES LESS THAN (20),
+partition p3 VALUES LESS THAN (30),
+partition p4 VALUES LESS THAN (MAXVALUE)
+);
+insert into t_identity(col2) values('abc');
+insert into t_identity(col2) values('def');
+insert into t_identity(col2) values('ghi');
+insert into t_identity(col2) values('jkl');
+select * from t_identity partition(p1);
+select * from t_identity partition(p2);
+select * from t_identity partition(p3);
+select * from t_identity partition(p4);
+drop table t_identity;
+
+CREATE TABLE t_identity(id decimal(12) not null identity(1,1), col text);
+insert into t_identity values('aaa');
+select * from t_identity;
+drop table if exists t_identity;
+
+CREATE TABLE t_identity(id decimal(12,2) not null identity(1,1), col text);
+
diff --git a/contrib/shark/src/backend_parser/gram-tsql-decl.y b/contrib/shark/src/backend_parser/gram-tsql-decl.y
index b284521b6a4f71397ad660e0733d34d142aa6ede..9f25bddeddcc3dc485e4fb5e616d96a865f034e6 100644
--- a/contrib/shark/src/backend_parser/gram-tsql-decl.y
+++ b/contrib/shark/src/backend_parser/gram-tsql-decl.y
@@ -8,5 +8,6 @@
 %type <node> TSQL_computed_column
 %type <node> tsql_top_clause tsql_select_top_value
 %type <boolean> tsql_opt_ties tsql_opt_percent
+%type <list> identity_seed_increment
 %type <str> DirectColLabel
 %type <keyword> direct_label_keyword
diff --git a/contrib/shark/src/backend_parser/gram-tsql-nonassoc-ident-tokens b/contrib/shark/src/backend_parser/gram-tsql-nonassoc-ident-tokens
index 8bb73b236ee8dcd8addff5b6ec319c44ac03d373..056d456ff1262715fcd855ae1e70618211b6abae 100644
--- a/contrib/shark/src/backend_parser/gram-tsql-nonassoc-ident-tokens
+++ b/contrib/shark/src/backend_parser/gram-tsql-nonassoc-ident-tokens
@@ -1 +1,2 @@
-TSQL_PERSISTED /* these tokens can follow b_expr. To resolve ambiguity, we need to assign the same priority with IDENT. please see the comment in gram.y for details */
\ No newline at end of file
+IDENTITY_P
+TSQL_PERSISTED /* these tokens can follow b_expr. To resolve ambiguity, we need to assign the same priority with IDENT. please see the comment in gram.y for details */
diff --git a/contrib/shark/src/backend_parser/gram-tsql-rule.y b/contrib/shark/src/backend_parser/gram-tsql-rule.y
index 61f3343307c1d14c3b83fd884e3250e05aa1eb1c..34570f6ad36b33c551063583ac3ac68b3a841674 100644
--- a/contrib/shark/src/backend_parser/gram-tsql-rule.y
+++ b/contrib/shark/src/backend_parser/gram-tsql-rule.y
@@ -462,6 +462,27 @@ tsql_CreateProcedureStmt:
 				}
 		;
 
+ColConstraintElem:     IDENTITY_P identity_seed_increment
+                            {
+                                Constraint *n = makeNode(Constraint);
+                                n->contype = CONSTR_IDENTITY;
+                                n->generated_when = ATTRIBUTE_IDENTITY_ALWAYS;
+                                n->options = $2;
+                                n->location = @1;
+                                $$ = (Node *)n;
+                            }
+                        ;
+
+identity_seed_increment:
+                       '(' NumericOnly ',' NumericOnly ')'
+                       {
+                           $$ = list_make2(makeDefElem("start", (Node *)$2), makeDefElem("increment",(Node *)$4));
+                       }
+                       | /* EMPTY */
+                       {
+                           $$ = list_make2(makeDefElem("start", (Node *)makeInteger(1)), makeDefElem("increment", (Node *)makeInteger(1)));
+                       }
+                       ;
 rotate_clause:
 		ROTATE '(' func_application_list rotate_for_clause rotate_in_clause ')' alias_clause %prec ROTATE
 			{
diff --git a/src/common/backend/parser/analyze.cpp b/src/common/backend/parser/analyze.cpp
index 97b18c8a097dc9dc0ed71e4cfbea29767198bc01..8f369404c46e8465115b9117a784bc978050a94b 100644
--- a/src/common/backend/parser/analyze.cpp
+++ b/src/common/backend/parser/analyze.cpp
@@ -87,6 +87,7 @@
 #include "utils/typcache.h"
 #include "commands/explain.h"
 #include "commands/sec_rls_cmds.h"
+#include "commands/sequence.h"
 #include "streaming/streaming_catalog.h"
 #include "instruments/instr_unique_sql.h"
 #include "streaming/init.h"
diff --git a/src/common/backend/parser/parse_target.cpp b/src/common/backend/parser/parse_target.cpp
index c2c41a50bc4d2f805f6254241a68665074316182..bf81989a3fc0dca0fb6da60783c16189822d85ed 100644
--- a/src/common/backend/parser/parse_target.cpp
+++ b/src/common/backend/parser/parse_target.cpp
@@ -39,6 +39,7 @@
 #include "gs_ledger/ledger_utils.h"
 #include "mb/pg_wchar.h"
 #include "parser/parse_utilcmd.h"
+#include "commands/sequence.h"
 
 static void markTargetListOrigin(ParseState* pstate, TargetEntry* tle, Var* var, int levelsup);
 static Node* transformAssignmentIndirection(ParseState* pstate, Node* basenode, const char* targetName,
@@ -922,6 +923,27 @@ List* checkInsertTargets(ParseState* pstate, List* cols, List** attrnos)
             if (is_blockchain_rel && strcmp(col->name, "hash") == 0) {
                 continue;
             }
+            /* skip the identity default value in D format */
+            if (u_sess->attr.attr_sql.sql_compatibility == D_FORMAT) {
+                FunctionCallInfoData funcinfo;
+                Datum re_seq;
+                char* sequenceName = NULL;
+                int nargs = 2;
+
+                InitFunctionCallInfoData(funcinfo, NULL, nargs, InvalidOid, NULL, NULL);
+
+                funcinfo.arg[0] = CStringGetTextDatum(RelationGetRelationName(targetrel));
+                funcinfo.arg[1] = CStringGetTextDatum(col->name);
+
+                re_seq = pg_get_serial_sequence(&funcinfo);
+                if (re_seq != NULL) {
+                    sequenceName = TextDatumGetCString(re_seq);
+                    if (sequenceName != NULL && StrEndWith(sequenceName, "_seq_identity")) {
+                        continue;
+                    }
+                }
+            }
+
             col->indirection = NIL;
             col->val = NULL;
             col->location = -1;
diff --git a/src/common/backend/parser/parse_utilcmd.cpp b/src/common/backend/parser/parse_utilcmd.cpp
index 7667b668f96ef431f705cc2173d92dac8c564328..344e9118f11cb60b82b784a947331ad9e9852bec 100644
--- a/src/common/backend/parser/parse_utilcmd.cpp
+++ b/src/common/backend/parser/parse_utilcmd.cpp
@@ -156,6 +156,8 @@ static void transformColumnType(CreateStmtContext* cxt, ColumnDef* column);
 static void setSchemaName(char* context_schema, char** stmt_schema_name);
 static void TransformTempAutoIncrement(ColumnDef* column, CreateStmt* stmt);
 static int128 TransformAutoIncStart(CreateStmt* stmt);
+static void identity_type_dmod(ColumnDef* column);
+static int identity_only(ColumnDef* column);
 
 /*
  * @hdfs
@@ -589,6 +591,7 @@ Oid *namespaceid, bool isFirstNode)
      * Run through each primary element in the table creation clause. Separate
      * column defs from constraints, and do preliminary analysis.
      */
+    int count = 0;
     foreach (elements, stmt->tableElts) {
         TableLikeClause* tbl_like_clause = NULL;
         TransformTableType transformType = TRANSFORM_INVALID;
@@ -604,6 +607,14 @@ Oid *namespaceid, bool isFirstNode)
                         (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
                             errmsg("\"hash\" column is reserved for system in ledger schema.")));
                 }
+
+                count += identity_only((ColumnDef*)element);
+                if (count > 1) {
+                    ereport(ERROR,
+                        (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
+                            errmsg("table can only have one identity column")));
+                }
+
                 transformColumnDefinition(&cxt, (ColumnDef*)element, !isFirstNode && preCheck);
 
                 if (((ColumnDef *)element)->clientLogicColumnRef != NULL) {
@@ -1011,7 +1022,8 @@ static bool DropSetOwnedByTable(CreateStmtContext* cxt, char *colname)
  *		create a sequence owned by table, need to add record to pg_depend.
  *		used in CREATE TABLE and CREATE TABLE ... LIKE
  */
-static void createSeqOwnedByTable(CreateStmtContext* cxt, ColumnDef* column, bool preCheck, bool large, bool is_autoinc)
+static void createSeqOwnedByTable(CreateStmtContext* cxt, ColumnDef* column, bool preCheck, bool large,
+    bool isAutoinc, bool forIdentity = false, List *seqOptions = nullptr)
 {
     Oid snamespaceid;
     char* snamespace = NULL;
@@ -1025,6 +1037,54 @@ static void createSeqOwnedByTable(CreateStmtContext* cxt, ColumnDef* column, boo
     AlterSeqStmt* altseqstmt = NULL;
     List* attnamelist = NIL;
     Constraint* constraint = NULL;
+    Oid typeOid = InvalidOid;
+
+    if (column->typname) {
+        typeOid = typenameTypeId(NULL, column->typname);
+    }
+
+    if (forIdentity && nodeTag(cxt->node) == T_AlterTableStmt) {
+        Relation attRelation;
+        Relation rel;
+        SysScanDesc scan;
+        ScanKeyData key[1];
+        HeapTuple tuple;
+        Form_pg_attribute attrForm;
+        char* columnName;
+        FunctionCallInfoData funcinfo;
+        char* sequenceName;
+        Datum re_seq;
+
+        InitFunctionCallInfoData(funcinfo, NULL, 2, InvalidOid, NULL, NULL);
+        attRelation = heap_open(AttributeRelationId, AccessShareLock);
+        
+        rel = heap_openrv(cxt->relation, AccessShareLock);
+    
+        ScanKeyInit(&key[0], Anum_pg_attribute_attrelid, BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(RelationGetRelid(rel)));
+    
+        scan = systable_beginscan(attRelation, AttributeRelidNumIndexId, true, NULL, 1, key);
+        while ((tuple = systable_getnext(scan)) != NULL) {
+            attrForm = (Form_pg_attribute) GETSTRUCT(tuple);
+            columnName = NameStr(attrForm->attname);
+            funcinfo.arg[0] = CStringGetTextDatum(cxt->relation->relname);
+            funcinfo.arg[1] = CStringGetTextDatum(columnName);
+            re_seq = pg_get_serial_sequence(&funcinfo);
+            if (re_seq != NULL) {
+                sequenceName = TextDatumGetCString(re_seq);
+            } else {
+                continue;
+            }
+
+            if (sequenceName != NULL && StrEndWith(sequenceName, "_seq_identity"))
+                ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR),
+                                       errmsg("multiple identity specifications for column \"%s\" of table \"%s\"",
+                                       columnName, cxt->relation->relname),
+                                       parser_errposition(cxt->pstate, -1)));
+        }
+        systable_endscan(scan);
+        heap_close(rel, AccessShareLock);
+        heap_close(attRelation, AccessShareLock);
+    }
 
     /*
      * Determine namespace and name to use for the sequence.
@@ -1044,7 +1104,12 @@ static void createSeqOwnedByTable(CreateStmtContext* cxt, ColumnDef* column, boo
         RangeVarAdjustRelationPersistence(cxt->relation, snamespaceid);
     }
     snamespace = get_namespace_name(snamespaceid);
-    sname = ChooseRelationName(cxt->relation->relname, column->colname, "seq", strlen("seq"), snamespaceid);
+
+    if (forIdentity) {
+        sname = ChooseRelationName(cxt->relation->relname, column->colname, "seq_identity", strlen("seq_identity"), snamespaceid);
+    } else {
+        sname = ChooseRelationName(cxt->relation->relname, column->colname, "seq", strlen("seq"), snamespaceid);
+    }
 
     if (!preCheck || IS_SINGLE_NODE)
         ereport(NOTICE,
@@ -1061,13 +1126,27 @@ static void createSeqOwnedByTable(CreateStmtContext* cxt, ColumnDef* column, boo
      */
     seqstmt = makeNode(CreateSeqStmt);
     seqstmt->sequence = makeRangeVar(snamespace, sname, -1);
-    seqstmt->options = is_autoinc ? GetAutoIncSeqOptions(cxt) : NULL;
-    seqstmt->is_autoinc = is_autoinc;
+    seqstmt->options = seqOptions;
+    seqstmt->is_autoinc = isAutoinc;
 #ifdef PGXC
     seqstmt->is_serial = true;
 #endif
     seqstmt->is_large = large;
 
+    if (forIdentity) {
+        int typemod = 0;
+        if (column->typname->typmods != NULL) {
+            typemod = intVal(&((A_Const*)lfirst(list_head(column->typname->typmods)))->val);
+            seqstmt->options = lcons(makeDefElem("as", (Node *)makeTypeNameFromOid(typeOid, typemod)), seqstmt->options);
+        } else {
+            seqstmt->options = lcons(makeDefElem("as", (Node *)makeTypeNameFromOid(typeOid, -1)), seqstmt->options);
+        }
+    } else if (!large) {
+        seqstmt->options = list_make1(makeDefElem("as", (Node *)makeTypeNameFromOid(typeOid, -1)));
+    } else {
+        seqstmt->options = isAutoinc? GetAutoIncSeqOptions(cxt) : NULL;
+    }
+
     /* Assign UUID for create sequence */
     seqstmt->uuid = INVALIDSEQUUID;
 #ifdef ENABLE_MUTIPLE_NODES
@@ -1112,7 +1191,7 @@ static void createSeqOwnedByTable(CreateStmtContext* cxt, ColumnDef* column, boo
     attnamelist = list_make3(makeString(snamespace), makeString(cxt->relation->relname), makeString(column->colname));
     altseqstmt->options = list_make1(makeDefElem("owned_by", (Node*)attnamelist));
     altseqstmt->is_large = large;
-    altseqstmt->is_autoinc = is_autoinc;
+    altseqstmt->is_autoinc = isAutoinc;
 
     cxt->alist = lappend(cxt->alist, altseqstmt);
 
@@ -1149,7 +1228,7 @@ static void createSeqOwnedByTable(CreateStmtContext* cxt, ColumnDef* column, boo
     constraint = makeNode(Constraint);
     constraint->contype = CONSTR_DEFAULT;
     constraint->location = -1;
-    if (is_autoinc) {
+    if (isAutoinc) {
         autoincnode = makeNode(AutoIncrement);
         autoincnode->expr = (Node*)funccallnode;
         constraint->raw_expr = (Node*)autoincnode;
@@ -1160,7 +1239,7 @@ static void createSeqOwnedByTable(CreateStmtContext* cxt, ColumnDef* column, boo
     column->constraints = lappend(column->constraints, constraint);
     column->raw_default = constraint->raw_expr;
 
-    if (!is_autoinc) {
+    if (!isAutoinc) {
         constraint = makeNode(Constraint);
         constraint->contype = CONSTR_NOTNULL;
         constraint->location = -1;
@@ -1395,6 +1474,7 @@ static void TransformColumnDefinitionConstraints(CreateStmtContext* cxt, ColumnD
     bool saw_generated = false;
     Constraint* constraint = NULL;
     ListCell* clist = NULL;
+    bool sawIdentity = false;
 
     foreach (clist, column->constraints) {
         constraint = (Constraint*)lfirst(clist);
@@ -1442,6 +1522,33 @@ static void TransformColumnDefinitionConstraints(CreateStmtContext* cxt, ColumnD
                 saw_default = true;
                 break;
 
+            case CONSTR_IDENTITY:
+                {
+                    bool large = typenameTypeId(NULL, column->typname) == NUMERICOID;
+
+                    if (cxt->ofType) {
+                        ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+                           errmsg("identity columns are not supported on typed tables")));
+                    }
+                    if (sawIdentity) {
+                        ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR),
+                                       errmsg("multiple identity specifications for column \"%s\" of table \"%s\"",
+                                       column->colname, cxt->relation->relname),
+                                       parser_errposition(cxt->pstate, constraint->location)));
+                    }
+                    if (saw_default)
+                        ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR),
+                                       errmsg("Defaults cannot be created on columns with an IDENTITY attribute. Table '%s', column '%s'",
+                                       cxt->relation->relname, column->colname),
+                                       parser_errposition(cxt->pstate, constraint->location)));
+
+                    identity_type_dmod(column);
+                    createSeqOwnedByTable(cxt, column, preCheck, large, false, true, constraint->options);
+                    sawIdentity = true;
+                    column->is_not_null = true;
+                    break;
+                }
+
             case CONSTR_GENERATED:
                 if (cxt->ofType) {
                     ereport(ERROR, (errmodule(MOD_GEN_COL), errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
@@ -8581,3 +8688,43 @@ static void TransformModifyColumndef(CreateStmtContext* cxt, AlterTableCmd* cmd)
         cxt->blist = lappend(cxt->blist, rename);
     }
 }
+
+static int identity_only(ColumnDef* column)
+{
+    ListCell* clist = NULL;
+    Constraint* constraint = NULL;
+    int result = 0;
+    foreach (clist, column->constraints) {
+        constraint = (Constraint*)lfirst(clist);
+
+        if (constraint->contype == CONSTR_IDENTITY) {
+            result += 1;
+        }
+    }
+    return result;
+}
+
+/*
+ * Get the column type which is constrained by identity in D mode.
+ * Check the type is expected or not, and report error for invalid type.
+ */
+static void identity_type_dmod(ColumnDef* column)
+{
+    int s = 0;
+    Oid newtypid = typenameTypeId(NULL, column->typname);
+    if (newtypid == NUMERICOID &&
+        column->typname->typmods != NULL && 
+        list_length(column->typname->typmods) == 2) {
+            s = intVal(&((A_Const*)lfirst(list_tail(column->typname->typmods)))->val);
+    }
+
+    if (newtypid != INT2OID &&
+        newtypid != INT4OID &&
+        newtypid != INT8OID &&
+        newtypid != INT1OID &&
+        newtypid != INT16OID &&
+        (newtypid != NUMERICOID || (newtypid == NUMERICOID && s != 0)))
+            ereport(ERROR,
+                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+                     errmsg("Identity column '%s' must be of data type int, bigint, smallint, tinyint, or decimal or numeric with a scale of 0, unencrypted, and constrained to be nonnullable.", column->colname)));
+}
diff --git a/src/gausskernel/optimizer/commands/sequence/sequence.cpp b/src/gausskernel/optimizer/commands/sequence/sequence.cpp
index edb1a5f49b933f28479db1cb02196ce1c323c291..ba473af2f61a1f2b4e1fcd4beedc29929269b1de 100644
--- a/src/gausskernel/optimizer/commands/sequence/sequence.cpp
+++ b/src/gausskernel/optimizer/commands/sequence/sequence.cpp
@@ -77,10 +77,10 @@ static ObjectAddress AlterSequence(const AlterSeqStmt* stmt);
 #ifdef PGXC
 template<typename T_Form, typename T_Int, bool large>
 static void init_params(List* options, bool isInit, bool isUseLocalSeq, void* newm_p, List** owned_by,
-    bool* is_restart, bool* need_seq_rewrite);
+    bool* is_restart, bool* needSeqRewrite, bool isIdentity);
 #else
 static void init_params(List* options, bool isInit, bool isUseLocalSeq, void* newm_p, List** owned_by,
-    bool* need_seq_rewrite);
+    bool* needSeqRewrite, bool isIdentity);
 #endif
 template<typename T_Form, typename T_Int, bool large>
 static void do_setval(Oid relid, int128 next, bool iscalled, bool skip_perm_check = false);
@@ -913,22 +913,23 @@ static ObjectAddress DefineSequence(CreateSeqStmt* seq)
             (errcode(ERRCODE_DATA_CORRUPTED), errmsg("Invaild UUID for CREATE SEQUENCE %s.", seq->sequence->relname)));
 #endif
 
+    bool isIdentity = StrEndWith(seq->sequence->relname, "_seq_identity");
     /* Check and set all option values */
 #ifdef PGXC
     if (large) {
         init_params<Form_pg_large_sequence, int128, true>(seq->options, true, isUseLocalSeq, &newm, &owned_by,
-            &is_restart, &need_seq_rewrite);
+            &is_restart, &need_seq_rewrite, isIdentity);
     } else {
         init_params<Form_pg_sequence, int64, false>(seq->options, true, isUseLocalSeq, &newm, &owned_by,
-            &is_restart, &need_seq_rewrite);
+            &is_restart, &need_seq_rewrite, isIdentity);
     }
 #else
     if (large) {
         init_params<Form_pg_large_sequence, int128, true>(seq->options, true, isUseLocalSeq, &newm, &owned_by,
-            &need_seq_rewrite);
+            &need_seq_rewrite, isIdentity);
     } else {
         init_params<Form_pg_sequence, int64, false>(seq->options, true, isUseLocalSeq, &newm, &owned_by,
-            &need_seq_rewrite);
+            &need_seq_rewrite, isIdentity);
     }
 #endif
     /*
@@ -1221,6 +1222,7 @@ static ObjectAddress AlterSequence(const AlterSeqStmt* stmt)
 #endif
     bool need_seq_rewrite = false;
     ObjectAddress address;
+    bool isIdentity = false;
 
     /* Open and lock sequence. */
     relid = RangeVarGetRelid(stmt->sequence, ShareRowExclusiveLock, stmt->missing_ok);
@@ -1265,13 +1267,15 @@ static ObjectAddress AlterSequence(const AlterSeqStmt* stmt)
 
     newm = (T_Form)GETSTRUCT(tuple);
 
+    isIdentity = StrEndWith(RelationGetRelationName(seqrel), "_seq_identity");
+
     /* Check and set new values */
 #ifdef PGXC
     init_params<T_Form, T_Int, large>(stmt->options, false, isUseLocalSeq, newm, &owned_by,
-        &is_restart, &need_seq_rewrite);
+        &is_restart, &need_seq_rewrite, isIdentity);
 #else
     init_params<T_Form, T_Int, large>(stmt->options, false, isUseLocalSeq, newm, &owned_by,
-        &need_seq_rewrite);
+        &need_seq_rewrite, isIdentity);
 #endif
 #ifdef PGXC /* PGXC_COORD */
     /*
@@ -2119,6 +2123,7 @@ enum {
     DEF_IDX_MIN_VALUE,
     DEF_IDX_CACHE_VALUE,
     DEF_IDX_IS_CYCLED,
+    DEF_IDX_TYPE_ID,
     DEF_IDX_NUM
 };
 
@@ -2168,12 +2173,45 @@ static void PreProcessSequenceOptions(
         } else if (strcmp(defel->defname, "owned_by") == 0) {
             CheckDuplicateDef(*owned_by);
             *owned_by = defGetQualifiedName(defel);
+        } else if (strcmp(defel->defname, "as") == 0) {
+            CheckDuplicateDef(elms[DEF_IDX_TYPE_ID]);
+            elms[DEF_IDX_TYPE_ID] = defel;
+            *need_seq_rewrite = true;
         } else {
             ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("option \"%s\" not recognized", defel->defname)));
         }
     }
 }
 
+template<typename T_Form, bool large>
+static void ProcessSequenceOptAsType(DefElem* elm, T_Form newm, bool isInit,
+    bool isIdentity, Oid &type_id, int &typeMods)
+{
+    if (isIdentity) {
+        if (elm != NULL) {
+            Oid newtypid = typenameTypeId(NULL, defGetTypeName(elm));
+            type_id = newtypid;
+
+            if (nodeTag(elm->arg) == T_TypeName) {
+                TypeName* typname = (TypeName*)elm->arg;
+                typeMods = typname->typemod;
+            }
+
+            if (typeMods > DMODE_MAX_PRECISION) {
+                ereport(ERROR,
+                        (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+                         errmsg("Specified column precision %d is greater than the maximum precision of %d.",
+                             typeMods, DMODE_MAX_PRECISION)));
+            }
+        }
+    } else {
+        if (!large)
+            type_id = INT8OID;
+        else
+            type_id = INT16OID;
+    }
+}
+
 template<typename T_Form, typename T_Int, bool large>
 static void ProcessSequenceOptIncrementBy(DefElem* elm, T_Form newm, bool isInit)
 {
@@ -2301,6 +2339,76 @@ static void CrossCheckMinMax(T_Int min, T_Int max)
     }
 }
 
+template<typename T_Int, bool large>
+static bool CheckSeedIncrementValueMinMax(T_Int value, T_Int min_value, T_Int max_value)
+{
+    if (value < min_value || value > max_value) {
+        return false;
+    }
+    return true;
+}
+
+template<typename T_Form, typename T_Int, bool large>
+static void ProcessSequenceOptMaxMin(DefElem* elm, T_Form newm, bool isInit, Oid type_id, int typeMods)
+{
+    if (elm != NULL && elm->arg) {
+        int maxValue = defGetInt<T_Int, large>(elm);
+
+        if (maxValue == SEQ_MAXVALUE_8) {
+            AssignInt<T_Int, large>(&(newm->max_value), SEQ_MAXVALUE_8);
+            AssignInt<T_Int, large>(&(newm->min_value), SEQ_MINVALUE_8);
+        } else if (maxValue == SEQ_MAXVALUE_16) {
+            AssignInt<T_Int, large>(&(newm->max_value), SEQ_MAXVALUE_16);
+            AssignInt<T_Int, large>(&(newm->min_value), SEQ_MINVALUE_16);
+        } else if (maxValue == SEQ_MAXVALUE_32) {
+            AssignInt<T_Int, large>(&(newm->max_value), SEQ_MAXVALUE_32);
+            AssignInt<T_Int, large>(&(newm->min_value), SEQ_MINVALUE_32);
+        } else if (maxValue == SEQ_MAXVALUE) {
+            AssignInt<T_Int, large>(&(newm->max_value), SEQ_MAXVALUE);
+            AssignInt<T_Int, large>(&(newm->min_value), SEQ_MINVALUE);
+        } else {
+            int128 minValue = maxValue * (int128)-1 - (int128)1;
+            AssignInt<T_Int, large>(&(newm->max_value), maxValue);
+            AssignInt<T_Int, large>(&(newm->min_value), minValue);
+        }
+        newm->log_cnt = 0;
+    } else if (isInit || elm != NULL) {
+        /* ascending seq */
+        if (large) {
+            if (typeMods == -1)
+                typeMods = DMODE_DEFAULT_PRECISION;
+
+            int128 maxDecimalNum = 9;
+            int128 minDecimalNum = -1;
+            int maxSigDigit = 9;
+            int base = 10;
+            for (int i = 1; i < typeMods; i++) {
+                maxDecimalNum = maxDecimalNum * base + maxSigDigit;
+            }
+
+            minDecimalNum = maxDecimalNum * (int128)-1 - (int128)1;
+                
+            AssignInt<T_Int, large>(&(newm->max_value), maxDecimalNum);
+            AssignInt<T_Int, large>(&(newm->min_value), minDecimalNum);
+        } else {
+            if (type_id == INT1OID) {
+                AssignInt<T_Int, large>(&(newm->max_value), SEQ_MAXVALUE_8);
+                AssignInt<T_Int, large>(&(newm->min_value), SEQ_MINVALUE_8);
+            } else if (type_id == INT2OID) {
+                AssignInt<T_Int, large>(&(newm->max_value), SEQ_MAXVALUE_16);
+                AssignInt<T_Int, large>(&(newm->min_value), SEQ_MINVALUE_16);
+            } else if (type_id == INT4OID) {
+                AssignInt<T_Int, large>(&(newm->max_value), SEQ_MAXVALUE_32);
+                AssignInt<T_Int, large>(&(newm->min_value), SEQ_MINVALUE_32);
+            } else {
+                AssignInt<T_Int, large>(&(newm->max_value), SEQ_MAXVALUE);
+                AssignInt<T_Int, large>(&(newm->min_value), SEQ_MINVALUE);
+            }
+        }
+        newm->log_cnt = 0;
+    }
+}
+
 /*
  * init_params: process the options list of CREATE or ALTER SEQUENCE,
  * and store the values into appropriate fields of *new.  Also set
@@ -2320,15 +2428,16 @@ static void CrossCheckMinMax(T_Int min, T_Int max)
 #ifdef PGXC
 template<typename T_Form, typename T_Int, bool large>
 static void init_params(List* options, bool isInit, bool isUseLocalSeq, void* newm_p, List** owned_by,
-    bool* is_restart, bool* need_seq_rewrite)
+    bool* is_restart, bool* needSeqRewrite, bool isIdentity)
 #else
 template<typename T_Form, typename T_Int, bool large>
 static void init_params(List* options, bool isInit, bool isUseLocalSeq, void* newm_p, List** owned_by,
-    bool* need_seq_rewrite)
+    bool* needSeqRewrite, bool isIdentity)
 #endif
 {
     T_Form newm = (T_Form)newm_p;
     DefElem* elms[DEF_IDX_NUM] = {0};
+    Oid type_id = InvalidOid;
 
 #ifdef PGXC
     *is_restart = false;
@@ -2336,7 +2445,7 @@ static void init_params(List* options, bool isInit, bool isUseLocalSeq, void* ne
 
     *owned_by = NIL;
 
-    PreProcessSequenceOptions(options, elms, owned_by, need_seq_rewrite, isInit);
+    PreProcessSequenceOptions(options, elms, owned_by, needSeqRewrite, isInit);
 
     /*
      * We must reset log_cnt when isInit or when changing any parameters
@@ -2346,10 +2455,17 @@ static void init_params(List* options, bool isInit, bool isUseLocalSeq, void* ne
         newm->log_cnt = 0;
     }
 
+    int typeMods = -1;
+
+    ProcessSequenceOptAsType<T_Form, large>(elms[DEF_IDX_TYPE_ID], newm, isInit, isIdentity, type_id, typeMods);
     ProcessSequenceOptIncrementBy<T_Form, T_Int, large>(elms[DEF_IDX_INCREMENT_BY], newm, isInit);
     ProcessSequenceOptCycle<T_Form>(elms[DEF_IDX_IS_CYCLED], newm, isInit);
-    ProcessSequenceOptMax<T_Form, T_Int, large>(elms[DEF_IDX_MAX_VALUE], newm, isInit);
-    ProcessSequenceOptMin<T_Form, T_Int, large>(elms[DEF_IDX_MIN_VALUE], newm, isInit);
+    if (u_sess->attr.attr_sql.sql_compatibility == D_FORMAT && isIdentity) {
+        ProcessSequenceOptMaxMin<T_Form, T_Int, large>(elms[DEF_IDX_MAX_VALUE], newm, isInit, type_id, typeMods);
+    } else {
+        ProcessSequenceOptMax<T_Form, T_Int, large>(elms[DEF_IDX_MAX_VALUE], newm, isInit);
+        ProcessSequenceOptMin<T_Form, T_Int, large>(elms[DEF_IDX_MIN_VALUE], newm, isInit);
+    }
 
     /* crosscheck min/max */
     CrossCheckMinMax<T_Int, large>(newm->min_value, newm->max_value);
@@ -2357,7 +2473,20 @@ static void init_params(List* options, bool isInit, bool isUseLocalSeq, void* ne
     ProcessSequenceOptStartWith<T_Form, T_Int, large>(elms[DEF_IDX_START_VALUE], newm, isInit);
 
     /* crosscheck START */
-    CheckValueMinMax<T_Int, large>(newm->start_value, newm->min_value, newm->max_value, true);
+    if (u_sess->attr.attr_sql.sql_compatibility == D_FORMAT && isIdentity) {
+        if (!CheckSeedIncrementValueMinMax<T_Int, large>(newm->start_value, newm->min_value, newm->max_value)) {
+            ereport(ERROR,
+                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+                    errmsg("Identity column contains invalid SEED.")));
+        }
+        if (!CheckSeedIncrementValueMinMax<T_Int, large>(newm->increment_by, newm->min_value, newm->max_value)) {
+            ereport(ERROR,
+                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+                    errmsg("Identity column contains invalid INCREMENT.")));
+        }
+    } else {
+        CheckValueMinMax<T_Int, large>(newm->start_value, newm->min_value, newm->max_value, true);
+    }
 
     ProcessSequenceOptReStartWith<T_Form, T_Int, large>(
         elms[DEF_IDX_RESTART_VALUE], newm, isInit, is_restart, isUseLocalSeq);
@@ -3327,3 +3456,21 @@ int64 get_and_reset_last_value(text* txt, int64 new_value, bool need_reseed)
 
     return last_value;
 }
+
+bool StrEndWith(const char *str, const char *suffix)
+{
+    int str_len = strlen(str);
+    int suffix_len = strlen(suffix);
+
+    if (str_len < suffix_len) {
+        return false;
+    }
+    
+    for (int i = 0; i < suffix_len; i++) {
+        if (str[str_len - 1 - i] != suffix[suffix_len - 1 - i]) {
+            return false;
+        }
+    }
+
+    return true;
+}
diff --git a/src/gausskernel/optimizer/commands/tablecmds.cpp b/src/gausskernel/optimizer/commands/tablecmds.cpp
index dc3d75c1de943d1f891ec6950539bf122280d978..95870bfe8b7e4ccc380b9cfffca2094bfb97e1cd 100755
--- a/src/gausskernel/optimizer/commands/tablecmds.cpp
+++ b/src/gausskernel/optimizer/commands/tablecmds.cpp
@@ -13596,7 +13596,8 @@ static ObjectAddress ATExecAddColumn(List** wqueue, AlteredTableInfo* tab, Relat
                 if (colDef->is_not_null) {
                     ATExecAppendDefValExpr(attribute.attnum, defval, tab, colDef, true, is_addloc);
                 }
-            } else if (contain_specified_function((Node*)defval, NEXTVALFUNCOID)) {
+            } else if (contain_specified_function((Node*)defval, NEXTVALFUNCOID)
+                    && u_sess->attr.attr_sql.sql_compatibility != D_FORMAT) {
                 /* We don't support alter table add column which default with nextval expression. */
                 ereport(ERROR,
                     (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
diff --git a/src/include/commands/sequence.h b/src/include/commands/sequence.h
index 66d8655f76a2da96e56183995da1f57a56fde4ba..6682184ab034194510986b9981fc33f5ee41b50a 100644
--- a/src/include/commands/sequence.h
+++ b/src/include/commands/sequence.h
@@ -89,6 +89,9 @@ typedef FormData_pg_large_sequence* Form_pg_large_sequence;
 
 #define GS_NUM_OF_BUCKETS 1024
 
+#define DMODE_MAX_PRECISION 38
+#define DMODE_DEFAULT_PRECISION 18
+
 /*
  * The "special area" of a old version sequence's buffer page looks like this.
  */
@@ -262,5 +265,7 @@ extern SeqTable GetSessSeqElm(Oid relid);
 extern char* GetGlobalSeqNameForUpdate(Relation seqrel, char** dbname, char** schemaname);
 extern uint32 RelidGetHash(Oid seq_relid);
 extern SeqTable GetGlobalSeqElm(Oid relid, GlobalSeqInfoHashBucket* bucket);
+extern Oid pg_get_serial_sequence_oid(text* tablename, text* columnname);
+bool StrEndWith(const char *str, const char *suffix);
 
 #endif /* SEQUENCE_H */
diff --git a/src/include/nodes/parsenodes_common.h b/src/include/nodes/parsenodes_common.h
index f89e1d564c9245fd85e84e8d5fb228ef3660d851..38cf773c38b4c86c48e0f689775b170fb9eb6bf5 100644
--- a/src/include/nodes/parsenodes_common.h
+++ b/src/include/nodes/parsenodes_common.h
@@ -1445,7 +1445,8 @@ typedef enum ConstrType { /* types of constraints */
     CONSTR_ATTR_DEFERRED,
     CONSTR_ATTR_IMMEDIATE,
     CONSTR_GENERATED,
-    CONSTR_AUTO_INCREMENT
+    CONSTR_AUTO_INCREMENT,
+    CONSTR_IDENTITY
 } ConstrType;
 
 typedef struct Constraint {
diff --git a/src/include/pg_config_manual.h b/src/include/pg_config_manual.h
index 21dbff22815a6e59964ce4f93bd8c29c43bf447b..911bb2c46a866a36f4a506882de5d1e89b75b07c 100644
--- a/src/include/pg_config_manual.h
+++ b/src/include/pg_config_manual.h
@@ -57,6 +57,12 @@
  */
 #define SEQ_MAXVALUE INT64CONST(0x7FFFFFFFFFFFFFFF)
 #define SEQ_MINVALUE (-SEQ_MAXVALUE)
+#define SEQ_MAXVALUE_32 0x7FFFFFFF
+#define SEQ_MINVALUE_32 (-SEQ_MAXVALUE_32 - 1)
+#define SEQ_MAXVALUE_16 0x7FFF
+#define SEQ_MINVALUE_16 (-SEQ_MAXVALUE_16 - 1)
+#define SEQ_MAXVALUE_8 0xFF
+#define SEQ_MINVALUE_8 0x00
 
 #define LARGE_SEQ_MAXVALUE (int128)(((uint128)1 << 127) - 1)
 #define LARGE_SEQ_MINVALUE (-INT128_MAX - 1)