Add view support (#4459)

add view support in MO

Approved by: @fengttt, @iamlinjunhong, @aunjgr, @daviszhen, @nnsgmsone, @XuPeng-SH, @reusee, @aressu1985

pkg/frontend/mysql_cmd_executor.go

@@ -2046,6 +2046,7 @@ func (mce *MysqlCmdExecutor) doComQuery(sql string) (retErr error) {
 		//just status, no result set
 		case *tree.CreateTable, *tree.DropTable, *tree.CreateDatabase, *tree.DropDatabase,
 			*tree.CreateIndex, *tree.DropIndex,
+			*tree.CreateView, *tree.DropView,
 			*tree.Insert, *tree.Update,
 			*tree.BeginTransaction, *tree.CommitTransaction, *tree.RollbackTransaction,
 			*tree.SetVar,
@@ -2084,6 +2085,7 @@ func (mce *MysqlCmdExecutor) doComQuery(sql string) (retErr error) {
 			switch stmt.(type) {
 			case *tree.CreateTable, *tree.DropTable, *tree.CreateDatabase, *tree.DropDatabase,
 				*tree.CreateIndex, *tree.DropIndex, *tree.Insert, *tree.Update,
+				*tree.CreateView, *tree.DropView,
 				*tree.CreateUser, *tree.DropUser, *tree.AlterUser,
 				*tree.CreateRole, *tree.DropRole, *tree.Revoke, *tree.Grant,
 				*tree.SetDefaultRole, *tree.SetRole, *tree.SetPassword, *tree.Delete,
@@ -2248,7 +2250,9 @@ func StatementCanBeExecutedInUncommittedTransaction(stmt tree.Statement) bool {
 // IsDDL checks the statement is the DDL statement.
 func IsDDL(stmt tree.Statement) bool {
 	switch stmt.(type) {
-	case *tree.CreateTable, *tree.DropTable, *tree.CreateDatabase, *tree.DropDatabase,
+	case *tree.CreateTable, *tree.DropTable,
+		*tree.CreateView, *tree.DropView,
+		*tree.CreateDatabase, *tree.DropDatabase,
 		*tree.CreateIndex, *tree.DropIndex:
 		return true
 	}
@@ -2258,7 +2262,7 @@ func IsDDL(stmt tree.Statement) bool {
 // IsDropStatement checks the statement is the drop statement.
 func IsDropStatement(stmt tree.Statement) bool {
 	switch stmt.(type) {
-	case *tree.DropDatabase, *tree.DropTable, *tree.DropIndex:
+	case *tree.DropDatabase, *tree.DropTable, *tree.DropView, *tree.DropIndex:
 		return true
 	}
 	return false

pkg/frontend/session.go

@@ -633,6 +633,10 @@ func (tcc *TxnCompilerContext) DefaultDatabase() string {
 	return tcc.dbName
 }
 
+func (tcc *TxnCompilerContext) GetRootSql() string {
+	return tcc.ses.GetSql()
+}
+
 func (tcc *TxnCompilerContext) DatabaseExists(name string) bool {
 	var err error
 	//open database
@@ -700,10 +704,11 @@ func (tcc *TxnCompilerContext) Resolve(dbName string, tableName string) (*plan2.
 		return nil, nil
 	}
 
-	var defs []*plan2.ColDef
+	var cols []*plan2.ColDef
+	var defs []*plan2.TableDefType
 	for _, def := range engineDefs {
 		if attr, ok := def.(*engine.AttributeDef); ok {
-			defs = append(defs, &plan2.ColDef{
+			cols = append(cols, &plan2.ColDef{
 				Name: attr.Attr.Name,
 				Typ: &plan2.Type{
 					Id:        int32(attr.Attr.Type.Oid),
@@ -714,6 +719,29 @@ func (tcc *TxnCompilerContext) Resolve(dbName string, tableName string) (*plan2.
 				Primary: attr.Attr.Primary,
 				Default: attr.Attr.Default,
 			})
+		} else if pro, ok := def.(*engine.PropertiesDef); ok {
+			properties := make([]*plan2.Property, len(pro.Properties))
+			for i, p := range pro.Properties {
+				properties[i] = &plan2.Property{
+					Key:   p.Key,
+					Value: p.Value,
+				}
+			}
+			defs = append(defs, &plan2.TableDefType{
+				Def: &plan2.TableDef_DefType_Properties{
+					Properties: &plan2.PropertiesDef{
+						Properties: properties,
+					},
+				},
+			})
+		} else if viewDef, ok := def.(*engine.ViewDef); ok {
+			defs = append(defs, &plan2.TableDefType{
+				Def: &plan2.TableDef_DefType_View{
+					View: &plan2.ViewDef{
+						View: viewDef.View,
+					},
+				},
+			})
 		}
 	}
 	if tcc.QryTyp != TXN_DEFAULT {
@@ -722,7 +750,7 @@ func (tcc *TxnCompilerContext) Resolve(dbName string, tableName string) (*plan2.
 			return nil, nil
 		}
 		hideKey := hideKeys[0]
-		defs = append(defs, &plan2.ColDef{
+		cols = append(cols, &plan2.ColDef{
 			Name: hideKey.Name,
 			Typ: &plan2.Type{
 				Id:        int32(hideKey.Type.Oid),
@@ -742,7 +770,8 @@ func (tcc *TxnCompilerContext) Resolve(dbName string, tableName string) (*plan2.
 
 	tableDef := &plan2.TableDef{
 		Name: tableName,
-		Cols: defs,
+		Cols: cols,
+		Defs: defs,
 	}
 	return obj, tableDef
 }

pkg/sql/compile/ddl.go

@@ -124,6 +124,10 @@ func planDefsToExeDefs(planDefs []*plan.TableDef_DefType) []engine.TableDef {
 			exeDefs[i] = &engine.PropertiesDef{
 				Properties: properties,
 			}
+		case *plan.TableDef_DefType_View:
+			exeDefs[i] = &engine.ViewDef{
+				View: defVal.View.View,
+			}
 		}
 	}
 	return exeDefs

pkg/sql/parsers/dialect/mysql/mysql_sql.y

@@ -324,7 +324,7 @@ import (
 %type <statements> stmt_list
 %type <statement> create_stmt insert_stmt delete_stmt drop_stmt alter_stmt
 %type <statement> delete_without_using_stmt delete_with_using_stmt
-%type <statement> drop_ddl_stmt drop_database_stmt drop_table_stmt drop_index_stmt drop_prepare_stmt
+%type <statement> drop_ddl_stmt drop_database_stmt drop_table_stmt drop_index_stmt drop_prepare_stmt drop_view_stmt
 %type <statement> drop_account_stmt drop_role_stmt drop_user_stmt
 %type <statement> create_account_stmt create_user_stmt create_role_stmt
 %type <statement> create_ddl_stmt create_table_stmt create_database_stmt create_index_stmt create_view_stmt
@@ -2157,6 +2157,7 @@ drop_ddl_stmt:
     drop_database_stmt
 |   drop_prepare_stmt
 |   drop_table_stmt
+|   drop_view_stmt
 |   drop_index_stmt
 |   drop_role_stmt
 |   drop_user_stmt
@@ -2224,6 +2225,12 @@ drop_table_stmt:
         $$ = &tree.DropTable{IfExists: $3, Names: $4}
     }
 
+drop_view_stmt:
+    DROP VIEW exists_opt table_name_list
+    {
+        $$ = &tree.DropView{IfExists: $3, Names: $4}
+    }
+
 drop_database_stmt:
     DROP DATABASE exists_opt database_id
     {

pkg/sql/parsers/tree/drop.go

@@ -63,6 +63,29 @@ func NewDropTable(i bool, n TableNames) *DropTable {
 	}
 }
 
+// DropView DROP View statement
+type DropView struct {
+	statementImpl
+	IfExists bool
+	Names    TableNames
+}
+
+func (node *DropView) Format(ctx *FmtCtx) {
+	ctx.WriteString("drop view")
+	if node.IfExists {
+		ctx.WriteString(" if exists")
+	}
+	ctx.WriteByte(' ')
+	node.Names.Format(ctx)
+}
+
+func NewDropView(i bool, n TableNames) *DropView {
+	return &DropView{
+		IfExists: i,
+		Names:    n,
+	}
+}
+
 type DropIndex struct {
 	statementImpl
 	Name       Identifier

pkg/sql/plan/bind_context.go

@@ -33,6 +33,9 @@ func NewBindContext(builder *QueryBuilder, parent *BindContext) *BindContext {
 		bindingByCol:   make(map[string]*Binding),
 		parent:         parent,
 	}
+	if parent != nil {
+		bc.defaultDatabase = parent.defaultDatabase
+	}
 
 	return bc
 }

pkg/sql/plan/build.go

@@ -70,6 +70,10 @@ func BuildPlan(ctx CompilerContext, stmt tree.Statement) (*Plan, error) {
 		return buildCreateTable(stmt, ctx)
 	case *tree.DropTable:
 		return buildDropTable(stmt, ctx)
+	case *tree.DropView:
+		return buildDropView(stmt, ctx)
+	case *tree.CreateView:
+		return buildCreateView(stmt, ctx)
 	case *tree.CreateIndex:
 		return buildCreateIndex(stmt, ctx)
 	case *tree.DropIndex:

pkg/sql/plan/build_ddl.go

@@ -15,6 +15,7 @@
 package plan
 
 import (
+	"encoding/json"
 	"fmt"
 
 	"github.com/matrixorigin/matrixone/pkg/container/types"
@@ -25,6 +26,72 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/sql/parsers/tree"
 )
 
+func buildCreateView(stmt *tree.CreateView, ctx CompilerContext) (*Plan, error) {
+	viewName := stmt.Name.ObjectName
+	createTable := &plan.CreateTable{
+		IfNotExists: stmt.IfNotExists,
+		Temporary:   stmt.Temporary,
+		TableDef: &TableDef{
+			Name: string(viewName),
+		},
+	}
+
+	// get database name
+	if len(stmt.Name.SchemaName) == 0 {
+		createTable.Database = ""
+	} else {
+		createTable.Database = string(stmt.Name.SchemaName)
+	}
+
+	// check view statement
+	stmtPlan, err := runBuildSelectByBinder(plan.Query_SELECT, ctx, stmt.AsSource)
+	if err != nil {
+		return nil, err
+	}
+
+	query := stmtPlan.GetQuery()
+	cols := make([]*plan.ColDef, len(query.Headings))
+	for idx, expr := range query.Nodes[query.Steps[len(query.Steps)-1]].ProjectList {
+		cols[idx] = &plan.ColDef{
+			Name: query.Headings[idx],
+			Alg:  plan.CompressType_Lz4,
+			Typ:  expr.Typ,
+			Default: &plan.Default{
+				NullAbility:  false,
+				Expr:         nil,
+				OriginString: "",
+			},
+		}
+	}
+	createTable.TableDef.Cols = cols
+
+	viewData, err := json.Marshal(ViewData{
+		Stmt:            ctx.GetRootSql(),
+		DefaultDatabase: ctx.DefaultDatabase(),
+	})
+	if err != nil {
+		return nil, err
+	}
+	createTable.TableDef.Defs = append(createTable.TableDef.Defs, &plan.TableDef_DefType{
+		Def: &plan.TableDef_DefType_View{
+			View: &plan.ViewDef{
+				View: string(viewData),
+			},
+		},
+	})
+
+	return &Plan{
+		Plan: &plan.Plan_Ddl{
+			Ddl: &plan.DataDefinition{
+				DdlType: plan.DataDefinition_CREATE_TABLE,
+				Definition: &plan.DataDefinition_CreateTable{
+					CreateTable: createTable,
+				},
+			},
+		},
+	}, nil
+}
+
 func buildCreateTable(stmt *tree.CreateTable, ctx CompilerContext) (*Plan, error) {
 	createTable := &plan.CreateTable{
 		IfNotExists: stmt.IfNotExists,
@@ -263,6 +330,67 @@ func buildDropTable(stmt *tree.DropTable, ctx CompilerContext) (*Plan, error) {
 	}
 	dropTable.Table = string(stmt.Names[0].ObjectName)
 
+	_, tableDef := ctx.Resolve(dropTable.Database, dropTable.Table)
+	if tableDef == nil {
+		if !dropTable.IfExists {
+			return nil, errors.New("", fmt.Sprintf("table %s is not exists", dropTable.Table))
+		}
+	} else {
+		isView := false
+		for _, def := range tableDef.Defs {
+			if _, ok := def.Def.(*plan.TableDef_DefType_View); ok {
+				isView = true
+				break
+			}
+		}
+		if isView {
+			return nil, errors.New("", fmt.Sprintf("table %s is not exists", dropTable.Table))
+		}
+	}
+
+	return &Plan{
+		Plan: &plan.Plan_Ddl{
+			Ddl: &plan.DataDefinition{
+				DdlType: plan.DataDefinition_DROP_TABLE,
+				Definition: &plan.DataDefinition_DropTable{
+					DropTable: dropTable,
+				},
+			},
+		},
+	}, nil
+}
+
+func buildDropView(stmt *tree.DropView, ctx CompilerContext) (*Plan, error) {
+	dropTable := &plan.DropTable{
+		IfExists: stmt.IfExists,
+	}
+	if len(stmt.Names) != 1 {
+		return nil, errors.New(errno.SyntaxErrororAccessRuleViolation, "support drop one view now")
+	}
+	dropTable.Database = string(stmt.Names[0].SchemaName)
+	if dropTable.Database == "" {
+		dropTable.Database = ctx.DefaultDatabase()
+	}
+	dropTable.Table = string(stmt.Names[0].ObjectName)
+
+	_, tableDef := ctx.Resolve(dropTable.Database, dropTable.Table)
+	if tableDef == nil {
+		if !dropTable.IfExists {
+			return nil, errors.New("", fmt.Sprintf("view %s is not exists", dropTable.Table))
+		}
+	} else {
+		isView := false
+		for _, def := range tableDef.Defs {
+			if _, ok := def.Def.(*plan.TableDef_DefType_View); ok {
+				isView = true
+				break
+			}
+		}
+		if !isView {
+			return nil, errors.New("", fmt.Sprintf("%s is not a view", dropTable.Table))
+		}
+	}
+
 	return &Plan{
 		Plan: &plan.Plan_Ddl{
 			Ddl: &plan.DataDefinition{

pkg/sql/plan/build_test.go

@@ -31,7 +31,7 @@ func TestSingleSQL(t *testing.T) {
 	// sql := "SELECT nation2.* FROM nation2 natural join region"
 	// sql := `select n_name, avg(N_REGIONKEY) t from NATION where n_name != 'a' group by n_name having avg(N_REGIONKEY) > 10 order by t limit 20`
 	// sql := `select date_add('1997-12-31 23:59:59',INTERVAL 100000 SECOND)`
-	sql := "select 1, 2 union select 2, 3"
+	sql := "create view v1 as select * from nation"
 	// sql := "explain a"
 	// sql := "select 18446744073709551500"
 	// stmts, err := mysql.Parse(sql)
@@ -436,7 +436,7 @@ func TestSingleTableSQLBuilder(t *testing.T) {
 		"prepare stmt1 from 'delete from nation where n_nationkey > ?'",
 		"prepare stmt1 from 'insert into nation select * from nation2 where n_name = ?'",
 		"prepare stmt1 from 'select * from nation where n_name = ?'",
-		"prepare stmt1 from 'drop table t1'",
+		"prepare stmt1 from 'drop table if exists t1'",
 		"prepare stmt1 from 'create table t1 (a int)'",
 		"prepare stmt1 from select N_REGIONKEY from nation group by N_REGIONKEY having abs(nation.N_REGIONKEY - ?) > ?",
 		"execute stmt1",

pkg/sql/plan/mock.go

@@ -278,6 +278,10 @@ func (m *MockCompilerContext) DefaultDatabase() string {
 	return "tpch"
 }
 
+func (m *MockCompilerContext) GetRootSql() string {
+	return ""
+}
+
 func (m *MockCompilerContext) Resolve(dbName string, tableName string) (*ObjectRef, *TableDef) {
 	name := strings.ToLower(tableName)
 	return m.objects[name], m.tables[name]

pkg/sql/plan/query_builder.go

@@ -15,6 +15,7 @@
 package plan
 
 import (
+	"encoding/json"
 	"fmt"
 	"math"
 	"sort"
@@ -23,6 +24,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/sql/errors"
 	"github.com/matrixorigin/matrixone/pkg/sql/parsers/dialect"
+	"github.com/matrixorigin/matrixone/pkg/sql/parsers/dialect/mysql"
 	"github.com/matrixorigin/matrixone/pkg/sql/parsers/tree"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan/function"
 )
@@ -1467,6 +1469,10 @@ func (builder *QueryBuilder) buildTable(stmt tree.TableExpr, ctx *BindContext) (
 				subCtx := NewBindContext(builder, ctx)
 				subCtx.maskedCTEs = cteRef.maskedCTEs
 				subCtx.cteName = table
+				//reset defaultDatabase
+				if len(cteRef.defaultDatabase) > 0 {
+					subCtx.defaultDatabase = cteRef.defaultDatabase
+				}
 
 				switch stmt := cteRef.ast.Stmt.(type) {
 				case *tree.Select:
@@ -1503,6 +1509,7 @@ func (builder *QueryBuilder) buildTable(stmt tree.TableExpr, ctx *BindContext) (
 
 				break
 			}
+			schema = ctx.defaultDatabase
 		}
 
 		obj, tableDef := builder.compCtx.Resolve(schema, table)
@@ -1510,6 +1517,65 @@ func (builder *QueryBuilder) buildTable(stmt tree.TableExpr, ctx *BindContext) (
 			return 0, errors.New("", fmt.Sprintf("table %q does not exist", table))
 		}
 
+		// set view statment to CTE
+		viewDefString := ""
+		for _, def := range tableDef.Defs {
+			if viewDef, ok := def.Def.(*plan.TableDef_DefType_View); ok {
+				viewDefString = viewDef.View.View
+				break
+			}
+		}
+		if viewDefString != "" {
+			if ctx.cteByName == nil {
+				ctx.cteByName = make(map[string]*CTERef)
+			}
+
+			viewData := ViewData{}
+			err := json.Unmarshal([]byte(viewDefString), &viewData)
+			if err != nil {
+				return 0, err
+			}
+
+			originStmts, err := mysql.Parse(viewData.Stmt)
+			if err != nil {
+				return 0, err
+			}
+			viewStmt, ok := originStmts[0].(*tree.CreateView)
+			if !ok {
+				return 0, errors.New("", "can not get view statement")
+			}
+
+			// when use db1.v1 in db2 context, if you use v1 as ViewName， that may conflict
+			viewName := fmt.Sprintf("%s.%s", viewData.DefaultDatabase, viewStmt.Name.ObjectName)
+			var maskedCTEs map[string]any
+			if len(ctx.cteByName) > 0 {
+				maskedCTEs := make(map[string]any)
+				for name := range ctx.cteByName {
+					maskedCTEs[name] = nil
+				}
+			}
+
+			ctx.cteByName[string(viewName)] = &CTERef{
+				ast: &tree.CTE{
+					Name: &tree.AliasClause{
+						Alias: tree.Identifier(viewName),
+						Cols:  viewStmt.ColNames,
+					},
+					Stmt: viewStmt.AsSource,
+				},
+				defaultDatabase: viewData.DefaultDatabase,
+				maskedCTEs:      maskedCTEs,
+			}
+
+			newTableName := tree.NewTableName(tree.Identifier(viewName), tree.ObjectNamePrefix{
+				CatalogName:     tbl.CatalogName, // TODO unused now, if used in some code, that will be save in view
+				SchemaName:      tree.Identifier(""),
+				ExplicitCatalog: false,
+				ExplicitSchema:  false,
+			})
+			return builder.buildTable(newTableName, ctx)
+		}
+
 		nodeID = builder.appendNode(&plan.Node{
 			NodeType:    plan.Node_TABLE_SCAN,
 			Cost:        builder.compCtx.Cost(obj, nil),

pkg/sql/plan/types.go

@@ -29,6 +29,7 @@ const (
 	JoinSideCorrelated      = 1 << iota
 )
 
+type TableDefType = plan.TableDef_DefType
 type TableDef = plan.TableDef
 type ColDef = plan.ColDef
 type ObjectRef = plan.ObjectRef
@@ -42,6 +43,11 @@ type Query = plan.Query
 type Plan = plan.Plan
 type Type = plan.Type
 type Plan_Query = plan.Plan_Query
+type Property = plan.Property
+type TableDef_DefType_Properties = plan.TableDef_DefType_Properties
+type TableDef_DefType_View = plan.TableDef_DefType_View
+type PropertiesDef = plan.PropertiesDef
+type ViewDef = plan.ViewDef
 
 type CompilerContext interface {
 	// Default database/schema in context
@@ -58,6 +64,8 @@ type CompilerContext interface {
 	GetHideKeyDef(dbName string, tableName string) *ColDef
 	// get estimated cost by table & expr
 	Cost(obj *ObjectRef, e *Expr) *Cost
+	// get origin sql string of the root
+	GetRootSql() string
 }
 
 type Optimizer interface {
@@ -77,6 +85,11 @@ type BaseOptimizer struct {
 	ctx   CompilerContext
 }
 
+type ViewData struct {
+	Stmt            string
+	DefaultDatabase string
+}
+
 type ExecType int
 
 const (
@@ -106,8 +119,9 @@ type QueryBuilder struct {
 }
 
 type CTERef struct {
-	ast        *tree.CTE
-	maskedCTEs map[string]any
+	defaultDatabase string
+	ast             *tree.CTE
+	maskedCTEs      map[string]any
 }
 
 type BindContext struct {
@@ -150,6 +164,8 @@ type BindContext struct {
 	parent     *BindContext
 	leftChild  *BindContext
 	rightChild *BindContext
+
+	defaultDatabase string
 }
 
 type NameTuple struct {

pkg/vm/engine/memEngine/engine.go

@@ -107,3 +107,7 @@ func (e *MemEngine) GetHideKeyDef(_ string, _ string) *plan.ColDef {
 func (e *MemEngine) Cost(_ *plan.ObjectRef, _ *plan.Expr) *plan.Cost {
 	return &plan.Cost{}
 }
+
+func (e *MemEngine) GetRootSql() string {
+	return ""
+}

pkg/vm/engine/tae/catalog/schema.go

@@ -126,6 +126,7 @@ type Schema struct {
 	BlockMaxRows     uint32
 	SegmentMaxBlocks uint16
 	Comment          string
+	View             string
 
 	SortKey    *SortKey
 	PhyAddrKey *ColDef
@@ -260,6 +261,10 @@ func (s *Schema) ReadFrom(r io.Reader) (n int64, err error) {
 		return
 	}
 	n += sn
+	if s.View, sn, err = common.ReadString(r); err != nil {
+		return
+	}
+	n += sn
 	colCnt := uint16(0)
 	if err = binary.Read(r, binary.BigEndian, &colCnt); err != nil {
 		return
@@ -336,6 +341,9 @@ func (s *Schema) Marshal() (buf []byte, err error) {
 	if _, err = common.WriteString(s.Comment, &w); err != nil {
 		return
 	}
+	if _, err = common.WriteString(s.View, &w); err != nil {
+		return
+	}
 	if err = binary.Write(&w, binary.BigEndian, uint16(len(s.ColDefs))); err != nil {
 		return
 	}

pkg/vm/engine/tae/moengine/engine_test.go

@@ -16,9 +16,10 @@ package moengine
 
 import (
 	"context"
-	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"testing"
 
+	"github.com/matrixorigin/matrixone/pkg/pb/plan"
+
 	mobat "github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"

pkg/vm/engine/tae/moengine/helper.go

@@ -16,9 +16,10 @@ package moengine
 
 import (
 	"fmt"
-	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"strings"
 
+	"github.com/matrixorigin/matrixone/pkg/pb/plan"
+
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/catalog"
 )
@@ -29,6 +30,12 @@ func SchemaToDefs(schema *catalog.Schema) (defs []engine.TableDef, err error) {
 		commentDef.Comment = schema.Comment
 		defs = append(defs, commentDef)
 	}
+
+	if schema.View != "" {
+		viewDef := new(engine.ViewDef)
+		viewDef.View = schema.View
+		defs = append(defs, viewDef)
+	}
 	for _, col := range schema.ColDefs {
 		if col.IsPhyAddr() {
 			continue
@@ -90,12 +97,18 @@ func DefsToSchema(name string, defs []engine.TableDef) (schema *catalog.Schema,
 					return
 				}
 			}
+
 		case *engine.PropertiesDef:
 			for _, property := range defVal.Properties {
-				if strings.ToLower(property.Key) == "comment" {
+				switch strings.ToLower(property.Key) {
+				case catalog.SystemRelAttr_Comment:
 					schema.Comment = property.Value
 				}
 			}
+
+		case *engine.ViewDef:
+			schema.View = defVal.View
+
 		default:
 			// We will not deal with other cases for the time being
 		}

pkg/vm/engine/txn/operations.go

@@ -58,6 +58,7 @@ func mustEncodePayload(o any) []byte {
 func init() {
 
 	// register TableDef types
+	gob.Register(new(engine.ViewDef))
 	gob.Register(new(engine.CommentDef))
 	gob.Register(new(engine.AttributeDef))
 	gob.Register(new(engine.IndexTableDef))

pkg/vm/engine/txn/test/execution.go

@@ -45,6 +45,10 @@ func (e *Execution) DefaultDatabase() string {
 	return "test"
 }
 
+func (e *Execution) GetRootSql() string {
+	return ""
+}
+
 func (e *Execution) GetHideKeyDef(dbName string, tableName string) *plan.ColDef {
 	attrs, err := e.getTableAttrs(dbName, tableName)
 	if err != nil {