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 b1ff37116e..aa973c2619 100644
--- a/dingo-calcite/src/main/java/io/dingodb/calcite/DingoDdlExecutor.java
+++ b/dingo-calcite/src/main/java/io/dingodb/calcite/DingoDdlExecutor.java
@@ -17,6 +17,7 @@
 package io.dingodb.calcite;
 
 import com.codahale.metrics.Timer;
+import com.google.common.collect.ImmutableList;
 import io.dingodb.calcite.grammar.ddl.DingoSqlCreateTable;
 import io.dingodb.calcite.grammar.ddl.SqlAlterAddColumn;
 import io.dingodb.calcite.grammar.ddl.SqlAlterAddIndex;
@@ -46,6 +47,7 @@
 import io.dingodb.calcite.schema.SubCalciteSchema;
 import io.dingodb.calcite.schema.SubSnapshotSchema;
 import io.dingodb.calcite.utils.IndexParameterUtils;
+import io.dingodb.calcite.type.DingoSqlTypeFactory;
 import io.dingodb.common.ddl.ActionType;
 import io.dingodb.common.ddl.SchemaDiff;
 import io.dingodb.common.log.LogUtils;
@@ -95,6 +97,8 @@
 import io.dingodb.verify.service.UserService;
 import io.dingodb.verify.service.UserServiceProvider;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.calcite.config.CalciteConnectionConfigImpl;
+import org.apache.calcite.config.CalciteConnectionProperty;
 import org.apache.calcite.jdbc.CalcitePrepare;
 import org.apache.calcite.jdbc.CalciteSchema;
 import org.apache.calcite.jdbc.ContextSqlValidator;
@@ -103,20 +107,26 @@
 import org.apache.calcite.schema.Schema;
 import org.apache.calcite.server.DdlExecutorImpl;
 import org.apache.calcite.sql.SqlBasicTypeNameSpec;
+import org.apache.calcite.sql.SqlCall;
 import org.apache.calcite.sql.SqlDataTypeSpec;
 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.SqlSelect;
 import org.apache.calcite.sql.SqlTypeNameSpec;
 import org.apache.calcite.sql.SqlUtil;
 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;
 import org.apache.calcite.sql.ddl.SqlKeyConstraint;
 import org.apache.calcite.sql.dialect.AnsiSqlDialect;
+import org.apache.calcite.sql.dialect.CalciteSqlDialect;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.sql.validate.SqlValidator;
@@ -138,6 +148,7 @@
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
+import static io.dingodb.calcite.DingoParser.PARSER_CONFIG;
 import static io.dingodb.calcite.runtime.DingoResource.DINGO_RESOURCE;
 import static io.dingodb.common.util.Optional.mapOrNull;
 import static io.dingodb.common.util.PrivilegeUtils.getRealAddress;
@@ -454,7 +465,8 @@ public void execute(SqlCreateTable createT, CalcitePrepare.Context context) {
 
         tableDefinition.setIndices(indexTableDefinitions);
         DdlService ddlService = DdlService.root();
-        ddlService.createTableWithInfo(schema.getSchemaName(), tableName, tableDefinition, connId, create.getOriginalCreateSql());
+        ddlService.createTableWithInfo(schema.getSchemaName(), tableName, tableDefinition,
+            connId, create.getOriginalCreateSql());
 
         RootCalciteSchema rootCalciteSchema = (RootCalciteSchema) context.getMutableRootSchema();
         RootSnapshotSchema rootSnapshotSchema = (RootSnapshotSchema) rootCalciteSchema.schema;
@@ -535,6 +547,48 @@ public void execute(SqlDropTable drop, CalcitePrepare.Context context) throws Ex
         }
     }
 
+    public void execute(SqlDropView sqlDropView, CalcitePrepare.Context context) throws Exception {
+        final Timer.Context timeCtx = DingoMetrics.getTimeContext("dropView");
+        long start = System.currentTimeMillis();
+        LogUtils.info(log, "DDL execute: {}", sqlDropView);
+        String connId = (String) context.getDataContext().get("connId");
+        final SubSnapshotSchema schema = getSnapShotSchema(sqlDropView.name, context, sqlDropView.ifExists);
+        if (schema == null) {
+            return;
+        }
+        final String tableName = getTableName(sqlDropView.name);
+        SchemaInfo schemaInfo = schema.getSchemaInfo(schema.getSchemaName());
+        if (schemaInfo == null) {
+            throw DINGO_RESOURCE.unknownSchema(schema.getSchemaName()).ex();
+        }
+        Table table = schema.getTableInfo(tableName);
+        if (table == null) {
+            if (sqlDropView.ifExists) {
+                return;
+            } else {
+                throw DINGO_RESOURCE.unknownTable(schema.getSchemaName() + "." + tableName).ex();
+            }
+        }
+        DdlService ddlService = DdlService.root();
+        ddlService.dropTable(schemaInfo, table.tableId.seq, tableName, connId);
+
+        RootCalciteSchema rootCalciteSchema = (RootCalciteSchema) context.getMutableRootSchema();
+        RootSnapshotSchema rootSnapshotSchema = (RootSnapshotSchema) rootCalciteSchema.schema;
+        SchemaDiff diff = SchemaDiff.builder().schemaId(schema.getSchemaId())
+            .type(ActionType.ActionDropTable).tableId(table.tableId.seq).build();
+        rootSnapshotSchema.applyDiff(diff);
+
+        timeCtx.stop();
+        long cost = System.currentTimeMillis() - start;
+        if (cost > 10000) {
+            LogUtils.info(log, "drop view take long time, cost:{}, schemaName:{}, tableName:{}",
+                cost, schemaInfo.getName(), tableName);
+        } else {
+            LogUtils.info(log, "[ddl] drop view success, cost:{}, schemaName:{}, tableName:{}",
+                cost, schema.getSchemaName(), tableName);
+        }
+    }
+
     public void execute(@NonNull SqlTruncate truncate, CalcitePrepare.Context context) throws Exception {
         long start = System.currentTimeMillis();
         final Timer.Context timeCtx = DingoMetrics.getTimeContext("truncateTable");
@@ -666,6 +720,92 @@ public void execute(@NonNull SqlRevoke sqlRevoke, CalcitePrepare.Context context
         }
     }
 
+    public void execute(SqlCreateView sqlCreateView, CalcitePrepare.Context context) {
+        LogUtils.info(log, "DDL execute: {}", sqlCreateView);
+        String connId = (String) context.getDataContext().get("connId");
+        SubSnapshotSchema schema = getSnapShotSchema(sqlCreateView.name, context, false);
+        if (schema == null) {
+            if (context.getDefaultSchemaPath() != null && !context.getDefaultSchemaPath().isEmpty()) {
+                throw DINGO_RESOURCE.unknownSchema(context.getDefaultSchemaPath().get(0)).ex();
+            } else {
+                throw DINGO_RESOURCE.unknownSchema("DINGO").ex();
+            }
+        }
+        String tableName = getTableName(sqlCreateView.name);
+
+        // Check table exist
+        if (schema.getTable(tableName) != null) {
+            throw DINGO_RESOURCE.tableExists(tableName).ex();
+        }
+        SqlNode query = renameColumns(sqlCreateView.columnList, sqlCreateView.query);
+
+        List<String> schemas = new ArrayList<>();
+        schemas.add(schema.getSchemaName());
+        List<List<String>> schemaPaths = new ArrayList<>();
+        schemaPaths.add(schemas);
+        schemaPaths.add(new ArrayList<>());
+
+        CalciteConnectionConfigImpl config;
+        config = new CalciteConnectionConfigImpl(new Properties());
+        config.set(CalciteConnectionProperty.CASE_SENSITIVE, String.valueOf(PARSER_CONFIG.caseSensitive()));
+        DingoCatalogReader catalogReader = new DingoCatalogReader(context.getRootSchema(),
+            schemaPaths, DingoSqlTypeFactory.INSTANCE, config);
+        DingoSqlValidator sqlValidator = new DingoSqlValidator(catalogReader, DingoSqlTypeFactory.INSTANCE);
+        SqlNode sqlNode = sqlValidator.validate(query);
+        RelDataType type = sqlValidator.getValidatedNodeType(sqlNode);
+
+        RelDataType jdbcType = type;
+        if (!type.isStruct()) {
+            jdbcType = sqlValidator.getTypeFactory().builder().add("$0", type).build();
+        }
+
+        List<ColumnDefinition> columnDefinitionList = jdbcType.getFieldList()
+            .stream().map(f -> {
+                int precision = f.getType().getPrecision();
+                int scale = f.getType().getScale();
+                String name = f.getType().getSqlTypeName().getName();
+                if ("BIGINT".equals(name) || "FLOAT".equals(name)) {
+                    precision = -1;
+                    scale = -2147483648;
+                }
+                boolean nullable = f.getType().isNullable();
+                if (sqlValidator.isHybridSearch() && "BIGINT".equals(name)) {
+                    nullable = true;
+                }
+                return ColumnDefinition
+                    .builder()
+                    .name(f.getName())
+                    .type(f.getType().getSqlTypeName().getName())
+                    .scale(scale)
+                    .precision(precision)
+                    .nullable(nullable)
+                    .build();
+            })
+            .collect(Collectors.toList());
+
+        String schemaName = schema.getSchemaName();
+
+        String sql = query.toSqlString(CalciteSqlDialect.DEFAULT).getSql();
+        // build tableDefinition
+        TableDefinition tableDefinition = TableDefinition.builder()
+            .name(tableName)
+            .columns(columnDefinitionList)
+            .version(1)
+            .ttl(0)
+            .replica(3)
+            .createSql(sql)
+            .comment("VIEW")
+            .charset(null)
+            .collate(null)
+            .tableType("VIEW")
+            .rowFormat(null)
+            .createTime(System.currentTimeMillis())
+            .updateTime(0)
+            .build();
+        DdlService ddlService = DdlService.root();
+        ddlService.createViewWithInfo(schemaName, tableName, tableDefinition, connId, null);
+    }
+
     public void execute(@NonNull SqlCreateUser sqlCreateUser, CalcitePrepare.Context context) {
         LogUtils.info(log, "DDL execute: {}", sqlCreateUser.toLog());
         UserDefinition userDefinition = UserDefinition.builder().user(sqlCreateUser.user)
@@ -1742,4 +1882,22 @@ private static SubSnapshotSchema getSnapShotSchema(
         return Pair.of(getSnapShotSchema(id, context, false), getTableName(id));
     }
 
+    static SqlNode renameColumns(@Nullable SqlNodeList columnList,
+                                 SqlNode query) {
+        if (columnList == null) {
+            return query;
+        }
+        final SqlParserPos p = query.getParserPosition();
+        final SqlNodeList selectList = SqlNodeList.SINGLETON_STAR;
+        final SqlCall from =
+            SqlStdOperatorTable.AS.createCall(p,
+                ImmutableList.<SqlNode>builder()
+                    .add(query)
+                    .add(new SqlIdentifier("_", p))
+                    .addAll(columnList)
+                    .build());
+        return new SqlSelect(p, null, selectList, from, null, null, null, null,
+            null, null, null, null);
+    }
+
 }
diff --git a/dingo-calcite/src/main/java/io/dingodb/calcite/DingoTable.java b/dingo-calcite/src/main/java/io/dingodb/calcite/DingoTable.java
index 6cd9ff2fa1..4c4dc164f8 100644
--- a/dingo-calcite/src/main/java/io/dingodb/calcite/DingoTable.java
+++ b/dingo-calcite/src/main/java/io/dingodb/calcite/DingoTable.java
@@ -20,31 +20,45 @@
 import io.dingodb.calcite.rel.LogicalDingoTableScan;
 import io.dingodb.calcite.schema.SubSnapshotSchema;
 import io.dingodb.calcite.type.converter.DefinitionMapper;
+import io.dingodb.calcite.utils.HybridNodeUtils;
 import io.dingodb.common.CommonId;
 import io.dingodb.common.log.LogUtils;
-import io.dingodb.meta.TableStatistic;
+import io.dingodb.common.table.HybridSearchTable;
 import io.dingodb.meta.entity.IndexTable;
 import io.dingodb.meta.entity.Table;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.plan.ViewExpanders;
 import org.apache.calcite.prepare.Prepare;
 import org.apache.calcite.rel.RelDistribution;
 import org.apache.calcite.rel.RelDistributions;
 import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.hint.HintPredicate;
+import org.apache.calcite.rel.hint.HintStrategyTable;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.runtime.CalciteContextException;
 import org.apache.calcite.schema.Statistic;
 import org.apache.calcite.schema.TranslatableTable;
 import org.apache.calcite.schema.impl.AbstractTable;
+import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.parser.SqlParseException;
+import org.apache.calcite.sql.parser.SqlParser;
+import org.apache.calcite.sql.parser.dingo.DingoSqlParserImpl;
 import org.apache.calcite.sql2rel.InitializerExpressionFactory;
+import org.apache.calcite.sql2rel.SqlToRelConverter;
 import org.apache.calcite.util.ImmutableBitSet;
 import org.checkerframework.checker.nullness.qual.NonNull;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 
 @Slf4j
@@ -65,7 +79,6 @@ public class DingoTable extends AbstractTable implements TranslatableTable {
     public DingoTable(
         @NonNull DingoParserContext context,
         @NonNull List<String> names,
-        @NonNull TableStatistic tableStatistic,
         @NonNull Table table
     ) {
         super();
@@ -102,20 +115,87 @@ public RelDataType getRowType(RelDataTypeFactory typeFactory) {
 
     @Override
     public RelNode toRel(RelOptTable.@NonNull ToRelContext context, RelOptTable relOptTable) {
-        return new LogicalDingoTableScan(
-            context.getCluster(),
-            context.getCluster().traitSet(),
-            context.getTableHints(),
-            relOptTable,
-            null,
-            null,
-            null,
-            null,
-            null,
-            ((DingoParserContext) context.getCluster().getPlanner().getContext()).isPushDown(),
-                // && Optional.mapOrGet(table.engine, __ -> !__.contains("TXN"), () -> true),
-            false
-        );
+        DingoTable dingoTable = relOptTable.unwrap(DingoTable.class);
+        if ((!dingoTable.getTable().getTableType().equalsIgnoreCase("VIEW"))
+            || dingoTable.getSchema().getSchemaName().equalsIgnoreCase("INFORMATION_SCHEMA")) {
+            return new LogicalDingoTableScan(
+                context.getCluster(),
+                context.getCluster().traitSet(),
+                context.getTableHints(),
+                relOptTable,
+                null,
+                null,
+                null,
+                null,
+                null,
+                ((DingoParserContext) context.getCluster().getPlanner().getContext()).isPushDown(),
+                false
+            );
+        } else {
+            SqlParser.Config config = SqlParser.config().withParserFactory(DingoSqlParserImpl::new);
+            SqlParser parser = SqlParser.create(dingoTable.getTable().createSql, config);
+            try {
+                SqlNode sqlNode = parser.parseStmt();
+                DingoSqlValidator dingoSqlValidator = this.context.getSqlValidator();
+                sqlNode = dingoSqlValidator.validate(sqlNode);
+
+                if (dingoSqlValidator.isHybridSearch()) {
+                    SqlNode originalSqlNode;
+                    parser = SqlParser.create(dingoTable.getTable().createSql, config);
+                    originalSqlNode = parser.parseQuery();
+                    //syntacticSugar(originalSqlNode);
+                    if (dingoSqlValidator.getHybridSearchMap().size() == 1) {
+                        String hybridSearchSql = dingoSqlValidator.getHybridSearchSql();
+                        LogUtils.info(log, "HybridSearchSql: {}", hybridSearchSql);
+                        SqlNode hybridSqlNode;
+                        parser = SqlParser.create(hybridSearchSql, DingoParser.PARSER_CONFIG);
+                        hybridSqlNode = parser.parseQuery();
+                        //syntacticSugar(hybridSqlNode);
+                        HybridNodeUtils.lockUpHybridSearchNode(originalSqlNode, hybridSqlNode);
+                    } else {
+                        ConcurrentHashMap<SqlBasicCall, SqlNode> sqlNodeHashMap = new ConcurrentHashMap<>();
+                        for (Map.Entry<SqlBasicCall, String> entry : dingoSqlValidator
+                            .getHybridSearchMap().entrySet()) {
+                            SqlBasicCall key = entry.getKey();
+                            String value = entry.getValue();
+                            SqlNode hybridSqlNode;
+                            parser = SqlParser.create(value, config);
+                            hybridSqlNode = parser.parseQuery();
+                            //syntacticSugar(hybridSqlNode);
+                            sqlNodeHashMap.put(key, hybridSqlNode);
+                        }
+                        HybridNodeUtils.lockUpHybridSearchNode(originalSqlNode, sqlNodeHashMap);
+                    }
+                    //LogUtils.info(log, "HybridSearch Rewrite Sql: {}", originalSqlNode.toString());
+                    try {
+                        sqlNode = dingoSqlValidator.validate(originalSqlNode);
+                    } catch (CalciteContextException e) {
+                        LogUtils.error(log, "HybridSearch parse and validate error, sql: <[{}]>.",
+                            table.getCreateSql(), e);
+                        throw e;
+                    }
+                }
+                HintPredicate hintPredicate = (hint, rel) -> true;
+                HintStrategyTable hintStrategyTable = new HintStrategyTable.Builder()
+                    .hintStrategy("vector_pre", hintPredicate)
+                    .hintStrategy(HybridSearchTable.HINT_NAME, hintPredicate)
+                    .hintStrategy("disable_index", hintPredicate)
+                    .hintStrategy("text_search_pre", hintPredicate)
+                    .build();
+                SqlToRelConverter sqlToRelConverter = new DingoSqlToRelConverter(
+                    ViewExpanders.simpleContext(context.getCluster()),
+                    dingoSqlValidator,
+                    this.context.getCatalogReader(),
+                    context.getCluster(),
+                    sqlNode.getKind() == SqlKind.EXPLAIN,
+                    hintStrategyTable
+                );
+
+                return sqlToRelConverter.convertQuery(sqlNode, false, true).rel;
+            } catch (SqlParseException e) {
+                throw new RuntimeException(e);
+            }
+        }
     }
 
     @Override
diff --git a/dingo-calcite/src/main/java/io/dingodb/calcite/schema/SubSnapshotSchema.java b/dingo-calcite/src/main/java/io/dingodb/calcite/schema/SubSnapshotSchema.java
index 995c52f652..54657bcc5a 100644
--- a/dingo-calcite/src/main/java/io/dingodb/calcite/schema/SubSnapshotSchema.java
+++ b/dingo-calcite/src/main/java/io/dingodb/calcite/schema/SubSnapshotSchema.java
@@ -76,7 +76,6 @@ public SubSnapshotSchema(InfoSchema is, String schemaName, DingoParserContext co
         return new DingoTable(
             context,
             ImmutableList.<String>builder().addAll(names).add(tableName).build(),
-            null,
             table
         );
     }
@@ -116,7 +115,6 @@ public SubSnapshotSchema(InfoSchema is, String schemaName, DingoParserContext co
         return new DingoTable(
             context,
             ImmutableList.<String>builder().addAll(names).add(tableName).build(),
-            null,
             table
         );
     }
diff --git a/dingo-calcite/src/main/java/org/apache/calcite/sql/validate/TableHybridFunctionNamespace.java b/dingo-calcite/src/main/java/org/apache/calcite/sql/validate/TableHybridFunctionNamespace.java
index 39ce5935cf..e5de49a52c 100644
--- a/dingo-calcite/src/main/java/org/apache/calcite/sql/validate/TableHybridFunctionNamespace.java
+++ b/dingo-calcite/src/main/java/org/apache/calcite/sql/validate/TableHybridFunctionNamespace.java
@@ -162,9 +162,6 @@ protected RelDataType validateImpl(RelDataType targetRowType) {
             );
             this.rowType = rowType;
 
-//            if (((DingoSqlValidator)validator).isHybridSearch()) {
-//                throw new RuntimeException("Multiple hybridSearch in SQL is not supported");
-//            }
             String sql = HybridSearchSqlUtils.hybridSearchSqlReplace(
                 vectorWeight,
                 documentWeight,
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 63f450aa8a..cff193fa82 100644
--- a/dingo-calcite/src/test/java/io/dingodb/calcite/TestCreateTable.java
+++ b/dingo-calcite/src/test/java/io/dingodb/calcite/TestCreateTable.java
@@ -160,7 +160,7 @@ public void revoke() {
 
     @Test
     public void flush() {
-        String sql = "flush privileges";
+        String sql = "create view v2 as select count(1) from t1";
         SqlParser.Config config = SqlParser.config().withParserFactory(DingoSqlParserImpl::new);
         SqlParser parser = SqlParser.create(sql, config);
         try {
diff --git a/dingo-common/src/main/java/io/dingodb/common/ddl/ActionType.java b/dingo-common/src/main/java/io/dingodb/common/ddl/ActionType.java
index 336f0c50ea..82d9055b96 100644
--- a/dingo-common/src/main/java/io/dingodb/common/ddl/ActionType.java
+++ b/dingo-common/src/main/java/io/dingodb/common/ddl/ActionType.java
@@ -28,6 +28,8 @@ public enum ActionType {
     ActionDropIndex(8),
     ActionTruncateTable(11),
     ActionModifyColumn(12),
+    ActionCreateView(21),
+    ActionDropView(24),
     ActionAddPrimaryKey(32),
     ActionCreateTables(60),
     ;
diff --git a/dingo-common/src/main/java/io/dingodb/common/ddl/DdlJob.java b/dingo-common/src/main/java/io/dingodb/common/ddl/DdlJob.java
index 693f10a0ae..cf2f5ea2b2 100644
--- a/dingo-common/src/main/java/io/dingodb/common/ddl/DdlJob.java
+++ b/dingo-common/src/main/java/io/dingodb/common/ddl/DdlJob.java
@@ -226,7 +226,7 @@ public String decodeArgs() {
                 return null;
             }
             TypeReference t = null;
-            if (actionType == ActionType.ActionCreateTable) {
+            if (actionType == ActionType.ActionCreateTable || actionType == ActionType.ActionCreateView) {
                 t = new TypeReference<List<TableDefinition>>() {};
             } else if (actionType == ActionType.ActionCreateSchema) {
                 t = new TypeReference<List<SchemaInfo>>() {};
diff --git a/dingo-dist/conf/logback-executor-release.xml b/dingo-dist/conf/logback-executor-release.xml
index 1481fd9187..6c648d70e2 100644
--- a/dingo-dist/conf/logback-executor-release.xml
+++ b/dingo-dist/conf/logback-executor-release.xml
@@ -447,6 +447,9 @@
     <logger name="io.dingodb.store.proxy.ddl.DdlHandler" level="${log.level}" additivity="false">
         <appender-ref ref="ASYNC-Ddl"/>
     </logger>
+    <logger name="io.dingodb.meta.ddl" level="${log.level}" additivity="false">
+        <appender-ref ref="ASYNC-Ddl"/>
+    </logger>
 
     <appender name="ASYNC-Sql" class="ch.qos.logback.classic.AsyncAppender">
         <includeCallerData>true</includeCallerData>
diff --git a/dingo-dist/conf/logback-executor.xml b/dingo-dist/conf/logback-executor.xml
index b71531e776..62d20e17e0 100644
--- a/dingo-dist/conf/logback-executor.xml
+++ b/dingo-dist/conf/logback-executor.xml
@@ -447,6 +447,9 @@
     <logger name="io.dingodb.store.proxy.ddl.DdlHandler" level="${log.level}" additivity="false">
         <appender-ref ref="ASYNC-Ddl"/>
     </logger>
+    <logger name="io.dingodb.meta.ddl" level="${log.level}" additivity="false">
+        <appender-ref ref="ASYNC-Ddl"/>
+    </logger>
 
     <appender name="ASYNC-Sql" class="ch.qos.logback.classic.AsyncAppender">
         <includeCallerData>true</includeCallerData>
diff --git a/dingo-driver/host/src/main/java/io/dingodb/driver/DingoDriverParser.java b/dingo-driver/host/src/main/java/io/dingodb/driver/DingoDriverParser.java
index 6a6ce80f81..54ef2282a3 100644
--- a/dingo-driver/host/src/main/java/io/dingodb/driver/DingoDriverParser.java
+++ b/dingo-driver/host/src/main/java/io/dingodb/driver/DingoDriverParser.java
@@ -362,7 +362,7 @@ public Meta.Signature parseQuery(
         startTs = transaction.getStartTs();
         Meta.StatementType statementType;
         RelDataType type;
-        SqlValidator validator = getSqlValidator();
+        DingoSqlValidator validator = getSqlValidator();
         try {
             sqlNode = validator.validate(sqlNode);
             switch (sqlNode.getKind()) {
@@ -377,16 +377,17 @@ public Meta.Signature parseQuery(
                     type = validator.getValidatedNodeType(sqlNode);
                     break;
             }
-            if (((DingoSqlValidator) validator).isHybridSearch()) {
+            if (validator.isHybridSearch()) {
                 SqlNode originalSqlNode;
                 try {
-                    originalSqlNode = parse(sql);
+                    originalSqlNode
+                        = parse(sql);
                 } catch (SqlParseException e) {
                     throw ExceptionUtils.toRuntime(e);
                 }
                 syntacticSugar(originalSqlNode);
-                if (((DingoSqlValidator) validator).getHybridSearchMap().size() == 1) {
-                    String hybridSearchSql = ((DingoSqlValidator) validator).getHybridSearchSql();
+                if (validator.getHybridSearchMap().size() == 1) {
+                    String hybridSearchSql = validator.getHybridSearchSql();
                     LogUtils.info(log, "HybridSearchSql: {}", hybridSearchSql);
                     SqlNode hybridSqlNode;
                     try {
@@ -398,7 +399,7 @@ public Meta.Signature parseQuery(
                     HybridNodeUtils.lockUpHybridSearchNode(originalSqlNode, hybridSqlNode);
                 } else {
                     ConcurrentHashMap<SqlBasicCall, SqlNode> sqlNodeHashMap = new ConcurrentHashMap<>();
-                    for (Map.Entry<SqlBasicCall, String> entry : ((DingoSqlValidator) validator).getHybridSearchMap().entrySet()) {
+                    for (Map.Entry<SqlBasicCall, String> entry : validator.getHybridSearchMap().entrySet()) {
                         SqlBasicCall key = entry.getKey();
                         String value = entry.getValue();
                         SqlNode hybridSqlNode;
@@ -652,7 +653,6 @@ public Meta.Signature retryQuery(
         Meta.StatementType statementType;
         markAutoIncForDml(relNode);
         Location currentLocation = MetaService.root().currentLocation();
-        Set<RelOptTable> tables = useTables(relNode, sqlNode);
         switch (sqlNode.getKind()) {
             case INSERT:
             case DELETE:
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 18fc5749fc..fa763b39e5 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
@@ -254,6 +254,12 @@ public Pair<Long, String> runDdlJob(DdlContext dc, DdlJob job) {
             case ActionAddColumn:
                 res = onAddColumn(dc, job);
                 break;
+            case ActionCreateView:
+                res = onCreateView(dc, job);
+                break;
+            case ActionDropView:
+                res = onDropView(dc, job);
+                break;
             default:
                 job.setState(JobState.jobStateCancelled);
                 break;
@@ -481,6 +487,7 @@ public static Pair<Long, String> onDropTable(DdlContext dc, DdlJob job) {
                     DingoMetrics.timer("metaDropTable").update(sub, TimeUnit.MILLISECONDS);
                 } catch (Exception e) {
                     LogUtils.error(log, "drop table error", e);
+                    return Pair.of(0L, "dropTableError:" + e.getMessage());
                 }
                 job.finishTableJob(JobState.jobStateDone, SchemaState.SCHEMA_NONE);
                 break;
@@ -1086,6 +1093,99 @@ public ReorgCtx newReorgCtx(ReorgInfo reorgInfo) {
         return reorgCtx;
     }
 
+    public static Pair<Long, String> onCreateView(DdlContext dc, DdlJob ddlJob) {
+        String err = ddlJob.decodeArgs();
+        if (err != null) {
+            ddlJob.setState(JobState.jobStateCancelled);
+            return Pair.of(0L, err);
+        }
+        Pair<TableDefinition, String> res = TableUtil.createView(ddlJob);
+        if (res.getValue() != null) {
+            return Pair.of(0L, res.getValue());
+        }
+        LogUtils.info(log, "[ddl] create view info done, jobId:{}", ddlJob.getId());
+
+        Pair<Long, String> res1 = updateSchemaVersion(dc, ddlJob);
+
+        if (res1.getValue() != null) {
+            return res1;
+        }
+        ddlJob.finishTableJob(JobState.jobStateDone, SchemaState.SCHEMA_PUBLIC);
+        LogUtils.debug(log, "[ddl] onCreateView done, jobId:{}", ddlJob.getId());
+        return res1;
+    }
+
+    public static Pair<Long, String> onDropView(DdlContext dc, DdlJob job) {
+        Pair<TableDefinitionWithId, String> tableRes = checkTableExistAndCancelNonExistJob(job, job.getSchemaId());
+        if (tableRes.getValue() != null && tableRes.getKey() == null) {
+            return Pair.of(0L, tableRes.getValue());
+        }
+        TableDefinitionWithId tableInfo = tableRes.getKey();
+        if (tableInfo == null) {
+            return Pair.of(0L, "view not exists");
+        }
+        if (job.getError() != null) {
+            if ("Lock wait timeout exceeded".equalsIgnoreCase(job.decodeError())
+                && tableInfo.getTableDefinition().getSchemaState() != SCHEMA_PUBLIC) {
+                tableInfo.getTableDefinition().setSchemaState(SCHEMA_PUBLIC);
+                ActionType originType = job.getActionType();
+                job.setActionType(ActionType.ActionCreateTable);
+                job.setState(JobState.jobStateCancelling);
+                Pair<Long, String> res = TableUtil.updateVersionAndTableInfos(dc, job, tableInfo, true);
+                job.setActionType(originType);
+                DdlContext.INSTANCE.getSchemaSyncer().ownerUpdateExpVersion(res.getKey());
+                return res;
+            }
+        }
+        SchemaState originalState = job.getSchemaState();
+        Pair<Long, String> res;
+        switch (tableInfo.getTableDefinition().getSchemaState()) {
+            case SCHEMA_PUBLIC:
+                tableInfo.getTableDefinition()
+                    .setSchemaState(SCHEMA_WRITE_ONLY);
+                res = TableUtil.updateVersionAndTableInfos(dc, job, tableInfo,
+                    originalState.getCode() != tableInfo.getTableDefinition().getSchemaState().number());
+                if (res.getValue() != null) {
+                    return res;
+                }
+                break;
+            case SCHEMA_WRITE_ONLY:
+                //tableInfo.getTableDefinition().setSchemaState(SCHEMA_DELETE_ONLY);
+                //res = TableUtil.updateVersionAndTableInfos(dc, job, tableInfo,
+                //    originalState.getCode() != tableInfo.getTableDefinition().getSchemaState().number());
+                //if (res.getValue() != null) {
+                //    return res;
+                //}
+                //break;
+                //case SCHEMA_DELETE_ONLY:
+                tableInfo.getTableDefinition().setSchemaState(SCHEMA_NONE);
+                long start = System.currentTimeMillis();
+                res = TableUtil.updateVersionAndTableInfos(dc, job, tableInfo,
+                    originalState.getCode() != tableInfo.getTableDefinition().getSchemaState().number());
+                long sub = System.currentTimeMillis() - start;
+                DingoMetrics.timer("updateVerAndTable").update(sub, TimeUnit.MILLISECONDS);
+                if (res.getValue() != null) {
+                    return res;
+                }
+                MetaService rootMs = MetaService.root();
+                MetaService ms = rootMs.getSubMetaService(job.getSchemaName());
+                try {
+                    start = System.currentTimeMillis();
+                    ms.dropTable(job.getSchemaId(), tableInfo.getTableDefinition().getName());
+                    sub = System.currentTimeMillis() - start;
+                    DingoMetrics.timer("metaDropView").update(sub, TimeUnit.MILLISECONDS);
+                } catch (Exception e) {
+                    LogUtils.error(log, "drop view error", e);
+                }
+                job.finishTableJob(JobState.jobStateDone, SchemaState.SCHEMA_NONE);
+                break;
+            default:
+                return Pair.of(0L, "ErrInvalidDDLState:" + tableInfo.getTableDefinition().getSchemaState());
+        }
+        job.setSchemaStateNumber(tableInfo.getTableDefinition().getSchemaState().number);
+        return res;
+    }
+
     public static void checkDropColumnForStatePublic(ColumnDefinition columnDefinition) {
         if (!columnDefinition.isNullable() && columnDefinition.getDefaultVal() == null) {
             DdlColumn.generateOriginDefaultValue(columnDefinition);
diff --git a/dingo-executor/src/main/java/io/dingodb/server/executor/ddl/TableUtil.java b/dingo-executor/src/main/java/io/dingodb/server/executor/ddl/TableUtil.java
index 43abdfd59b..1121c00f36 100644
--- a/dingo-executor/src/main/java/io/dingodb/server/executor/ddl/TableUtil.java
+++ b/dingo-executor/src/main/java/io/dingodb/server/executor/ddl/TableUtil.java
@@ -64,6 +64,47 @@ public static Pair<TableDefinition, String> createTable(DdlJob ddlJob) {
                 metaService.rollbackCreateTable(schemaId, tableInfo, indices);
                 LogUtils.error(log, "[ddl-error]" + e.getMessage(), e);
                 ddlJob.setState(JobState.jobStateCancelled);
+                if (e instanceof NullPointerException) {
+                    return Pair.of(null, "NullPointerException");
+                }
+                return Pair.of(null, e.getMessage());
+            }
+        }
+        return Pair.of(tableInfo, "ErrInvalidDDLState");
+    }
+
+    public static Pair<TableDefinition, String> createView(DdlJob ddlJob) {
+        long schemaId = ddlJob.getSchemaId();
+        TableDefinition tableInfo = (TableDefinition) ddlJob.getArgs().get(0);
+        tableInfo.setSchemaState(SchemaState.SCHEMA_NONE);
+        long tableId = ddlJob.getTableId();
+        tableInfo.setPrepareTableId(tableId);
+
+        InfoSchemaService service = InfoSchemaService.root();
+        Object tabObj = service.getTable(schemaId, tableInfo.getName());
+        if (tabObj != null) {
+            ddlJob.setState(JobState.jobStateCancelled);
+            return Pair.of(null, "view has existed");
+        }
+        if (tableInfo.getSchemaState() == SchemaState.SCHEMA_NONE) {
+            tableInfo.setSchemaState(SchemaState.SCHEMA_PUBLIC);
+            MetaService metaService = MetaService.root();
+            MetaService subMs = metaService.getSubMetaService(ddlJob.getSchemaName());
+            List<IndexDefinition> indices = tableInfo.getIndices();
+            if (indices != null) {
+                indices.forEach(index -> index.setSchemaState(SchemaState.SCHEMA_PUBLIC));
+            }
+            try {
+                assert indices != null;
+                subMs.createView(tableInfo.getName(), tableInfo);
+                return Pair.of(tableInfo, null);
+            } catch (Exception e) {
+                subMs.rollbackCreateTable(tableInfo, indices);
+                LogUtils.error(log, "[ddl-error]" + e.getMessage(), e);
+                ddlJob.setState(JobState.jobStateCancelled);
+                if (e instanceof NullPointerException) {
+                    return Pair.of(null, "NullPointerException");
+                }
                 return Pair.of(null, e.getMessage());
             }
         }
diff --git a/dingo-meta-api/src/main/java/io/dingodb/meta/DdlService.java b/dingo-meta-api/src/main/java/io/dingodb/meta/DdlService.java
index a7d2de106c..969bce7963 100644
--- a/dingo-meta-api/src/main/java/io/dingodb/meta/DdlService.java
+++ b/dingo-meta-api/src/main/java/io/dingodb/meta/DdlService.java
@@ -34,7 +34,11 @@ static DdlService root() {
 
     void dropSchema(SchemaInfo schemaInfo, String connId);
 
-    void createTableWithInfo(String schemaName, String tableName, @NonNull TableDefinition tableDefinition, String connId, String sql);
+    void createTableWithInfo(String schemaName, String tableName,
+                             TableDefinition tableDefinition, String connId, String sql);
+
+    default void createViewWithInfo(String schemaName, String tableName,
+                             TableDefinition tableDefinition, String connId, String sql) {}
 
     void dropTable(SchemaInfo schemaInfo, Long tableId, String tableName, String connId);
 
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 210c7ff21e..34860c071a 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
@@ -107,6 +107,8 @@ default void createTable(@NonNull String tableName, @NonNull TableDefinition tab
         createTables(tableDefinition, Collections.emptyList());
     }
 
+    default void createView(String viewName, TableDefinition tableDefinition) {}
+
     default long createReplicaTable(long schemaId, Object tableDefinition, String tableName) {
         return 0L;
     }
@@ -254,11 +256,11 @@ default void invalidateDistribution(CommonId tableId) {}
 
     default void deleteRegionByTableId(CommonId tableId) {}
 
-    default void createTenant(Tenant tenant) {};
+    default void createTenant(Tenant tenant) {}
 
-    default void updateTenant(Tenant tenant) {};
+    default void updateTenant(Tenant tenant) {}
 
-    default void deleteTenant(long tenantId) {};
+    default void deleteTenant(long tenantId) {}
 
     void close();
 }
diff --git a/dingo-meta-api/src/main/java/io/dingodb/meta/ddl/InfoSchemaBuilder.java b/dingo-meta-api/src/main/java/io/dingodb/meta/ddl/InfoSchemaBuilder.java
index 59b7038c3c..bce0347589 100644
--- a/dingo-meta-api/src/main/java/io/dingodb/meta/ddl/InfoSchemaBuilder.java
+++ b/dingo-meta-api/src/main/java/io/dingodb/meta/ddl/InfoSchemaBuilder.java
@@ -162,6 +162,7 @@ public Pair<List<Long>, String> applyDiff(InfoSchemaService infoSchemaService, S
                 return applyDropTable(schemaDiff);
             case ActionDropIndex:
                 return applyDropIndex(schemaDiff);
+            case ActionCreateView:
             case ActionCreateTable:
                 return applyCreateTable(schemaDiff);
             case ActionTruncateTable:
diff --git a/dingo-store-proxy/src/main/java/io/dingodb/store/proxy/ddl/DdlHandler.java b/dingo-store-proxy/src/main/java/io/dingodb/store/proxy/ddl/DdlHandler.java
index 207ac875b7..eac3cbe919 100644
--- a/dingo-store-proxy/src/main/java/io/dingodb/store/proxy/ddl/DdlHandler.java
+++ b/dingo-store-proxy/src/main/java/io/dingodb/store/proxy/ddl/DdlHandler.java
@@ -168,6 +168,22 @@ public static void createTableWithInfo(
         }
     }
 
+    public static void createViewWithInfo(
+        String schemaName,
+        TableDefinition tableDefinition,
+        String connId,
+        String sql
+    ) {
+        DdlJob ddlJob = createViewWithInfoJob(schemaName, tableDefinition);
+        ddlJob.setConnId(connId);
+        try {
+            doDdlJob(ddlJob);
+        } catch (Exception e) {
+            LogUtils.error(log, "[ddl-error] create table error,reason:" + e.getMessage() + ", tabDef" + tableDefinition, e);
+            throw e;
+        }
+    }
+
     public static void dropTable(SchemaInfo schemaInfo, Long tableId, String tableName, String connId) {
         DdlJob job = DdlJob.builder()
             .actionType(ActionType.ActionDropTable)
@@ -394,6 +410,32 @@ public static DdlJob createTableWithInfoJob(String schemaName, TableDefinition t
             .build();
     }
 
+    public static DdlJob createViewWithInfoJob(String schemaName, TableDefinition tableDefinition) {
+        InfoSchemaService infoSchemaService = InfoSchemaService.root();
+        assert infoSchemaService != null;
+        SchemaInfo schemaInfo = infoSchemaService.getSchema(schemaName);
+        List<Object> args = new ArrayList<>();
+        args.add(tableDefinition);
+
+        CoordinatorService coordinatorService = Services.coordinatorService(Configuration.coordinatorSet());
+        long tableEntityId = coordinatorService.createIds(
+            TsoService.getDefault().tso(),
+            CreateIdsRequest.builder()
+                .idEpochType(IdEpochType.ID_NEXT_TABLE).count(1)
+                .build()
+        ).getIds().get(0);
+        return DdlJob.builder()
+            .schemaId(schemaInfo.getSchemaId())
+            .schemaName(schemaInfo.getName())
+            .tableName(tableDefinition.getName())
+            .actionType(ActionType.ActionCreateView)
+            .state(JobState.jobStateQueueing)
+            .args(args)
+            .tableId(tableEntityId)
+            .id(0)
+            .build();
+    }
+
     public static void doDdlJob(DdlJob job) {
         long timeout = ScopeVariables.getDdlWaitTimeout();
         if (job.mayNeedReorg()) {
diff --git a/dingo-store-proxy/src/main/java/io/dingodb/store/proxy/ddl/DdlService.java b/dingo-store-proxy/src/main/java/io/dingodb/store/proxy/ddl/DdlService.java
index c4a5ac73fc..d505d70628 100644
--- a/dingo-store-proxy/src/main/java/io/dingodb/store/proxy/ddl/DdlService.java
+++ b/dingo-store-proxy/src/main/java/io/dingodb/store/proxy/ddl/DdlService.java
@@ -54,10 +54,17 @@ public void dropSchema(SchemaInfo schemaInfo, String connId) {
     }
 
     @Override
-    public void createTableWithInfo(String schemaName, String tableName, @NonNull TableDefinition tableDefinition, String connId, String sql) {
+    public void createTableWithInfo(String schemaName, String tableName,
+        TableDefinition tableDefinition, String connId, String sql) {
         DdlHandler.createTableWithInfo(schemaName, tableDefinition, connId, sql);
     }
 
+    @Override
+    public void createViewWithInfo(String schemaName, String tableName,
+        TableDefinition tableDefinition, String connId, String sql) {
+        DdlHandler.createViewWithInfo(schemaName, tableDefinition, connId, sql);
+    }
+
     @Override
     public void dropTable(SchemaInfo schemaInfo, Long tableId, String tableName, String connId) {
         DdlHandler.dropTable(schemaInfo, tableId, tableName, connId);
diff --git a/dingo-store-proxy/src/main/java/io/dingodb/store/proxy/mapper/TableMapper.java b/dingo-store-proxy/src/main/java/io/dingodb/store/proxy/mapper/TableMapper.java
index 885289592c..8515062b11 100644
--- a/dingo-store-proxy/src/main/java/io/dingodb/store/proxy/mapper/TableMapper.java
+++ b/dingo-store-proxy/src/main/java/io/dingodb/store/proxy/mapper/TableMapper.java
@@ -53,8 +53,6 @@
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
-import javax.xml.bind.ValidationException;
-
 import static io.dingodb.partition.DingoPartitionServiceProvider.HASH_FUNC_NAME;
 import static io.dingodb.partition.DingoPartitionServiceProvider.RANGE_FUNC_NAME;
 import static io.dingodb.sdk.service.entity.meta.PartitionStrategy.PT_STRATEGY_HASH;
@@ -190,14 +188,18 @@ default Table tableFrom(
         Table.TableBuilder builder = Table.builder();
         tableFrom(definition, builder);
         PartitionRule partitionRule = definition.getTablePartition();
-        builder.partitionStrategy(fromPartitionStrategy(partitionRule.getStrategy()));
+        if (partitionRule != null) {
+            builder.partitionStrategy(fromPartitionStrategy(partitionRule.getStrategy()));
+        }
         KeyValueCodec codec = CodecService.INSTANCE
             .createKeyValueCodec(definition.getVersion(), columnDefinitionFrom(definition.getColumns()));
-        builder.partitions(partitionFrom(
-            definition.getTablePartition().getPartitions(),
-            codec,
-            fromPartitionStrategy(partitionRule.getStrategy()))
-        );
+        if (definition.getTablePartition() != null) {
+            builder.partitions(partitionFrom(
+                definition.getTablePartition().getPartitions(),
+                codec,
+                fromPartitionStrategy(partitionRule.getStrategy()))
+            );
+        }
         builder.schemaState(io.dingodb.common.meta.SchemaState.get(tableWithId.getTableDefinition().getSchemaState().number));
         builder.tableId(MAPPER.idFrom(tableWithId.getTableId()));
         builder.indexes(indexes.stream().map($ -> indexTableFrom(builder, $, Collections.emptyList()))
@@ -251,12 +253,15 @@ default TableDefinitionWithId tableTo(
         tableDefinition.setEngine(tableDefinition.getEngine().toUpperCase());
         byte namespace = (byte) (tableDefinition.getEngine().startsWith("TXN") ? 't' : 'r');
         io.dingodb.sdk.service.entity.meta.TableDefinition definition = tableTo(tableDefinition);
-        definition.setTablePartition(
-            partitionTo(tableDefinition.getPartDefinition(), ids.getPartIds(), encoder, namespace)
-        );
+        if (tableDefinition.getPartDefinition() != null) {
+            definition.setTablePartition(
+                partitionTo(tableDefinition.getPartDefinition(), ids.getPartIds(), encoder, namespace)
+            );
+        }
         definition.setName(definition.getName().toUpperCase());
         definition.setSchemaState(convertSchemaState(tableDefinition.getSchemaState()));
-        return TableDefinitionWithId.builder().tenantId(tenantId).tableDefinition(definition).tableId(ids.getTableId()).build();
+        return TableDefinitionWithId.builder().tenantId(tenantId)
+            .tableDefinition(definition).tableId(ids.getTableId()).build();
     }
 
     default SchemaState convertSchemaState(io.dingodb.common.meta.SchemaState schemaState) {
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 3184c3c3cb..d4528d50c6 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
@@ -254,6 +254,50 @@ public boolean dropSubMetaService(String name) {
         return true;
     }
 
+    @Override
+    public void createView(String tableName, TableDefinition tableDefinition) {
+        CoordinatorService coordinatorService = Services.coordinatorService(Configuration.coordinatorSet());
+
+        long tableEntityId;
+        if (tableDefinition.getPrepareTableId() != 0) {
+            tableEntityId = tableDefinition.getPrepareTableId();
+        } else {
+            // Generate new table ids.
+            tableEntityId = coordinatorService.createIds(
+                tso(),
+                CreateIdsRequest.builder()
+                    .idEpochType(IdEpochType.ID_NEXT_TABLE).count(1)
+                    .build()
+            ).getIds().get(0);
+        }
+        DingoCommonId tableId = DingoCommonId.builder()
+            .entityType(EntityType.ENTITY_TYPE_TABLE)
+            .parentEntityId(id.getEntityId())
+            .entityId(tableEntityId).build();
+//        List<DingoCommonId> tablePartIds = coordinatorService.createIds(tso(), CreateIdsRequest.builder()
+//                .idEpochType(IdEpochType.ID_NEXT_TABLE)
+//                .count(tableDefinition.getPartDefinition().getDetails().size())
+//                .build()
+//            )
+//            .getIds()
+//            .stream()
+//            .map(id -> DingoCommonId.builder()
+//                .entityType(EntityType.ENTITY_TYPE_PART)
+//                .parentEntityId(tableEntityId)
+//                .entityId(id).build())
+//            .collect(Collectors.toList());
+        TableIdWithPartIds tableIdWithPartIds =
+            TableIdWithPartIds.builder().tableId(tableId).build();
+        TableDefinitionWithId tableDefinitionWithId = MAPPER.tableTo(tableIdWithPartIds, tableDefinition, TenantConstant.TENANT_ID);
+        // create view
+        infoSchemaService.createTableOrView(
+            id.getEntityId(),
+            tableDefinitionWithId.getTableId().getEntityId(),
+            tableDefinitionWithId
+        );
+    }
+
+
     @Override
     public long createTables(
         @NonNull TableDefinition tableDefinition, @NonNull List<IndexDefinition> indexTableDefinitions
@@ -933,28 +977,29 @@ public boolean dropTable(long tenantId, long schemaId, String tableName) {
         if (table == null) {
             return false;
         }
-        List<IndexTable> indexes = table.getIndexes();
-
-        loadDistribution(table.tableId, tenantId, table).values().forEach(rangeDistribution -> {
-            coordinatorService.dropRegion(tso(), DropRegionRequest.builder().regionId(rangeDistribution.id().seq).build());
-        });
+        if (!"view".equalsIgnoreCase(table.getTableType())) {
+            loadDistribution(table.tableId, tenantId, table).values().forEach(rangeDistribution -> {
+                coordinatorService.dropRegion(tso(), DropRegionRequest.builder().regionId(rangeDistribution.id().seq).build());
+            });
+            List<IndexTable> indexes = table.getIndexes();
 
-        for (IndexTable index : indexes) {
-            loadDistribution(index.tableId, tenantId, index)
-                .values()
-                .forEach(rangeDistribution ->
-                    coordinatorService.dropRegion(
-                        tso(),
-                        DropRegionRequest.builder().regionId(rangeDistribution.id().seq).build()
-                    )
-                );
+            for (IndexTable index : indexes) {
+                loadDistribution(index.tableId, tenantId, index)
+                    .values()
+                    .forEach(rangeDistribution ->
+                        coordinatorService.dropRegion(
+                            tso(),
+                            DropRegionRequest.builder().regionId(rangeDistribution.id().seq).build()
+                        )
+                    );
+            }
+            List<CommonId> indexIds = indexes.stream().map(Table::getTableId).collect(Collectors.toList());
+            indexIds.forEach(indexId -> {
+                infoSchemaService.dropIndex(indexId.domain, indexId.seq);
+            });
         }
 
-        List<CommonId> indexIds = indexes.stream().map(Table::getTableId).collect(Collectors.toList());
         infoSchemaService.dropTable(table.getTableId().domain, table.tableId.seq);
-        indexIds.forEach(indexId -> {
-            infoSchemaService.dropIndex(indexId.domain, indexId.seq);
-        });
         return true;
     }
 
diff --git a/dingo-test/meta-local/src/main/java/io/dingodb/meta/local/LocalMetaService.java b/dingo-test/meta-local/src/main/java/io/dingodb/meta/local/LocalMetaService.java
index 104c83a71f..5c276538c7 100644
--- a/dingo-test/meta-local/src/main/java/io/dingodb/meta/local/LocalMetaService.java
+++ b/dingo-test/meta-local/src/main/java/io/dingodb/meta/local/LocalMetaService.java
@@ -144,6 +144,10 @@ public void createTable(@NonNull String tableName, @NonNull TableDefinition tabl
         }
     }
 
+    public void createView(String tableName, TableDefinition tableDefinition) {
+
+    }
+
     public long createTablesByTruncate(@NonNull TableDefinition tableDefinition,
                              @NonNull List<IndexDefinition> indexTableDefinitions) {
         CommonId tableId = new CommonId(TABLE, id.seq, tableSeq.incrementAndGet());