implement select from view

Author: AndrewDi

executor/executor_test.go

@@ -3416,3 +3416,34 @@ func (s *testSuite) TestSelectHashPartitionTable(c *C) {
 		"  └─TableScan_13 10000.00 cop table:th, partition:, range:[-inf,+inf], keep order:false, stats:pseudo",
 	))
 }
+
+func (s *testSuite) TestSelectView(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("create table view_t (a int,b int)")
+	tk.MustExec("insert into view_t values(1,2)")
+	tk.MustExec("create view view1 as select * from view_t")
+	tk.MustExec("create view view2(c,d) as select * from view_t")
+	tk.MustExec("create view view3(c,d) as select a,b from view_t")
+	tk.MustQuery("select * from view1;").Check(testkit.Rows("1 2"))
+	tk.MustQuery("select * from view2;").Check(testkit.Rows("1 2"))
+	tk.MustQuery("select * from view3;").Check(testkit.Rows("1 2"))
+	tk.MustExec("drop table view_t;")
+	tk.MustExec("create table view_t(c int,d int)")
+	_, err := tk.Exec("select * from view1")
+	c.Assert(err.Error(), Equals, "[planner:1054]Unknown column 'a' in 'field_list'")
+	_, err = tk.Exec("select * from view2")
+	c.Assert(err.Error(), Equals, "[planner:1054]Unknown column 'a' in 'field_list'")
+	_, err = tk.Exec("select * from view3")
+	c.Assert(err.Error(), Equals, "[planner:1054]Unknown column 'a' in 'field list'")
+	tk.MustExec("drop table view_t;")
+	tk.MustExec("create table view_t(a int,b int,c int)")
+	tk.MustExec("insert into view_t values(1,2,3)")
+	_, err = tk.Exec("select * from view1")
+	c.Assert(err.Error(), Equals, "View view1's UnderTable has changed with different column count")
+	_, err = tk.Exec("select * from view2")
+	c.Assert(err.Error(), Equals, "View view2's UnderTable has changed with different column count")
+	tk.MustQuery("select * from view3;").Check(testkit.Rows("1 2"))
+	defer tk.MustExec("drop table view_t;")
+	defer tk.MustExec("drop view view1,view2,view3;")
+}

planner/core/logical_plan_builder.go

@@ -1901,6 +1901,38 @@ func (b *PlanBuilder) buildDataSource(tn *ast.TableName) (LogicalPlan, error) {
 
 	tableInfo := tbl.Meta()
 	b.visitInfo = appendVisitInfo(b.visitInfo, mysql.SelectPriv, dbName.L, tableInfo.Name.L, "")
+	charset, collation := b.ctx.GetSessionVars().GetCharsetInfo()
+	if tableInfo.IsView() {
+		var (
+			selectNode        ast.StmtNode
+			selectLogicalPlan Plan
+		)
+		selectNode, err = parser.New().ParseOneStmt(tableInfo.View.SelectStmt, charset, collation)
+		if err != nil {
+			return nil, err
+		}
+		selectLogicalPlan, err = b.Build(selectNode)
+		if err != nil {
+			return nil, err
+		}
+		var viewCols []*expression.Column
+		var selectSchema = selectLogicalPlan.Schema()
+		if len(selectSchema.Columns) != len(tableInfo.View.Cols) {
+			return nil, errors.New(fmt.Sprintf("View %s's UnderTable has changed with different column count", tableInfo.Name.L))
+		}
+		for i := range tableInfo.View.Cols {
+			if selectSchema.Columns[i].ColName != tableInfo.View.Cols[i] {
+				return nil, ErrUnknownColumn.GenWithStackByArgs(tableInfo.View.Cols[i].O, "field_list")
+			}
+			selectSchema.Columns[i].ColName = tableInfo.Cols()[i].Name
+			selectSchema.Columns[i].OrigColName = tableInfo.View.Cols[i]
+			viewCols = append(viewCols, selectSchema.Columns[i])
+		}
+		selectProjection := LogicalProjection{Exprs: expression.Column2Exprs(viewCols)}.Init(b.ctx)
+		selectProjection.SetChildren(selectLogicalPlan.(LogicalPlan))
+		selectProjection.SetSchema(selectLogicalPlan.Schema())
+		return selectProjection, nil
+	}
 
 	if tableInfo.GetPartitionInfo() != nil {
 		b.optFlag = b.optFlag | flagPartitionProcessor