diff --git a/dingo-calcite/src/main/codegen/config.fmpp b/dingo-calcite/src/main/codegen/config.fmpp
index 49075929a2..6cedf9bf14 100644
--- a/dingo-calcite/src/main/codegen/config.fmpp
+++ b/dingo-calcite/src/main/codegen/config.fmpp
@@ -67,6 +67,7 @@ data: {
       "io.dingodb.calcite.grammar.ddl.SqlAnalyze"
       "io.dingodb.calcite.grammar.ddl.SqlCreateTenant"
       "io.dingodb.calcite.grammar.ddl.SqlCreateSchema"
+      "io.dingodb.calcite.grammar.ddl.DingoSqlCreateView"
       "io.dingodb.calcite.grammar.ddl.SqlCreateUser"
       "io.dingodb.calcite.grammar.ddl.SqlCreateIndex"
       "io.dingodb.calcite.grammar.ddl.SqlCreateVectorIndex"
@@ -161,7 +162,10 @@ data: {
   # List of new keywords. Example: "DATABASES", "TABLES". If the keyword is
   # not a reserved keyword, add it to the 'nonReservedKeywords' section.
   keywords: [
+      "ALGORITHM"
+      "INPLACE"
       "IF"
+      "COPY"
       "MATERIALIZED"
       "STORED"
       "VIRTUAL"
@@ -178,6 +182,7 @@ data: {
       "ENGINE"
       "ENGINES"
       "EXCHANGE"
+      "EXCLUSIVE"
       "EXECUTOR"
       "EXECUTORS"
       "ENFORCED"
@@ -208,6 +213,7 @@ data: {
       "SCAN"
       "SSL"
       "STARTING"
+      "SHARED"
       "STATUS"
       "TABLES"
       "STARTTS"
@@ -228,6 +234,7 @@ data: {
       "SAMPLES"
       "SAMPLERATE"
       "PESSIMISTIC"
+      "PARSER"
       "OPTIMISTIC"
       "BLOCKS"
       "HASH"
@@ -260,6 +267,13 @@ data: {
       "CACHE"
       "MODIFY"
       "CHANGE"
+      "SPATIAL"
+      "FULLTEXT"
+      "KEY_BLOCK_SIZE"
+      "BTREE"
+      "RTREE"
+      "UNDEFINED"
+      "TEMPTABLE"
     ]
 
     # List of non-reserved keywords to add;
diff --git a/dingo-calcite/src/main/codegen/includes/AlterTable.ftl b/dingo-calcite/src/main/codegen/includes/AlterTable.ftl
index 500429b12c..9e05666938 100644
--- a/dingo-calcite/src/main/codegen/includes/AlterTable.ftl
+++ b/dingo-calcite/src/main/codegen/includes/AlterTable.ftl
@@ -64,6 +64,16 @@ SqlAlterSchema SqlAlterSchema(Span s, String scope): {
    }
 }
 
+SqlAlterTable SqlAlterIgnoreTable(Span s, String scope): {
+    SqlAlterTable alterTable = null;
+} {
+  <IGNORE>
+  alterTable = SqlAlterTable(s, scope)
+  {
+    return alterTable;
+  }
+}
+
 SqlAlterTable SqlAlterTable(Span s, String scope): {
     SqlIdentifier id;
     SqlAlterTable alterTable = null;
@@ -81,6 +91,10 @@ SqlAlterTable SqlAlterTable(Span s, String scope): {
             alterTable = addColumn(s, scope, id)
         |
             alterTable = addConstraint(s, scope, id)
+        |
+            <FULLTEXT> alterTable = addIndexByMode(s, scope, id, "fulltext")
+        |
+           <SPATIAL> alterTable = addIndexByMode(s, scope, id, "spetail")
 	    )
 	  |
          <DROP>
@@ -166,6 +180,7 @@ SqlAlterTable addColumn(Span s, String scope, SqlIdentifier id): {
     boolean primary = false;
     String comment = "";
     boolean autoInc = false;
+    String collate = "utf8_bin";
 } {
     <COLUMN>
     columnId = SimpleIdentifier()
@@ -188,6 +203,8 @@ SqlAlterTable addColumn(Span s, String scope, SqlIdentifier id): {
        <PRIMARY> <KEY> { primary=true; }
     |
        <COMMENT> comment = dingoIdentifier()
+    |
+       <COLLATE> { collate = this.getNextToken().image;}
     |
        <AUTO_INCREMENT> { autoInc = true;}
     |
@@ -199,7 +216,7 @@ SqlAlterTable addColumn(Span s, String scope, SqlIdentifier id): {
                     : ColumnStrategy.NOT_NULLABLE;
         }
         return new SqlAlterAddColumn(s.end(this), id, DingoSqlDdlNodes.createColumn(
-            s.end(this), columnId, type.withNullable(nullable), e, strategy, autoInc, comment, primary
+            s.end(this), columnId, type.withNullable(nullable), e, strategy, autoInc, comment, primary, collate
         ));
     }
 }
@@ -233,17 +250,20 @@ SqlAlterTable addIndex(Span s, String scope, SqlIdentifier id): {
     int replica = 0;
     String indexType = "scalar";
     SqlNodeList withColumnList = null;
-    final SqlNodeList columnList;
+    List<SqlNode> columnList;
     String engine = null;
+    Properties prop = new Properties();
+    String indexAlg = null;
+    String indexLockOpt = null;
 } {
 <INDEX> { s.add(this); }
     { index = getNextToken().image; }
     (
-        <VECTOR> { indexType = "vector"; } columnList = ParenthesizedSimpleIdentifierList()
+        <VECTOR> { indexType = "vector"; } columnList = indexColumns()
     |
-        <TEXT> { indexType = "text"; } columnList = ParenthesizedSimpleIdentifierList()
+        <TEXT> { indexType = "text"; } columnList = indexColumns()
     |
-        [<SCALAR>] columnList = ParenthesizedSimpleIdentifierList()
+        [<SCALAR>] columnList = indexColumns()
     )
     (
        <WITH> withColumnList = ParenthesizedSimpleIdentifierList()
@@ -261,11 +281,14 @@ SqlAlterTable addIndex(Span s, String scope, SqlIdentifier id): {
      |
         <PARAMETERS> properties = readProperties()
     )*
+    prop = indexOption()
+    [ indexAlg = indexAlg()]
+    [ indexLockOpt = indexLockOpt()]
     {
         return new SqlAlterAddIndex(
             s.end(this), id,
             new SqlIndexDeclaration(
-                s.end(this), index, columnList, withColumnList, properties,partitionDefinition, replica, indexType, engine, false
+                s.end(this), index, columnList, withColumnList, properties,partitionDefinition, replica, indexType, engine, "", prop, indexAlg, indexLockOpt
             )
         );
     }
@@ -279,12 +302,15 @@ SqlAlterTable addUniqueIndex(Span s, String scope, SqlIdentifier id): {
     int replica = 0;
     String indexType = "scalar";
     SqlNodeList withColumnList = null;
-    final SqlNodeList columnList;
+    List<SqlNode> columnList;
     String engine = null;
+    Properties prop = null;
+    String indexAlg = null;
+    String indexLockOpt = null;
 } {
  <UNIQUE> [<INDEX>][<KEY>] { s.add(this); }
     { index = getNextToken().image; }
-    [<SCALAR>] columnList = ParenthesizedSimpleIdentifierList()
+    [<SCALAR>] columnList = indexColumns()
     (
        <WITH> withColumnList = ParenthesizedSimpleIdentifierList()
      |
@@ -301,16 +327,66 @@ SqlAlterTable addUniqueIndex(Span s, String scope, SqlIdentifier id): {
      |
         <PARAMETERS> properties = readProperties()
     )*
+    prop = indexOption()
+    [ indexAlg = indexAlg()]
+    [ indexLockOpt = indexLockOpt()]
     {
         return new SqlAlterAddIndex(
             s.end(this), id,
             new SqlIndexDeclaration(
-                s.end(this), index, columnList, withColumnList, properties,partitionDefinition, replica, indexType, engine, true
+                s.end(this), index, columnList, withColumnList, properties,partitionDefinition, replica, indexType, engine, "unique", prop, indexAlg, indexLockOpt
             )
         );
     }
 }
 
+SqlAlterTable addIndexByMode(Span s, String scope, SqlIdentifier id, String mode): {
+    final String index;
+    Boolean autoIncrement = false;
+    Properties properties = null;
+    PartitionDefinition partitionDefinition = null;
+    int replica = 0;
+    String indexType = "scalar";
+    SqlNodeList withColumnList = null;
+    List<SqlNode> columnList;
+    String engine = null;
+    Properties prop = null;
+    String indexAlg = null;
+    String indexLockOpt = null;
+} {
+   <INDEX> { s.add(this); }
+    { index = getNextToken().image; }
+    columnList = indexColumns()
+    (
+       <WITH> withColumnList = ParenthesizedSimpleIdentifierList()
+     |
+       <ENGINE> <EQ> engine = dingoIdentifier() { if (engine.equalsIgnoreCase("innodb")) { engine = "TXN_LSM";} }
+     |
+        <PARTITION> <BY>
+            {
+                partitionDefinition = new PartitionDefinition();
+                partitionDefinition.setFuncName(getNextToken().image);
+                partitionDefinition.setDetails(readPartitionDetails());
+            }
+     |
+        <REPLICA> <EQ> {replica = Integer.parseInt(getNextToken().image);}
+     |
+        <PARAMETERS> properties = readProperties()
+    )*
+    prop = indexOption()
+    [ indexAlg = indexAlg()]
+    [ indexLockOpt = indexLockOpt()]
+    {
+        return new SqlAlterAddIndex(
+            s.end(this), id,
+            new SqlIndexDeclaration(
+                s.end(this), index, columnList, withColumnList, properties,partitionDefinition, replica, indexType, engine, mode, prop, indexAlg, indexLockOpt
+            )
+        );
+    }
+}
+
+
 SqlAlterTable alterIndex(Span s, String scope, SqlIdentifier id): {
     final String index;
     Properties properties = new Properties();
@@ -501,6 +577,7 @@ SqlAlterTable modifyColumn(Span s, String scope, SqlIdentifier id, SqlAlterTable
     PartitionDefinition partitionDefinition = null;
     int replica = 0;
     String engine = null;
+    String collate = "utf8_bin";
     String indexType = "scalar";
     Boolean primaryKey = false;
     String comment = "";
@@ -508,6 +585,7 @@ SqlAlterTable modifyColumn(Span s, String scope, SqlIdentifier id, SqlAlterTable
     SqlIdentifier refTable = null;
     String updateRefOpt = null;
     String deleteRefOpt = null;
+    SqlIdentifier afterCol = null;
 } {
    <COLUMN> name = SimpleIdentifier()
     (
@@ -540,6 +618,8 @@ SqlAlterTable modifyColumn(Span s, String scope, SqlIdentifier id, SqlAlterTable
            <COMMENT> (<IDENTIFIER>|<QUOTED_STRING>) { comment = token.image; }
          |
           <ON> <UPDATE> <CURRENT_TIMESTAMP>
+         |
+          <COLLATE> { collate = this.getNextToken().image; }
          |
           <CONSTRAINT> { s.add(this); } [name = SimpleIdentifier()] <CHECK> <LPAREN>
              checkExpr = Expression(ExprContext.ACCEPT_SUB_QUERY)
@@ -553,20 +633,22 @@ SqlAlterTable modifyColumn(Span s, String scope, SqlIdentifier id, SqlAlterTable
           )
           )*
         )*
+        [ <AFTER> afterCol = SimpleIdentifier() ]
         {
             if (e == null) {
                 strategy = nullable ? ColumnStrategy.NULLABLE
                     : ColumnStrategy.NOT_NULLABLE;
             }
-            columnDec = DingoSqlDdlNodes.createColumn(s.add(id).end(this), name, type.withNullable(nullable), e, strategy, autoIncrement, comment, primaryKey);
+            columnDec = DingoSqlDdlNodes.createColumn(s.add(id).end(this), name, type.withNullable(nullable), e, strategy, autoIncrement, comment, primaryKey, collate);
         }
     )
    {
      if (alterTable == null) {
-         return new SqlAlterModifyColumn(s.end(this), id, columnDec);
+         return new SqlAlterModifyColumn(s.end(this), id, columnDec, afterCol);
      } else {
          SqlAlterModifyColumn alterModifyCol = (SqlAlterModifyColumn)alterTable;
          alterModifyCol.addSqlColumn(columnDec);
+         alterModifyCol.addAlterColumn(afterCol);
          return alterModifyCol;
      }
    }
@@ -617,6 +699,7 @@ SqlAlterTable changeColumn(Span s, String scope, SqlIdentifier id): {
     SqlIdentifier refTable = null;
     String updateRefOpt = null;
     String deleteRefOpt = null;
+    String collate = "utf8_bin";
 } {
   <COLUMN> name = SimpleIdentifier() newName = SimpleIdentifier()
    (
@@ -649,6 +732,8 @@ SqlAlterTable changeColumn(Span s, String scope, SqlIdentifier id): {
            <COMMENT> (<IDENTIFIER>|<QUOTED_STRING>) { comment = token.image; }
          |
           <ON> <UPDATE> <CURRENT_TIMESTAMP>
+         |
+          <COLLATE> { collate = this.getNextToken().image;}
          |
           <CONSTRAINT> { s.add(this); } [name = SimpleIdentifier()] <CHECK> <LPAREN>
              checkExpr = Expression(ExprContext.ACCEPT_SUB_QUERY)
@@ -667,7 +752,7 @@ SqlAlterTable changeColumn(Span s, String scope, SqlIdentifier id): {
                 strategy = nullable ? ColumnStrategy.NULLABLE
                     : ColumnStrategy.NOT_NULLABLE;
             }
-            columnDec = DingoSqlDdlNodes.createColumn(s.add(id).end(this), name, type.withNullable(nullable), e, strategy, autoIncrement, comment, primaryKey);
+            columnDec = DingoSqlDdlNodes.createColumn(s.add(id).end(this), name, type.withNullable(nullable), e, strategy, autoIncrement, comment, primaryKey, collate);
         }
     )?
   {
diff --git a/dingo-calcite/src/main/codegen/includes/parserImpls.ftl b/dingo-calcite/src/main/codegen/includes/parserImpls.ftl
index 36d178b48a..44ce7e3f64 100644
--- a/dingo-calcite/src/main/codegen/includes/parserImpls.ftl
+++ b/dingo-calcite/src/main/codegen/includes/parserImpls.ftl
@@ -155,6 +155,10 @@ void TableElement(List<SqlNode> list) :
     SqlIdentifier refTable = null;
     String updateRefOpt = null;
     String deleteRefOpt = null;
+    String collate = "utf8_bin";
+    Properties prop = null;
+    String indexAlg = null;
+    String indexLockOpt = null;
 }
 {
     LOOKAHEAD(2) id = SimpleIdentifier()
@@ -187,7 +191,8 @@ void TableElement(List<SqlNode> list) :
          |
            <COMMENT> (<IDENTIFIER>|<QUOTED_STRING>) { comment = token.image; }
          |
-
+           <COLLATE> { collate = this.getNextToken().image; }
+         |
           <ON> <UPDATE> <CURRENT_TIMESTAMP>
          |
           <CONSTRAINT> { s.add(this); } [name = SimpleIdentifier()] <CHECK> <LPAREN>
@@ -199,7 +204,7 @@ void TableElement(List<SqlNode> list) :
                 strategy = nullable ? ColumnStrategy.NULLABLE
                     : ColumnStrategy.NOT_NULLABLE;
             }
-            columnDec = DingoSqlDdlNodes.createColumn(s.add(id).end(this), id, type.withNullable(nullable), e, strategy, autoIncrement, comment, primaryKey);
+            columnDec = DingoSqlDdlNodes.createColumn(s.add(id).end(this), id, type.withNullable(nullable), e, strategy, autoIncrement, comment, primaryKey, collate);
             list.add(columnDec);
         }
     )
@@ -240,6 +245,9 @@ void TableElement(List<SqlNode> list) :
          |
            <PARAMETERS> properties = readProperties()
         )*
+        prop = indexOption()
+        [ indexAlg = indexAlg()]
+        [ indexLockOpt = indexLockOpt()]
         {
             list.add(new SqlIndexDeclaration(s.end(this), index, columnList, withColumnList, properties,
             partitionDefinition, replica, indexType, engine, false));
@@ -267,6 +275,9 @@ void TableElement(List<SqlNode> list) :
         [
             <REPLICA> <EQ> {replica = Integer.parseInt(getNextToken().image);}
         ]
+        prop = indexOption()
+        [ indexAlg = indexAlg()]
+        [ indexLockOpt = indexLockOpt()]
     |
         <PRIMARY>  { s.add(this); } <KEY>
         columnList = ParenthesizedSimpleIdentifierList() {
@@ -602,18 +613,28 @@ Number number(): {
       )
 }
 
+
 SqlCreate SqlCreateView(Span s, boolean replace) :
 {
     final SqlIdentifier id;
     SqlNodeList columnList = null;
     final SqlNode query;
+    String user = null;
+    String host = null;
+    String alg = null;
+    String security = null;
+    String checkOpt = null;
 }
 {
+    [<ALGORITHM> <EQ> (<UNDEFINED>|<MERGE>|<TEMPTABLE>)]
+    [ <DEFINER> <EQ> user = strIdent() [ <AT_SPLIT> strIdent() { host = token.image; }  ]]
+    [ <SQL> <SECURITY> (<DEFINER>|<INVOKER>)]
     <VIEW> id = CompoundIdentifier()
     [ columnList = ParenthesizedSimpleIdentifierList() ]
-    <AS> query = OrderedQueryOrExpr(ExprContext.ACCEPT_QUERY) {
-        return SqlDdlNodes.createView(s.end(this), replace, id, columnList,
-            query);
+    <AS> query = OrderedQueryOrExpr(ExprContext.ACCEPT_QUERY)
+    [ <WITH> (<CASCADED> {checkOpt = "cascaded";}|<LOCAL> {checkOpt = "local";}) <CHECK> <OPTION>]
+    {
+        return new DingoSqlCreateView(s.end(this), replace, id, columnList, query, security, alg, user, host, checkOpt);
     }
 }
 
@@ -684,34 +705,152 @@ SqlCreate SqlCreateFunction(Span s, boolean replace) :
 SqlCreate SqlCreateIndex(Span s, boolean replace) :
 {
     boolean isUnique = false;
+    String mode = "";
     final String index;
     SqlIdentifier table;
-    SqlIdentifier column;
-    List<SqlIdentifier> columns;
+    SqlNode column;
+    List<SqlNode> columns;
     int replica = 0;
     SqlNode create = null;
     Boolean ifNotExists = false;
+    Properties prop = null;
+    String indexAlg = null;
+    String indexLockOpt = null;
 }
 {
-    [ <UNIQUE> { isUnique = true;}]
+    [ (<UNIQUE> { isUnique = true; mode="unique"; } | <FULLTEXT> { mode = "fulltext";}|<SPATIAL> {mode = "fulltext";})]
     <INDEX> ifNotExists = IfNotExistsOpt()
     ( <QUOTED_STRING> | <IDENTIFIER> )
      { index = token.image.toUpperCase(); }
     <ON> table = CompoundIdentifier()
-    <LPAREN> column = SimpleIdentifier() { columns = new ArrayList<SqlIdentifier>(); columns.add(column); }
+    columns = indexColumns()
+    [
+        <REPLICA> <EQ> {replica = Integer.parseInt(getNextToken().image);}
+    ]
+    prop = indexOption()
+    [ indexAlg = indexAlg()]
+    [ indexLockOpt = indexLockOpt()]
+    {
+       return new SqlCreateIndex(s.end(this), replace, ifNotExists, index, table, columns, isUnique, replica, mode, prop, indexAlg, indexLockOpt);
+    }
+}
+
+String indexAlg(): {
+  String alg = null;
+} {
+  <ALGORITHM> <EQ>
+  (
+   <DEFAULT_> {alg = "default";}
+   |
+   <INPLACE> { alg = "inplace";}
+   |
+   <COPY> { alg = "copy";}
+  )
+  {
+    return alg;
+  }
+}
+
+String indexLockOpt(): {
+  String lockOpt = null;
+} {
+  <LOCK> [<EQ>]
+  (
+    <DEFAULT_> { lockOpt = "default";}
+    |
+    <NONE> { lockOpt = "none";}
+    |
+    <SHARED> { lockOpt = "shared";}
+    |
+    <EXCLUSIVE> { lockOpt = "exclusive"; }
+  )
+  {
+    return lockOpt;
+  }
+}
+
+Properties indexOption(): {
+  Properties prop = new Properties();
+} {
+   (
+    <KEY_BLOCK_SIZE> [<EQ>] <UNSIGNED_INTEGER_LITERAL> { prop.put("key_block_size", Integer.parseInt(this.token.image));}
+    |
+    <RTREE> { prop.put("indexType", "bdb"); }
+    |
+    <BTREE> { prop.put("indexType", "bdb"); }
+    |
+    <HASH> { prop.put("indexType", "hash"); }
+    |
+    <WITH> <PARSER> { prop.put("parser", strIdent()); }
+    |
+    <COMMENT> { prop.put("comment", strIdent());}
+    |
+    <VISIBLE> { prop.put("visible", "true"); }
+    |
+    <INVISIBLE> { prop.put("visible", "false"); }
+    |
+    <GLOBAL> { prop.put("isGlobal", "true"); }
+    |
+    <LOCAL> { prop.put("isGlobal", "false"); }
+   )*
+   {
+     return prop;
+   }
+}
+
+String strIdent(): {
+  String str = null;
+} {
+   (
+    <IDENTIFIER> { str = token.image; }
+   |
+    <QUOTED_STRING> { str = token.image.replace("'", "");}
+   )
+   {
+     return str;
+   }
+}
+
+List<SqlNode> indexColumns(): {
+   List<SqlNode> columns = new ArrayList<>();
+   SqlNode column = null;
+} {
+  <LPAREN> column = keyPart() { columns.add(column); }
     (
        <COMMA>
-       column = SimpleIdentifier() { columns.add(column); }
+       column = keyPart() { columns.add(column); }
     )*
     <RPAREN>
-    [
-        <REPLICA> <EQ> {replica = Integer.parseInt(getNextToken().image);}
-    ]
     {
-       return new SqlCreateIndex(s.end(this), replace, ifNotExists, index, table, columns, isUnique, replica);
+      return columns;
     }
 }
 
+SqlNode keyPart():{
+ SqlIdentifier column = null;
+ SqlNode expr = null;
+ int precision = -1;
+} {
+  (
+  column = SimpleIdentifier()
+  [ <LPAREN>
+         precision = UnsignedIntLiteral()
+    <RPAREN>
+  ]
+  [<ASC>] [<DESC>]
+  {
+     return column;
+  }
+  |
+  <LPAREN>
+      expr = Expression(ExprContext.ACCEPT_SUB_QUERY)
+  <RPAREN>[<ASC>] [<DESC>]
+  {
+     return expr;
+  }
+  )
+}
+
 SqlCreate SqlCreateVectorIndex(Span s, boolean replace) :
 {
     final String index;
diff --git a/dingo-calcite/src/main/codegen/templates/Parser.jj b/dingo-calcite/src/main/codegen/templates/Parser.jj
index 00f99550a2..efb9ae13cc 100644
--- a/dingo-calcite/src/main/codegen/templates/Parser.jj
+++ b/dingo-calcite/src/main/codegen/templates/Parser.jj
@@ -6206,6 +6206,9 @@ SqlTypeNameSpec CharacterTypeName(Span s) :
         charSetName = Identifier()
     ]
     {
+        if (charSetName != null) {
+            charSetName = charSetName.toUpperCase();
+        }
         return new SqlBasicTypeNameSpec(sqlTypeName, precision, charSetName, s.end(this));
     }
 }
diff --git a/dingo-calcite/src/main/java/io/dingodb/calcite/DingoDdlExecutor.java b/dingo-calcite/src/main/java/io/dingodb/calcite/DingoDdlExecutor.java
index 63405bb553..325cb9dc2e 100644
--- a/dingo-calcite/src/main/java/io/dingodb/calcite/DingoDdlExecutor.java
+++ b/dingo-calcite/src/main/java/io/dingodb/calcite/DingoDdlExecutor.java
@@ -20,6 +20,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableList;
 import io.dingodb.calcite.grammar.ddl.DingoSqlCreateTable;
+import io.dingodb.calcite.grammar.ddl.DingoSqlCreateView;
 import io.dingodb.calcite.grammar.ddl.SqlAdminResetAutoInc;
 import io.dingodb.calcite.grammar.ddl.SqlAlterAddColumn;
 import io.dingodb.calcite.grammar.ddl.SqlAlterAddConstraint;
@@ -156,7 +157,6 @@
 import org.apache.calcite.sql.ddl.DingoSqlColumn;
 import org.apache.calcite.sql.ddl.DingoSqlKeyConstraint;
 import org.apache.calcite.sql.ddl.SqlCreateTable;
-import org.apache.calcite.sql.ddl.SqlCreateView;
 import org.apache.calcite.sql.ddl.SqlDropSchema;
 import org.apache.calcite.sql.ddl.SqlDropTable;
 import org.apache.calcite.sql.ddl.SqlDropView;
@@ -194,6 +194,7 @@
 import static io.dingodb.common.mysql.error.ErrorCode.ErrNotFoundDropSchema;
 import static io.dingodb.common.mysql.error.ErrorCode.ErrNotFoundDropTable;
 import static io.dingodb.common.mysql.error.ErrorCode.ErrPartitionMgmtOnNonpartitioned;
+import static io.dingodb.common.mysql.error.ErrorCode.ErrUnsupportedDDLOperation;
 import static io.dingodb.common.mysql.error.ErrorCode.ErrUnsupportedModifyVec;
 import static io.dingodb.common.util.Optional.mapOrNull;
 import static io.dingodb.common.util.PrivilegeUtils.getRealAddress;
@@ -752,7 +753,7 @@ public void execute(@NonNull SqlRevoke sqlRevoke, CalcitePrepare.Context context
         }
     }
 
-    public void execute(SqlCreateView sqlCreateView, CalcitePrepare.Context context) {
+    public void execute(DingoSqlCreateView sqlCreateView, CalcitePrepare.Context context) {
         LogUtils.info(log, "DDL execute: {}", sqlCreateView);
         String connId = (String) context.getDataContext().get("connId");
         SubSnapshotSchema schema = getSnapShotSchema(sqlCreateView.name, context, false);
@@ -827,13 +828,19 @@ public void execute(SqlCreateView sqlCreateView, CalcitePrepare.Context context)
             .replica(3)
             .createSql(sql)
             .comment("VIEW")
-            .charset(null)
-            .collate(null)
+            .charset("utf8mb4")
+            .collate("utf8mb4_bin")
             .tableType("VIEW")
             .rowFormat(null)
             .createTime(System.currentTimeMillis())
             .updateTime(0)
             .build();
+        Properties properties = new Properties();
+        properties.setProperty("user", sqlCreateView.definer);
+        properties.setProperty("host", sqlCreateView.host);
+        properties.setProperty("check_option", sqlCreateView.checkOpt);
+        properties.setProperty("security_type", sqlCreateView.security);
+        tableDefinition.setProperties(properties);
         DdlService ddlService = DdlService.root();
         ddlService.createViewWithInfo(schemaName, tableDefinition, connId, null);
     }
@@ -1008,7 +1015,7 @@ public void execute(@NonNull SqlAlterTableAddPart sqlAlterTableAddPart, CalciteP
         boolean duplicatePart = table.getPartitions().stream()
             .anyMatch(partition -> detail.getPartName().equalsIgnoreCase(partition.getName()));
         if (duplicatePart) {
-            
+            throw DingoErrUtil.newStdErr("duplicate part");
         }
 
         DdlService.root().alterTableAddPart(
@@ -1019,6 +1026,11 @@ public void execute(@NonNull SqlAlterTableAddPart sqlAlterTableAddPart, CalciteP
     }
 
     public void execute(@NonNull SqlAlterAddIndex sqlAlterAddIndex, CalcitePrepare.Context context) {
+        if ("fulltext".equalsIgnoreCase(sqlAlterAddIndex.getIndexDeclaration().mode)
+            || "spatial".equalsIgnoreCase(sqlAlterAddIndex.getIndexDeclaration().mode)) {
+            throw DingoErrUtil.newStdErr(sqlAlterAddIndex.getIndexDeclaration().mode + " index is not supported",
+                ErrUnsupportedDDLOperation);
+        }
         final Pair<SubSnapshotSchema, String> schemaTableName
             = getSchemaAndTableName(sqlAlterAddIndex.table, context);
         final String tableName = Parameters.nonNull(schemaTableName.right, "table name");
@@ -1096,6 +1108,10 @@ public void execute(@NonNull SqlAlterIndexVisible sqlAlterIndexVisible, CalciteP
     }
 
     public void execute(@NonNull SqlCreateIndex sqlCreateIndex, CalcitePrepare.Context context) {
+        if ("fulltext".equalsIgnoreCase(sqlCreateIndex.mode) || "spatial".equalsIgnoreCase(sqlCreateIndex.mode)) {
+            throw DingoErrUtil.newStdErr(sqlCreateIndex.mode + " index is not supported",
+                ErrUnsupportedDDLOperation);
+        }
         final Pair<SubSnapshotSchema, String> schemaTableName
             = getSchemaAndTableName(sqlCreateIndex.table, context);
         final String tableName = Parameters.nonNull(schemaTableName.right, "table name");
@@ -1113,14 +1129,20 @@ public void execute(@NonNull SqlCreateIndex sqlCreateIndex, CalcitePrepare.Conte
             null,
             sqlCreateIndex.replica,
             "scalar",
-            "TXN_LSM"
+            "TXN_LSM",
+            sqlCreateIndex.mode,
+            sqlCreateIndex.properties,
+            sqlCreateIndex.indexAlg,
+            sqlCreateIndex.indexLockOpt
         );
         sqlIndexDeclaration.unique = sqlCreateIndex.isUnique;
         if (isNotTxnEngine(table.getTable().getEngine())) {
-            throw new IllegalArgumentException("Table with index, the engine must be transactional.");
+            throw DingoErrUtil.newStdErr("Table with index, the engine must be transactional.");
         }
         IndexDefinition indexDef = fromSqlIndexDeclaration(sqlIndexDeclaration, fromTable(table.getTable()));
         indexDef.setUnique(sqlCreateIndex.isUnique);
+        indexDef.setVisible(sqlIndexDeclaration.isVisible());
+        indexDef.setComment(sqlIndexDeclaration.getComment());
         validateIndex(schema, tableName, indexDef);
         DdlService.root().createIndex(schema.getSchemaName(), tableName, indexDef);
 
@@ -1441,7 +1463,8 @@ public void execute(SqlAlterModifyColumn sqlAlterModifyColumn, CalcitePrepare.Co
         if (sqlAlterModifyColumn.dingoSqlColumnList.size() > 1) {
             throw DingoErrUtil.newStdErr("not support");
         }
-        for (DingoSqlColumn dingoSqlColumn : sqlAlterModifyColumn.dingoSqlColumnList) {
+        for (int j = 0; j < sqlAlterModifyColumn.dingoSqlColumnList.size(); j ++) {
+            DingoSqlColumn dingoSqlColumn = sqlAlterModifyColumn.dingoSqlColumnList.get(j);
             String name = dingoSqlColumn.name.getSimple();
             boolean indexMatch = table.getIndexes().stream().anyMatch(indexTable -> {
                 if (indexTable.getName().startsWith(DdlUtil.ddlTmpTableName)) {
@@ -1468,6 +1491,20 @@ public void execute(SqlAlterModifyColumn sqlAlterModifyColumn, CalcitePrepare.Co
             if (column == null) {
                 throw DINGO_RESOURCE.unknownColumn(name, tableName).ex();
             }
+            SqlIdentifier afterCol = sqlAlterModifyColumn.afterColumnList.get(j);
+            String afterColName;
+            if (afterCol != null) {
+                afterColName = afterCol.getSimple();
+                Column afterColumn = table.getColumns().stream()
+                    .filter(col -> col.getSchemaState() == SchemaState.SCHEMA_PUBLIC
+                        && col.getName().equalsIgnoreCase(afterColName)).findFirst().orElse(null);
+
+                if (afterColumn == null) {
+                    throw DINGO_RESOURCE.unknownColumn(name, tableName).ex();
+                }
+            } else {
+                afterColName = null;
+            }
             ColumnDefinition oldColumn = mapTo(column);
 
             ColumnDefinition newColumn = fromSqlColumnDeclaration(
@@ -1487,6 +1524,7 @@ public void execute(SqlAlterModifyColumn sqlAlterModifyColumn, CalcitePrepare.Co
                 .newCol(newColumn)
                 .changingCol(oldColumn)
                 .oldColName(oldColumn.getName())
+                .afterColName(afterColName)
                 .build();
             modifyingColInfoList.add(modifyingColInfo);
         }
@@ -2368,7 +2406,8 @@ private static IndexDefinition fromSqlIndexDeclaration(
             indexTableDefinition,
             indexTableDefinition.getPartDefinition()
         );
-        indexTableDefinition.setVisible(true);
+        indexTableDefinition.setVisible(indexDeclaration.isVisible());
+        indexTableDefinition.setComment(indexDeclaration.getComment());
 
         return indexTableDefinition;
     }
diff --git a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/DingoSqlCreateView.java b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/DingoSqlCreateView.java
new file mode 100644
index 0000000000..05964df6a8
--- /dev/null
+++ b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/DingoSqlCreateView.java
@@ -0,0 +1,126 @@
+/*
+ * Copyright 2021 DataCanvas
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.dingodb.calcite.grammar.ddl;
+
+
+import org.apache.calcite.sql.SqlCreate;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlSpecialOperator;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.util.ImmutableNullableList;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+
+public class DingoSqlCreateView extends SqlCreate {
+    public final SqlIdentifier name;
+    public final @Nullable SqlNodeList columnList;
+    public final SqlNode query;
+    private static final SqlOperator OPERATOR;
+    public String security;
+    public String alg;
+    public String definer;
+    public String host;
+    public String checkOpt;
+
+    public DingoSqlCreateView(
+        SqlParserPos pos,
+        boolean replace,
+        SqlIdentifier name,
+        @Nullable SqlNodeList columnList,
+        SqlNode query,
+        String security,
+        String alg,
+        String definer,
+        String host,
+        String checkOpt
+    ) {
+        super(OPERATOR, pos, replace, false);
+        this.name = (SqlIdentifier) Objects.requireNonNull(name, "name");
+        this.columnList = columnList;
+        this.query = (SqlNode)Objects.requireNonNull(query, "query");
+        this.security = security;
+        this.alg = alg;
+        this.definer = definer;
+        this.host = host;
+        this.checkOpt = checkOpt;
+    }
+
+    public List<SqlNode> getOperandList() {
+        return ImmutableNullableList.of(this.name, this.columnList, this.query);
+    }
+
+    public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
+        if (this.getReplace()) {
+            writer.keyword("CREATE OR REPLACE");
+        } else {
+            writer.keyword("CREATE");
+        }
+        if (alg != null) {
+            writer.keyword("ALGORITHM =");
+            writer.keyword(alg);
+        }
+        if (definer != null) {
+            writer.keyword("DEFINER = ");
+            writer.keyword(definer);
+            if (host != null) {
+                writer.keyword("@");
+                writer.keyword(host);
+            }
+        }
+        if (security != null) {
+            writer.keyword("SQL SECURITY");
+            writer.keyword(security);
+        }
+
+
+        writer.keyword("VIEW");
+        this.name.unparse(writer, leftPrec, rightPrec);
+        if (this.columnList != null) {
+            SqlWriter.Frame frame = writer.startList("(", ")");
+            Iterator var5 = this.columnList.iterator();
+
+            while(var5.hasNext()) {
+                SqlNode c = (SqlNode)var5.next();
+                writer.sep(",");
+                c.unparse(writer, 0, 0);
+            }
+
+            writer.endList(frame);
+        }
+
+        writer.keyword("AS");
+        writer.newlineAndIndent();
+        this.query.unparse(writer, 0, 0);
+        if (checkOpt != null) {
+            writer.keyword("WITH");
+            writer.keyword(checkOpt);
+            writer.keyword("CHECK OPTION");
+        }
+    }
+
+    static {
+        OPERATOR = new SqlSpecialOperator("CREATE VIEW", SqlKind.CREATE_VIEW);
+    }
+}
diff --git a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/DingoSqlDdlNodes.java b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/DingoSqlDdlNodes.java
index 9ac8c6dd3b..4a33a0b1c3 100644
--- a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/DingoSqlDdlNodes.java
+++ b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/DingoSqlDdlNodes.java
@@ -79,9 +79,10 @@ public static DingoSqlColumn createColumn(
         ColumnStrategy strategy,
         boolean autoIncrement,
         String comment,
-        boolean primary
+        boolean primary,
+        String collate
     ) {
-        return new DingoSqlColumn(pos, name, dataType, expression, strategy, autoIncrement, comment, primary);
+        return new DingoSqlColumn(pos, name, dataType, expression, strategy, autoIncrement, comment, primary, collate);
     }
 
 }
diff --git a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterChangeColumn.java b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterChangeColumn.java
index 60fd388601..ff02b10d6b 100644
--- a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterChangeColumn.java
+++ b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterChangeColumn.java
@@ -20,6 +20,7 @@
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.SqlSpecialOperator;
+import org.apache.calcite.sql.SqlWriter;
 import org.apache.calcite.sql.ddl.DingoSqlColumn;
 import org.apache.calcite.sql.parser.SqlParserPos;
 
@@ -46,4 +47,12 @@ public SqlAlterChangeColumn(
         this.dingoSqlColumn = sqlColumn;
     }
 
+    @Override
+    public void unparseAlterOperation(SqlWriter writer, int leftPrec, int rightPrec) {
+        writer.keyword("ALTER TABLE");
+        this.table.unparse(writer, leftPrec, rightPrec);
+        writer.keyword("CHANGE COLUMN");
+        dingoSqlColumn.unparse(writer, leftPrec, rightPrec);
+    }
+
 }
diff --git a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterIndexVisible.java b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterIndexVisible.java
index dcf4997a9f..21dd01f578 100644
--- a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterIndexVisible.java
+++ b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterIndexVisible.java
@@ -21,6 +21,7 @@
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.SqlSpecialOperator;
+import org.apache.calcite.sql.SqlWriter;
 import org.apache.calcite.sql.parser.SqlParserPos;
 
 public class SqlAlterIndexVisible extends SqlAlterTable {
@@ -36,4 +37,17 @@ public SqlAlterIndexVisible(SqlParserPos pos, SqlIdentifier sqlIdentifier, Strin
         this.index = indexName;
         this.invisible = invisible;
     }
+
+    @Override
+    public void unparseAlterOperation(SqlWriter writer, int leftPrec, int rightPrec) {
+        writer.keyword("ALTER TABLE");
+        this.table.unparse(writer, leftPrec, rightPrec);
+        writer.keyword("ALTER INDEX");
+        writer.keyword(index);
+        if (invisible) {
+            writer.keyword("INVISIBLE");
+        } else {
+            writer.keyword("VISIBLE");
+        }
+    }
 }
diff --git a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterModifyColumn.java b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterModifyColumn.java
index e6435893a6..137f63063d 100644
--- a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterModifyColumn.java
+++ b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterModifyColumn.java
@@ -20,6 +20,7 @@
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.SqlSpecialOperator;
+import org.apache.calcite.sql.SqlWriter;
 import org.apache.calcite.sql.ddl.DingoSqlColumn;
 import org.apache.calcite.sql.parser.SqlParserPos;
 
@@ -29,16 +30,31 @@
 public class SqlAlterModifyColumn extends SqlAlterTable {
 
     public List<DingoSqlColumn> dingoSqlColumnList = new ArrayList<>();
+    public List<SqlIdentifier> afterColumnList = new ArrayList<>();
 
     private static final SqlOperator OPERATOR =
         new SqlSpecialOperator("ALTER TABLE MODIFY COLUMN", SqlKind.ALTER_TABLE);
 
-    public SqlAlterModifyColumn(SqlParserPos pos, SqlIdentifier sqlIdentifier, DingoSqlColumn sqlColumn) {
+    public SqlAlterModifyColumn(
+        SqlParserPos pos, SqlIdentifier sqlIdentifier, DingoSqlColumn sqlColumn, SqlIdentifier afterCol) {
         super(pos, sqlIdentifier, OPERATOR);
         this.dingoSqlColumnList.add(sqlColumn);
+        this.afterColumnList.add(afterCol);
     }
 
     public void addSqlColumn(DingoSqlColumn sqlColumn) {
         this.dingoSqlColumnList.add(sqlColumn);
     }
+
+    public void addAlterColumn(SqlIdentifier alterCol) {
+        this.afterColumnList.add(alterCol);
+    }
+
+    @Override
+    public void unparseAlterOperation(SqlWriter writer, int leftPrec, int rightPrec) {
+        writer.keyword("ALTER TABLE");
+        this.table.unparse(writer, leftPrec, rightPrec);
+        writer.keyword("MODIFY COLUMN");
+        this.dingoSqlColumnList.get(0).unparse(writer, leftPrec, rightPrec);
+    }
 }
diff --git a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterTableAddPart.java b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterTableAddPart.java
index 5f6f6fd5f0..29f1e5aa91 100644
--- a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterTableAddPart.java
+++ b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterTableAddPart.java
@@ -21,6 +21,7 @@
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.SqlSpecialOperator;
+import org.apache.calcite.sql.SqlWriter;
 import org.apache.calcite.sql.parser.SqlParserPos;
 
 public class SqlAlterTableAddPart extends SqlAlterTable {
@@ -35,4 +36,28 @@ public SqlAlterTableAddPart(SqlParserPos pos, SqlIdentifier sqlIdentifier, Parti
         this.part = part;
     }
 
+    @Override
+    public void unparseAlterOperation(SqlWriter writer, int leftPrec, int rightPrec) {
+        writer.keyword("ALTER TABLE");
+        this.table.unparse(writer, leftPrec, rightPrec);
+        writer.keyword("ADD PARTITION");
+        if (part.getPartName() != null) {
+            writer.keyword(part.getPartName());
+        }
+        writer.keyword("VALUES LESS THAN(");
+        int i = 0;
+        int opSize = part.getOperand().length;
+        for (Object operand : part.getOperand()) {
+            if (operand == null) {
+                writer.keyword("null");
+            } else {
+                writer.keyword(operand.toString());
+            }
+            if (i < opSize - 1) {
+                writer.keyword(",");
+            }
+            i ++;
+        }
+        writer.keyword(")");
+    }
 }
diff --git a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterTruncatePart.java b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterTruncatePart.java
index 0f4a02b875..cd8c25ab52 100644
--- a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterTruncatePart.java
+++ b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlAlterTruncatePart.java
@@ -20,6 +20,7 @@
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.SqlSpecialOperator;
+import org.apache.calcite.sql.SqlWriter;
 import org.apache.calcite.sql.parser.SqlParserPos;
 
 public class SqlAlterTruncatePart extends SqlAlterTable {
@@ -33,4 +34,12 @@ public SqlAlterTruncatePart(SqlParserPos pos, SqlIdentifier sqlIdentifier, SqlId
         super(pos, sqlIdentifier, OPERATOR);
         this.part = part;
     }
+
+    @Override
+    public void unparseAlterOperation(SqlWriter writer, int leftPrec, int rightPrec) {
+        writer.keyword("ALTER TABLE");
+        this.table.unparse(writer, leftPrec, rightPrec);
+        writer.keyword("TRUNCATE PART");
+        part.unparse(writer, leftPrec, rightPrec);
+    }
 }
diff --git a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlCreateIndex.java b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlCreateIndex.java
index f442c6e16a..7351ba666f 100644
--- a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlCreateIndex.java
+++ b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlCreateIndex.java
@@ -26,6 +26,7 @@
 import org.apache.calcite.sql.parser.SqlParserPos;
 
 import java.util.List;
+import java.util.Properties;
 import java.util.stream.Collectors;
 
 public class SqlCreateIndex extends SqlCreate {
@@ -34,28 +35,43 @@ public class SqlCreateIndex extends SqlCreate {
 
     public SqlIdentifier table;
 
-    public List<String> columns;
+    public List<SqlNode> columns;
 
     public boolean isUnique;
 
+    public String mode;
+
     public int replica;
 
+    public Properties properties;
+
+    public String indexAlg;
+
+    public String indexLockOpt;
+
     private static final SqlOperator OPERATOR =
         new SqlSpecialOperator("CREATE INDEX", SqlKind.OTHER_DDL);
 
     public SqlCreateIndex(SqlParserPos pos, boolean replace, boolean ifNotExists,
                           String index,
                           SqlIdentifier table,
-                          List<SqlIdentifier> columns,
-                          boolean isUnique, int replica) {
+                          List<SqlNode> columns,
+                          boolean isUnique,
+                          int replica,
+                          String mode,
+                          Properties prop,
+                          String indexAlg,
+                          String indexLockOpt) {
         super(OPERATOR, pos, replace, ifNotExists);
         this.index = index;
         this.table = table;
-        this.columns = columns.stream()
-            .map(SqlIdentifier::getSimple)
-            .map(String::toUpperCase).collect(Collectors.toList());
+        this.columns = columns;
         this.isUnique = isUnique;
         this.replica = replica;
+        this.mode = mode;
+        this.properties = prop;
+        this.indexAlg = indexAlg;
+        this.indexLockOpt = indexLockOpt;
     }
 
     @Override
@@ -65,8 +81,24 @@ public List<SqlNode> getOperandList() {
 
     @Override
     public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
-        writer.keyword("create index");
+        writer.keyword("create");
+        if (mode != null) {
+            writer.keyword(mode);
+        }
+        writer.keyword("index");
         writer.keyword(index);
+        writer.keyword("on");
+        table.unparse(writer, leftPrec, rightPrec);
+        writer.keyword("(");
+        int i = 0;
+        for (SqlNode col : columns) {
+            col.unparse(writer, leftPrec, rightPrec);
+            if (i < columns.size() - 1) {
+                writer.keyword(",");
+            }
+            i ++;
+        }
+        writer.keyword(")");
     }
 
 }
diff --git a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlIndexDeclaration.java b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlIndexDeclaration.java
index c90c742c81..ad502f24be 100644
--- a/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlIndexDeclaration.java
+++ b/dingo-calcite/src/main/java/io/dingodb/calcite/grammar/ddl/SqlIndexDeclaration.java
@@ -38,6 +38,7 @@ public class SqlIndexDeclaration extends SqlCall {
 
     public String index;
     public boolean unique;
+    public String mode;
 
     public List<String> columnList;
 
@@ -58,9 +59,60 @@ public class SqlIndexDeclaration extends SqlCall {
     @Getter
     String engine;
 
+    Properties indexOpt;
+    String algorithm;
+    String lockOpt;
+
     private static final SqlSpecialOperator OPERATOR =
         new SqlSpecialOperator("INDEX_DECL", SqlKind.CREATE_INDEX);
 
+    public SqlIndexDeclaration(
+        SqlParserPos pos,
+        String index,
+        List<SqlNode> columnList,
+        SqlNodeList withColumnList,
+        Properties properties,
+        PartitionDefinition partDefinition,
+        int replica,
+        String indexType,
+        String engine,
+        String mode,
+        Properties option,
+        String algorithm,
+        String lockOpt
+    ) {
+        super(pos);
+        this.index = index;
+        if (columnList != null) {
+            this.columnList = columnList.stream()
+                .filter(Objects::nonNull)
+                .map(SqlIdentifier.class::cast)
+                .map(SqlIdentifier::getSimple)
+                .map(String::toUpperCase)
+                .collect(Collectors.toCollection(ArrayList::new));
+        }
+        if (withColumnList != null) {
+            this.withColumnList = withColumnList.getList().stream()
+                .filter(Objects::nonNull)
+                .map(SqlIdentifier.class::cast)
+                .map(SqlIdentifier::getSimple)
+                .map(String::toUpperCase)
+                .collect(Collectors.toCollection(ArrayList::new));
+        }
+        this.properties = properties;
+        this.partDefinition = partDefinition;
+        this.replica = replica;
+        this.indexType = indexType;
+        this.engine = engine;
+        if ("unique".equalsIgnoreCase(mode)) {
+            this.unique = true;
+        }
+        this.mode = mode;
+        this.algorithm = algorithm;
+        this.lockOpt = lockOpt;
+        this.indexOpt = option;
+    }
+
     public SqlIndexDeclaration(
         SqlParserPos pos,
         String index,
@@ -136,4 +188,19 @@ public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
         writer.keyword("index");
         writer.keyword(index);
     }
+
+    public boolean isVisible() {
+        if (indexOpt != null) {
+            String visibleStr = indexOpt.getOrDefault("visible", "true").toString();
+            return Boolean.parseBoolean(visibleStr);
+        }
+        return true;
+    }
+
+    public String getComment() {
+        if (indexOpt != null) {
+            return indexOpt.getOrDefault("comment", "").toString();
+        }
+        return "";
+    }
 }
diff --git a/dingo-calcite/src/main/java/org/apache/calcite/sql/ddl/DingoSqlColumn.java b/dingo-calcite/src/main/java/org/apache/calcite/sql/ddl/DingoSqlColumn.java
index 730f911421..8d74f452be 100644
--- a/dingo-calcite/src/main/java/org/apache/calcite/sql/ddl/DingoSqlColumn.java
+++ b/dingo-calcite/src/main/java/org/apache/calcite/sql/ddl/DingoSqlColumn.java
@@ -37,6 +37,9 @@ public class DingoSqlColumn extends SqlColumnDeclaration {
     @Setter
     String comment = "";
 
+    @Getter
+    String collate = "utf8_bin";
+
     public DingoSqlColumn(
         SqlParserPos pos,
         SqlIdentifier name,
@@ -45,12 +48,14 @@ public DingoSqlColumn(
         ColumnStrategy strategy,
         boolean autoIncrement,
         String comment,
-        boolean primary
+        boolean primary,
+        String collate
     ) {
         super(pos, name, dataType, expression, strategy);
         this.autoIncrement = autoIncrement;
         this.comment = comment;
         this.primaryKey = primary;
+        this.collate = collate;
     }
 
     @Override
diff --git a/dingo-calcite/src/test/java/io/dingodb/calcite/TestCreateTable.java b/dingo-calcite/src/test/java/io/dingodb/calcite/TestCreateTable.java
index 40d79b2f3f..cdc845de22 100644
--- a/dingo-calcite/src/test/java/io/dingodb/calcite/TestCreateTable.java
+++ b/dingo-calcite/src/test/java/io/dingodb/calcite/TestCreateTable.java
@@ -16,6 +16,7 @@
 
 package io.dingodb.calcite;
 
+import io.dingodb.calcite.grammar.ddl.DingoSqlCreateView;
 import io.dingodb.calcite.grammar.ddl.SqlAlterAddConstraint;
 import io.dingodb.calcite.grammar.ddl.SqlAlterAddForeign;
 import io.dingodb.calcite.grammar.ddl.SqlAlterAddIndex;
@@ -25,6 +26,7 @@
 import io.dingodb.calcite.grammar.ddl.SqlAlterDropConstraint;
 import io.dingodb.calcite.grammar.ddl.SqlAlterDropForeign;
 import io.dingodb.calcite.grammar.ddl.SqlAlterModifyColumn;
+import io.dingodb.calcite.grammar.ddl.SqlCreateIndex;
 import io.dingodb.calcite.grammar.ddl.SqlGrant;
 import io.dingodb.calcite.grammar.ddl.SqlSetPassword;
 import org.apache.calcite.sql.SqlIdentifier;
@@ -32,6 +34,7 @@
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlNodeList;
 import org.apache.calcite.sql.ddl.SqlCreateTable;
+import org.apache.calcite.sql.ddl.SqlCreateView;
 import org.apache.calcite.sql.ddl.SqlKeyConstraint;
 import org.apache.calcite.sql.parser.SqlParser;
 import org.apache.calcite.sql.parser.dingo.DingoSqlParserImpl;
@@ -41,6 +44,8 @@
 import java.util.Objects;
 import java.util.stream.Collectors;
 
+import static io.dingodb.calcite.DingoParser.PARSER_CONFIG;
+
 public class TestCreateTable {
 
     public static void parseCreateTable(SqlCreateTable create) {
@@ -347,12 +352,52 @@ public void sqlAlterColumnSetDefault() {
 
     @Test
     public void sqlAlterChangeColumn() {
-        String sql = "ALTER TABLE table_name change column a b int not null";
-        SqlParser.Config config = SqlParser.config().withParserFactory(DingoSqlParserImpl::new);
-        SqlParser parser = SqlParser.create(sql, config);
+        String sql = "ALTER TABLE `config_info` MODIFY COLUMN `src_ip` varchar(50) CHARACTER SET utf8  DEFAULT NULL COMMENT 'source ip'";
+        SqlParser parser = SqlParser.create(sql, PARSER_CONFIG);
+        try {
+            SqlNode sqlNode = parser.parseStmt();
+            assert sqlNode instanceof SqlAlterModifyColumn;
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Test
+    public void sqlCreateIndex() {
+        String sql = "create fulltext index ix on tx(col1(10) asc, col2(20)) btree comment 'commitsss' "
+            + "algorithm=inplace "
+            + "lock =none";
+        SqlParser parser = SqlParser.create(sql, PARSER_CONFIG);
+        try {
+            SqlNode sqlNode = parser.parseStmt();
+            assert sqlNode instanceof SqlCreateIndex;
+            SqlCreateIndex sqlCreateIndex = (SqlCreateIndex) sqlNode;
+            assert sqlCreateIndex.mode.contentEquals("fulltext");
+            assert sqlCreateIndex.properties.get("comment").equals("commitsss");
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Test
+    public void sqlAlterAddIndex() {
+        String sql = "alter table t1 add fulltext index ix(age) btree comment 'commitsss' algorithm=inplace lock=none";
+        SqlParser parser = SqlParser.create(sql, PARSER_CONFIG);
+        try {
+            SqlNode sqlNode = parser.parseStmt();
+            assert sqlNode instanceof SqlAlterAddIndex;
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Test
+    public void sqlCreateView() {
+        String sql = "create algorithm=merge definer=dingo sql security invoker view v1 as select * from t1";
+        SqlParser parser = SqlParser.create(sql, PARSER_CONFIG);
         try {
             SqlNode sqlNode = parser.parseStmt();
-            assert sqlNode instanceof SqlAlterChangeColumn;
+            assert sqlNode instanceof DingoSqlCreateView;
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
diff --git a/dingo-common/src/main/java/io/dingodb/common/ddl/ModifyingColInfo.java b/dingo-common/src/main/java/io/dingodb/common/ddl/ModifyingColInfo.java
index 9a07e1736f..0dfd9edf45 100644
--- a/dingo-common/src/main/java/io/dingodb/common/ddl/ModifyingColInfo.java
+++ b/dingo-common/src/main/java/io/dingodb/common/ddl/ModifyingColInfo.java
@@ -31,6 +31,7 @@ public class ModifyingColInfo {
     ColumnDefinition changingCol;
     String oldColName;
     int modifyTp;
+    String afterColName;
 
     public ModifyingColInfo() {
 
@@ -40,11 +41,13 @@ public ModifyingColInfo(
         ColumnDefinition newCol,
         ColumnDefinition changingCol,
         String oldColName,
-        int modifyTp
+        int modifyTp,
+        String afterColName
     ) {
         this.newCol = newCol;
         this.changingCol = changingCol;
         this.oldColName = oldColName;
         this.modifyTp = modifyTp;
+        this.afterColName = afterColName;
     }
 }
diff --git a/dingo-exec/src/main/java/io/dingodb/exec/operator/InfoSchemaScanOperator.java b/dingo-exec/src/main/java/io/dingodb/exec/operator/InfoSchemaScanOperator.java
index e1d70c018e..64a2f35b54 100644
--- a/dingo-exec/src/main/java/io/dingodb/exec/operator/InfoSchemaScanOperator.java
+++ b/dingo-exec/src/main/java/io/dingodb/exec/operator/InfoSchemaScanOperator.java
@@ -80,6 +80,7 @@ private InfoSchemaScanOperator() {
             case "SCHEMA_PRIVILEGES":
             case "TABLE_PRIVILEGES":
             case "VIEWS":
+                return getView();
             case "COLUMN_PRIVILEGES":
             case "COLLATIONS":
                 return getEmpty();
@@ -388,6 +389,47 @@ private static Iterator<Object[]> getMdlView() {
         return TransactionService.getDefault().getMdlInfo();
     }
 
+    public static Iterator<Object[]> getView() {
+        InfoSchema is = DdlService.root().getIsLatest();
+        return is.getSchemaMap().values()
+            .stream()
+            .filter(schemaTables ->
+                !schemaTables.getSchemaInfo().getName().equalsIgnoreCase("INFORMATION_SCHEMA"))
+            .flatMap(e -> {
+                Collection<Table> tables = e.getTables().values();
+                return tables.stream()
+                    .filter(td -> td.getTableType().equalsIgnoreCase("VIEW"))
+                    .map(td -> {
+                        String checkOpt = td.getProperties().getProperty("check_option", "");
+                        String isUpdaTable = "NO";
+                        String user = td.getProperties().getProperty("user", "");
+                        String host = td.getProperties().getProperty("host", "");
+                        String definer = user + "@" + host;
+                        String security = td.getProperties().getProperty("security_type");
+                        String character = "utf8";
+                        String collate = "utf8mb4_bin";
+                        try {
+                            return new Object[]{"def",
+                                e.getSchemaInfo().getName(),
+                                td.getName(),
+                                td.createSql,
+                                checkOpt,
+                                isUpdaTable,
+                                definer,
+                                security,
+                                character,
+                                collate
+                            };
+                        } catch (Exception e1) {
+                            LogUtils.error(log, e1.getMessage(), e1);
+                            return null;
+                        }
+                    }).filter(Objects::nonNull)
+                    .collect(Collectors.toList()).stream();
+            })
+            .iterator();
+    }
+
     /**
     * Api to get remote txn informations.
     */
diff --git a/dingo-executor/src/main/java/io/dingodb/server/executor/ddl/DdlWorker.java b/dingo-executor/src/main/java/io/dingodb/server/executor/ddl/DdlWorker.java
index 8bb09b664a..4901c497d5 100644
--- a/dingo-executor/src/main/java/io/dingodb/server/executor/ddl/DdlWorker.java
+++ b/dingo-executor/src/main/java/io/dingodb/server/executor/ddl/DdlWorker.java
@@ -2025,16 +2025,26 @@ public Pair<Long, String> onTruncatePart(DdlContext dc, DdlJob job) {
                 job.setSchemaState(SchemaState.SCHEMA_DELETE_REORG);
                 return updateSchemaVersion(dc, job);
             case SCHEMA_DELETE_REORG:
-                partitionList.remove(matchPart);
                 CommonId tableId = Mapper.MAPPER.idFrom(tableWithId.getTableId());
                 NavigableMap<ByteArrayUtils.ComparableByteArray, RangeDistribution> ranges =
                     MetaService.root().getRangeDistribution(Mapper.MAPPER.idFrom(tableWithId.getTableId()));
                 List<RangeDistribution> regionList = ranges.values().stream()
                     .filter(region -> region.getId().domain == matchPart.getId().getEntityId())
                     .collect(Collectors.toList());
-                LogUtils.info(log, "delete region size:{}, partId:{}", regionList.size(), matchPart.getId());
                 MetaService.root().deleteRegion(tableId, job.getId(), job.getRealStartTs(), false, regionList);
+                long newPartId = MetaService.root().generatePartId();
+                matchPart.getId().setEntityId(newPartId);
+                byte[] originStartKey = matchPart.getRange().getStartKey();
+                byte[] originEndKey = matchPart.getRange().getEndKey();
+                matchPart.getRange().setStartKey(Mapper.MAPPER.realKey(originStartKey, matchPart.getId(), (byte)'t'));
+                if (originEndKey.length == 9) {
+                    matchPart.getRange().setEndKey(Mapper.MAPPER.nextKey(matchPart.getId(), (byte)'t'));
+                } else {
+                    matchPart.getRange().setEndKey(Mapper.MAPPER.realKey(originEndKey, matchPart.getId(), (byte)'t'));
+                }
 
+                MetaService.root().rebaseRegion(tableWithId, matchPart);
+                // disable index
                 indexList.forEach(obj -> {
                     TableDefinitionWithId indexWithId = (TableDefinitionWithId) obj;
                     MetaService.root().dropIndex(
diff --git a/dingo-meta-api/src/main/java/io/dingodb/meta/MetaService.java b/dingo-meta-api/src/main/java/io/dingodb/meta/MetaService.java
index a7a5dc7a05..105e76e4c1 100644
--- a/dingo-meta-api/src/main/java/io/dingodb/meta/MetaService.java
+++ b/dingo-meta-api/src/main/java/io/dingodb/meta/MetaService.java
@@ -299,6 +299,17 @@ default void deleteRegion(
 
     }
 
+    default void rebaseRegion(
+        Object tableWithId,
+        Object partition
+    ) {
+
+    }
+
+    default long generatePartId() {
+        return 0;
+    }
+
     default void createTenant(Tenant tenant) {
 
     }
diff --git a/dingo-store-proxy/src/main/java/io/dingodb/store/proxy/meta/MetaService.java b/dingo-store-proxy/src/main/java/io/dingodb/store/proxy/meta/MetaService.java
index b15b868323..23d7af178c 100644
--- a/dingo-store-proxy/src/main/java/io/dingodb/store/proxy/meta/MetaService.java
+++ b/dingo-store-proxy/src/main/java/io/dingodb/store/proxy/meta/MetaService.java
@@ -1067,7 +1067,6 @@ public void dropRegionByTable(
         long ts,
         boolean autoInc
     ) {
-        CoordinatorService coordinatorService = Services.coordinatorService(Configuration.coordinatorSet());
         Collection<RangeDistribution> rangeDistributions = getRangeDistribution(tableId)
             .values();
         LogUtils.info(log, "dropRegion size:{}, tableId:{}", rangeDistributions.size(), tableId);
@@ -1101,6 +1100,43 @@ public void deleteRegion(
         }
     }
 
+    public void rebaseRegion(
+        Object tableWithId,
+        Object partition
+    ) {
+        TableDefinitionWithId tableDefinitionWithId = (TableDefinitionWithId) tableWithId;
+        io.dingodb.sdk.service.entity.meta.TableDefinition withIdTableDefinition
+            = tableDefinitionWithId.getTableDefinition();
+        long schemaId = tableDefinitionWithId.getTableId().getParentEntityId();
+        Partition partition1 = (Partition) partition;
+        CreateRegionRequest request = CreateRegionRequest
+            .builder()
+            .regionName("T_" + schemaId + "_" + withIdTableDefinition.getName()
+                + "_part_" + partition1.getId().getEntityId())
+            .regionType(RegionType.STORE_REGION)
+            .replicaNum(withIdTableDefinition.getReplica())
+            .range(partition1.getRange())
+            .rawEngine(getRawEngine(withIdTableDefinition.getEngine()))
+            .storeEngine(withIdTableDefinition.getStoreEngine())
+            .schemaId(schemaId)
+            .tableId(tableDefinitionWithId.getTableId().getEntityId())
+            .partId(partition1.getId().getEntityId())
+            .tenantId(tableDefinitionWithId.getTenantId())
+            .build();
+        CoordinatorService coordinatorService = Services.coordinatorService(Configuration.coordinatorSet());
+        coordinatorService.createRegion(tso(), request);
+    }
+
+    public long generatePartId() {
+        CoordinatorService coordinatorService = Services.coordinatorService(Configuration.coordinatorSet());
+        return coordinatorService.createIds(tso(), CreateIdsRequest.builder()
+                .idEpochType(IdEpochType.ID_NEXT_TABLE)
+                .count(1)
+                .build()
+            )
+            .getIds().get(0);
+    }
+
     private static void resetTableId(TableIdWithPartIds newTableId, TableDefinitionWithId table) {
         table.setTableId(newTableId.getTableId());
         List<Partition> partitions = table.getTableDefinition().getTablePartition().getPartitions();
@@ -1145,9 +1181,7 @@ public boolean dropTable(long tenantId, long schemaId, String tableName, long jo
                 return true;
             }
             List<CommonId> indexIds = indexes.stream().map(Table::getTableId).collect(Collectors.toList());
-            indexIds.forEach(indexId -> {
-                infoSchemaService.dropIndex(indexId.domain, indexId.seq);
-            });
+            indexIds.forEach(indexId -> infoSchemaService.dropIndex(indexId.domain, indexId.seq));
         }
         return true;
     }