plan, executor: no longer treat dml as logical/physical plan.

executor/builder.go

@@ -313,9 +313,7 @@ func (b *executorBuilder) buildInsert(v *plan.Insert) Executor {
 		GenExprs:              v.GenCols.Exprs,
 		needFillDefaultValues: v.NeedFillDefaultValue,
 	}
-	if len(v.Children()) > 0 {
-		ivs.SelectExec = b.build(v.Children()[0])
-	}
+	ivs.SelectExec = b.build(v.SelectPlan)
 	ivs.Table = v.Table
 	if v.IsReplace {
 		return b.buildReplace(ivs)
@@ -778,12 +776,12 @@ func (b *executorBuilder) buildUnion(v *plan.Union) Executor {
 
 func (b *executorBuilder) buildUpdate(v *plan.Update) Executor {
 	tblID2table := make(map[int64]table.Table)
-	for id := range v.Schema().TblID2Handle {
+	for id := range v.SelectPlan.Schema().TblID2Handle {
 		tblID2table[id], _ = b.is.TableByID(id)
 	}
 	return &UpdateExec{
 		baseExecutor: newBaseExecutor(nil, b.ctx),
-		SelectExec:   b.build(v.Children()[0]),
+		SelectExec:   b.build(v.SelectPlan),
 		OrderedList:  v.OrderedList,
 		tblID2table:  tblID2table,
 		IgnoreErr:    v.IgnoreErr,
@@ -792,12 +790,12 @@ func (b *executorBuilder) buildUpdate(v *plan.Update) Executor {
 
 func (b *executorBuilder) buildDelete(v *plan.Delete) Executor {
 	tblID2table := make(map[int64]table.Table)
-	for id := range v.Schema().TblID2Handle {
+	for id := range v.SelectPlan.Schema().TblID2Handle {
 		tblID2table[id], _ = b.is.TableByID(id)
 	}
 	return &DeleteExec{
 		baseExecutor: newBaseExecutor(nil, b.ctx),
-		SelectExec:   b.build(v.Children()[0]),
+		SelectExec:   b.build(v.SelectPlan),
 		Tables:       v.Tables,
 		IsMultiTable: v.IsMultiTable,
 		tblID2Table:  tblID2table,

executor/metrics.go

@@ -256,6 +256,14 @@ func (pa *stmtAttributes) fromPlan(p plan.Plan) {
 		pa.setIsSystemTable(x.DBName)
 	case *plan.PhysicalHashSemiJoin:
 		pa.hasJoin = true
+	case *plan.Insert:
+		if x.SelectPlan != nil {
+			pa.fromPlan(x.SelectPlan)
+		}
+	case *plan.Delete:
+		pa.fromPlan(x.SelectPlan)
+	case *plan.Update:
+		pa.fromPlan(x.SelectPlan)
 	case *plan.PhysicalTableReader:
 		for _, child := range x.TablePlans {
 			pa.fromPlan(child)

plan/column_pruning.go

@@ -198,15 +198,6 @@ func (p *Exists) PruneColumns(parentUsedCols []*expression.Column) {
 	p.children[0].(LogicalPlan).PruneColumns(nil)
 }
 
-// PruneColumns implements LogicalPlan interface.
-func (p *Insert) PruneColumns(_ []*expression.Column) {
-	if len(p.Children()) == 0 {
-		return
-	}
-	child := p.children[0].(LogicalPlan)
-	child.PruneColumns(child.Schema().Columns)
-}
-
 func (p *LogicalJoin) extractUsedCols(parentUsedCols []*expression.Column) (leftCols []*expression.Column, rightCols []*expression.Column) {
 	for _, eqCond := range p.EqualConditions {
 		parentUsedCols = append(parentUsedCols, expression.ExtractColumns(eqCond)...)
@@ -288,13 +279,3 @@ func (p *SelectLock) PruneColumns(parentUsedCols []*expression.Column) {
 		p.children[0].(LogicalPlan).PruneColumns(parentUsedCols)
 	}
 }
-
-// PruneColumns implements LogicalPlan interface.
-func (p *Update) PruneColumns(parentUsedCols []*expression.Column) {
-	p.baseLogicalPlan.PruneColumns(p.children[0].Schema().Columns)
-}
-
-// PruneColumns implements LogicalPlan interface.
-func (p *Delete) PruneColumns(parentUsedCols []*expression.Column) {
-	p.baseLogicalPlan.PruneColumns(p.children[0].Schema().Columns)
-}

plan/dag_plan_test.go

@@ -519,33 +519,33 @@ func (s *testPlanSuite) TestDAGPlanBuilderBasePhysicalPlan(c *C) {
 		// Test complex update.
 		{
 			sql:  "update t set a = 5 where b < 1 order by d limit 1",
-			best: "TableReader(Table(t)->Sel([lt(test.t.b, 1)])->TopN([test.t.d],0,1))->TopN([test.t.d],0,1)->*plan.Update",
+			best: "TableReader(Table(t)->Sel([lt(test.t.b, 1)])->TopN([test.t.d],0,1))->TopN([test.t.d],0,1)->Update",
 		},
 		// Test simple update.
 		{
 			sql:  "update t set a = 5",
-			best: "TableReader(Table(t))->*plan.Update",
+			best: "TableReader(Table(t))->Update",
 		},
 		// TODO: Test delete/update with join.
 		// Test complex delete.
 		{
 			sql:  "delete from t where b < 1 order by d limit 1",
-			best: "TableReader(Table(t)->Sel([lt(test.t.b, 1)])->TopN([test.t.d],0,1))->TopN([test.t.d],0,1)->*plan.Delete",
+			best: "TableReader(Table(t)->Sel([lt(test.t.b, 1)])->TopN([test.t.d],0,1))->TopN([test.t.d],0,1)->Delete",
 		},
 		// Test simple delete.
 		{
 			sql:  "delete from t",
-			best: "TableReader(Table(t))->*plan.Delete",
+			best: "TableReader(Table(t))->Delete",
 		},
 		// Test complex insert.
 		{
 			sql:  "insert into t select * from t where b < 1 order by d limit 1",
-			best: "TableReader(Table(t)->Sel([lt(test.t.b, 1)])->TopN([test.t.d],0,1))->TopN([test.t.d],0,1)->*plan.Insert",
+			best: "TableReader(Table(t)->Sel([lt(test.t.b, 1)])->TopN([test.t.d],0,1))->TopN([test.t.d],0,1)->Insert",
 		},
 		// Test simple insert.
 		{
 			sql:  "insert into t values(0,0,0,0,0,0,0)",
-			best: "*plan.Insert",
+			best: "Insert",
 		},
 		// Test dual.
 		{

plan/eliminate_projection.go

@@ -129,7 +129,7 @@ func (pe *projectionEliminater) eliminate(p LogicalPlan, replace map[string]*exp
 	setParentAndChildren(p, children...)
 
 	switch p.(type) {
-	case *Sort, *TopN, *Limit, *Selection, *MaxOneRow, *Update, *SelectLock:
+	case *Sort, *TopN, *Limit, *Selection, *MaxOneRow, *SelectLock:
 		p.SetSchema(p.Children()[0].Schema())
 	case *LogicalJoin, *LogicalApply:
 		var joinTp JoinType

plan/initialize.go

@@ -179,23 +179,17 @@ func (p MaxOneRow) init(ctx context.Context) *MaxOneRow {
 }
 
 func (p Update) init(ctx context.Context) *Update {
-	p.basePlan = newBasePlan(TypeUpate, ctx, &p)
-	p.baseLogicalPlan = newBaseLogicalPlan(p.basePlan)
-	p.basePhysicalPlan = newBasePhysicalPlan(p.basePlan)
+	p.basePlan = *newBasePlan(TypeUpate, ctx, &p)
 	return &p
 }
 
 func (p Delete) init(ctx context.Context) *Delete {
-	p.basePlan = newBasePlan(TypeDelete, ctx, &p)
-	p.baseLogicalPlan = newBaseLogicalPlan(p.basePlan)
-	p.basePhysicalPlan = newBasePhysicalPlan(p.basePlan)
+	p.basePlan = *newBasePlan(TypeDelete, ctx, &p)
 	return &p
 }
 
 func (p Insert) init(ctx context.Context) *Insert {
-	p.basePlan = newBasePlan(TypeInsert, ctx, &p)
-	p.baseLogicalPlan = newBaseLogicalPlan(p.basePlan)
-	p.basePhysicalPlan = newBasePhysicalPlan(p.basePlan)
+	p.basePlan = *newBasePlan(TypeInsert, ctx, &p)
 	return &p
 }
 

plan/logical_plan_builder.go

@@ -1781,7 +1781,7 @@ func (b *planBuilder) buildSemiJoin(outerPlan, innerPlan LogicalPlan, onConditio
 	return joinPlan
 }
 
-func (b *planBuilder) buildUpdate(update *ast.UpdateStmt) LogicalPlan {
+func (b *planBuilder) buildUpdate(update *ast.UpdateStmt) Plan {
 	b.inUpdateStmt = true
 	sel := &ast.SelectStmt{Fields: &ast.FieldList{}, From: update.TableRefs, Where: update.Where, OrderBy: update.Order, Limit: update.Limit}
 	p := b.buildResultSetNode(sel.From.TableRefs)
@@ -1827,8 +1827,8 @@ func (b *planBuilder) buildUpdate(update *ast.UpdateStmt) LogicalPlan {
 		OrderedList: orderedList,
 		IgnoreErr:   update.IgnoreErr,
 	}.init(b.ctx)
-	setParentAndChildren(updt, p)
 	updt.SetSchema(p.Schema())
+	updt.SelectPlan, b.err = doOptimize(b.optFlag, p, b.ctx)
 	return updt
 }
 
@@ -1943,7 +1943,7 @@ func extractTableAsNameForUpdate(p Plan, asNames map[*model.TableInfo][]*model.C
 	}
 }
 
-func (b *planBuilder) buildDelete(delete *ast.DeleteStmt) LogicalPlan {
+func (b *planBuilder) buildDelete(delete *ast.DeleteStmt) Plan {
 	sel := &ast.SelectStmt{Fields: &ast.FieldList{}, From: delete.TableRefs, Where: delete.Where, OrderBy: delete.Order, Limit: delete.Limit}
 	p := b.buildResultSetNode(sel.From.TableRefs)
 	if b.err != nil {
@@ -1978,7 +1978,10 @@ func (b *planBuilder) buildDelete(delete *ast.DeleteStmt) LogicalPlan {
 		Tables:       tables,
 		IsMultiTable: delete.IsMultiTable,
 	}.init(b.ctx)
-	setParentAndChildren(del, p)
+	del.SelectPlan, b.err = doOptimize(b.optFlag, p, b.ctx)
+	if b.err != nil {
+		return nil
+	}
 	del.SetSchema(expression.NewSchema())
 
 	var tableList []*ast.TableName

plan/logical_plan_test.go

@@ -612,19 +612,19 @@ func (s *testPlanSuite) TestPlanBuilder(c *C) {
 		},
 		{
 			sql:  "update t set t.a = t.a * 1.5 where t.a >= 1000 order by t.a desc limit 10",
-			plan: "DataScan(t)->Sel([ge(test.t.a, 1000)])->Sort->Limit->*plan.Update",
+			plan: "TableReader(Table(t)->Limit)->Limit->Update",
 		},
 		{
 			sql:  "delete from t where t.a >= 1000 order by t.a desc limit 10",
-			plan: "DataScan(t)->Sel([ge(test.t.a, 1000)])->Sort->Limit->*plan.Delete",
+			plan: "TableReader(Table(t)->Limit)->Limit->Delete",
 		},
 		{
 			sql:  "explain select * from t union all select * from t limit 1, 1",
 			plan: "*plan.Explain",
 		},
 		{
 			sql:  "insert into t select * from t",
-			plan: "DataScan(t)->Projection->*plan.Insert",
+			plan: "TableReader(Table(t))->Insert",
 		},
 		{
 			sql:  "show columns from t where `Key` = 'pri' like 't*'",

plan/logical_plans.go

@@ -35,12 +35,9 @@ var (
 	_ LogicalPlan = &DataSource{}
 	_ LogicalPlan = &Union{}
 	_ LogicalPlan = &Sort{}
-	_ LogicalPlan = &Update{}
-	_ LogicalPlan = &Delete{}
 	_ LogicalPlan = &SelectLock{}
 	_ LogicalPlan = &Limit{}
 	_ LogicalPlan = &Show{}
-	_ LogicalPlan = &Insert{}
 )
 
 // JoinType contains CrossJoin, InnerJoin, LeftOuterJoin, RightOuterJoin, FullOuterJoin, SemiJoin.
@@ -377,23 +374,3 @@ type Limit struct {
 
 	expectedProp *requiredProp
 }
-
-// Update represents Update plan.
-type Update struct {
-	*basePlan
-	baseLogicalPlan
-	basePhysicalPlan
-
-	OrderedList []*expression.Assignment
-	IgnoreErr   bool
-}
-
-// Delete represents a delete plan.
-type Delete struct {
-	*basePlan
-	baseLogicalPlan
-	basePhysicalPlan
-
-	Tables       []*ast.TableName
-	IsMultiTable bool
-}

plan/optimizer.go

@@ -79,6 +79,7 @@ func Optimize(ctx context.Context, node ast.Node, is infoschema.InfoSchema) (Pla
 	if logic, ok := p.(LogicalPlan); ok {
 		return doOptimize(builder.optFlag, logic, ctx)
 	}
+	p.ResolveIndices()
 	if execPlan, ok := p.(*Execute); ok {
 		err := execPlan.optimizePreparedPlan(ctx, is)
 		return p, errors.Trace(err)

plan/physical_plans.go

@@ -31,12 +31,9 @@ var (
 	_ PhysicalPlan = &TableDual{}
 	_ PhysicalPlan = &Union{}
 	_ PhysicalPlan = &Sort{}
-	_ PhysicalPlan = &Update{}
-	_ PhysicalPlan = &Delete{}
 	_ PhysicalPlan = &SelectLock{}
 	_ PhysicalPlan = &Limit{}
 	_ PhysicalPlan = &Show{}
-	_ PhysicalPlan = &Insert{}
 	_ PhysicalPlan = &PhysicalIndexScan{}
 	_ PhysicalPlan = &PhysicalTableScan{}
 	_ PhysicalPlan = &PhysicalTableReader{}
@@ -423,15 +420,6 @@ func (p *MaxOneRow) Copy() PhysicalPlan {
 	return &np
 }
 
-// Copy implements the PhysicalPlan Copy interface.
-func (p *Insert) Copy() PhysicalPlan {
-	np := *p
-	np.basePlan = p.basePlan.copy()
-	np.baseLogicalPlan = newBaseLogicalPlan(np.basePlan)
-	np.basePhysicalPlan = newBasePhysicalPlan(np.basePlan)
-	return &np
-}
-
 // Copy implements the PhysicalPlan Copy interface.
 func (p *Limit) Copy() PhysicalPlan {
 	np := *p
@@ -494,24 +482,6 @@ func (p *PhysicalAggregation) Copy() PhysicalPlan {
 	return &np
 }
 
-// Copy implements the PhysicalPlan Copy interface.
-func (p *Update) Copy() PhysicalPlan {
-	np := *p
-	np.basePlan = p.basePlan.copy()
-	np.baseLogicalPlan = newBaseLogicalPlan(np.basePlan)
-	np.basePhysicalPlan = newBasePhysicalPlan(np.basePlan)
-	return &np
-}
-
-// Copy implements the PhysicalPlan Copy interface.
-func (p *Delete) Copy() PhysicalPlan {
-	np := *p
-	np.basePlan = p.basePlan.copy()
-	np.baseLogicalPlan = newBaseLogicalPlan(np.basePlan)
-	np.basePhysicalPlan = newBasePhysicalPlan(np.basePlan)
-	return &np
-}
-
 // Copy implements the PhysicalPlan Copy interface.
 func (p *Show) Copy() PhysicalPlan {
 	np := *p

plan/plan.go

@@ -52,6 +52,9 @@ type Plan interface {
 	// findColumn finds the column in basePlan's schema.
 	// If the column is not in the schema, returns error.
 	findColumn(*ast.ColumnName) (*expression.Column, int, error)
+
+	// ResolveIndices resolves the indices for columns. After doing this, the columns can evaluate the rows by their indices.
+	ResolveIndices()
 }
 
 // taskType is the type of execution task.
@@ -197,9 +200,6 @@ type PhysicalPlan interface {
 
 	// statsProfile will return the stats for this plan.
 	statsProfile() *statsProfile
-
-	// ResolveIndices resolves the indices for columns. After doing this, the columns can evaluate the rows by their indices.
-	ResolveIndices()
 }
 
 type baseLogicalPlan struct {

plan/planbuilder.go

@@ -920,7 +920,10 @@ func (b *planBuilder) buildInsert(insert *ast.InsertStmt) Plan {
 				return nil
 			}
 		}
-		setParentAndChildren(insertPlan, selectPlan)
+		insertPlan.SelectPlan, b.err = doOptimize(b.optFlag, selectPlan.(LogicalPlan), b.ctx)
+		if b.err != nil {
+			return nil
+		}
 	}
 
 	// Calculate generated columns.

plan/plans.go

@@ -205,9 +205,7 @@ type InsertGeneratedColumns struct {
 
 // Insert represents an insert plan.
 type Insert struct {
-	*basePlan
-	baseLogicalPlan
-	basePhysicalPlan
+	basePlan
 
 	Table       table.Table
 	tableSchema *expression.Schema
@@ -224,6 +222,28 @@ type Insert struct {
 	NeedFillDefaultValue bool
 
 	GenCols InsertGeneratedColumns
+
+	SelectPlan PhysicalPlan
+}
+
+// Update represents Update plan.
+type Update struct {
+	basePlan
+
+	OrderedList []*expression.Assignment
+	IgnoreErr   bool
+
+	SelectPlan PhysicalPlan
+}
+
+// Delete represents a delete plan.
+type Delete struct {
+	basePlan
+
+	Tables       []*ast.TableName
+	IsMultiTable bool
+
+	SelectPlan PhysicalPlan
 }
 
 // AnalyzeColumnsTask is used for analyze columns.