[WIP] executor: inline projection for sort

executor/benchmark_test.go

@@ -467,6 +467,7 @@ func buildWindowExecutor(ctx sessionctx.Context, windowFunc string, funcs int, f
 			byItems = append(byItems, &core.ByItems{Expr: col, Desc: false})
 		}
 		sort := &core.PhysicalSort{ByItems: byItems}
+		sort.SetSchema(src.Schema().Clone())
 		sort.SetChildren(src)
 		win.SetChildren(sort)
 		tail = sort

executor/builder.go

@@ -1473,10 +1473,21 @@ func (b *executorBuilder) buildSort(v *plannercore.PhysicalSort) Executor {
 	if b.err != nil {
 		return nil
 	}
+	childrenUsedSchema := markChildrenUsedCols(v.Schema(), v.Children()[0].Schema())
+	var used []int
+	if childrenUsedSchema != nil {
+		used = make([]int, 0, len(childrenUsedSchema[0]))
+		for i, usedSchema := range childrenUsedSchema[0] {
+			if usedSchema {
+				used = append(used, i)
+			}
+		}
+	}
 	sortExec := SortExec{
 		baseExecutor: newBaseExecutor(b.ctx, v.Schema(), v.ExplainID(), childExec),
 		ByItems:      v.ByItems,
 		schema:       v.Schema(),
+		used:         used,
 	}
 	executorCounterSortExec.Inc()
 	return &sortExec

executor/sort.go

@@ -39,6 +39,11 @@ type SortExec struct {
 	Idx     int
 	fetched bool
 	schema  *expression.Schema
+	// used show which columns are used by father for child
+	// NOTE:
+	// 1. every columns are used if used is nil.
+	// 2. no columns are used if used is not nil but the size of used is 0.
+	used []int
 
 	keyExprs []expression.Expression
 	keyTypes []*types.FieldType
@@ -167,6 +172,13 @@ func (e *SortExec) Next(ctx context.Context, req *chunk.Chunk) error {
 func (e *SortExec) generatePartition() {
 	e.initPointers()
 	sort.Slice(e.rowPtrs, e.keyColumnsLess)
+	if e.used != nil {
+		numChunks := e.rowChunks.NumChunks()
+		for chkIdx := 0; chkIdx < numChunks; chkIdx++ {
+			rowChk := e.rowChunks.GetChunk(chkIdx)
+			rowChk.Prune(e.used)
+		}
+	}
 	e.partitionList = append(e.partitionList, e.rowChunks)
 	e.partitionRowPtrs = append(e.partitionRowPtrs, e.rowPtrs)
 }

planner/cascades/implementation_rules.go

@@ -310,6 +310,7 @@ func (r *ImplSort) OnImplement(expr *memo.GroupExpr, reqProp *property.PhysicalP
 		ls.SelectBlockOffset(),
 		&property.PhysicalProperty{ExpectedCnt: math.MaxFloat64},
 	)
+	ps.SetSchema(ls.Schema())
 	return []memo.Implementation{impl.NewSortImpl(ps)}, nil
 }
 

planner/core/exhaust_physical_plans.go

@@ -1747,6 +1747,7 @@ func (p *LogicalUnionAll) exhaustPhysicalPlans(prop *property.PhysicalProperty)
 
 func (ls *LogicalSort) getPhysicalSort(prop *property.PhysicalProperty) *PhysicalSort {
 	ps := PhysicalSort{ByItems: ls.ByItems}.Init(ls.ctx, ls.stats.ScaleByExpectCnt(prop.ExpectedCnt), ls.blockOffset, &property.PhysicalProperty{ExpectedCnt: math.MaxFloat64})
+	ps.SetSchema(ls.Schema().Clone())
 	return ps
 }
 
@@ -1757,6 +1758,7 @@ func (ls *LogicalSort) getNominalSort(reqProp *property.PhysicalProperty) *Nomin
 	}
 	prop.ExpectedCnt = reqProp.ExpectedCnt
 	ps := NominalSort{OnlyColumn: onlyColumn, ByItems: ls.ByItems}.Init(ls.ctx, ls.blockOffset, prop)
+	ps.SetSchema(ls.Schema().Clone())
 	return ps
 }
 

planner/core/logical_plan_builder.go

@@ -1224,6 +1224,8 @@ func (b *PlanBuilder) buildSort(ctx context.Context, p LogicalPlan, byItems []*a
 		exprs = append(exprs, &ByItems{Expr: it, Desc: item.Desc})
 	}
 	sort.ByItems = exprs
+	sort.SetSchema(p.Schema().Clone())
+	sort.names = p.OutputNames()
 	sort.SetChildren(p)
 	return sort, nil
 }

planner/core/logical_plans.go

@@ -872,7 +872,7 @@ type LogicalUnionAll struct {
 
 // LogicalSort stands for the order by plan.
 type LogicalSort struct {
-	baseLogicalPlan
+	logicalSchemaProducer
 
 	ByItems []*ByItems
 }

planner/core/physical_plans.go

@@ -486,15 +486,15 @@ type PhysicalStreamAgg struct {
 
 // PhysicalSort is the physical operator of sort, which implements a memory sort.
 type PhysicalSort struct {
-	basePhysicalPlan
+	physicalSchemaProducer
 
 	ByItems []*ByItems
 }
 
 // NominalSort asks sort properties for its child. It is a fake operator that will not
 // appear in final physical operator tree. It will be eliminated or converted to Projection.
 type NominalSort struct {
-	basePhysicalPlan
+	physicalSchemaProducer
 
 	// These two fields are used to switch ScalarFunctions to Constants. For these
 	// NominalSorts, we need to converted to Projections check if the ScalarFunctions

planner/core/planbuilder.go

@@ -1689,6 +1689,7 @@ func (b *PlanBuilder) buildShow(ctx context.Context, show *ast.ShowStmt) (Plan,
 		sort := LogicalSort{
 			ByItems: []*ByItems{{Expr: orderByCol}},
 		}.Init(b.ctx, b.getSelectOffset())
+		sort.SetSchema(np.Schema().Clone())
 		sort.SetChildren(np)
 		np = sort
 	}

planner/core/resolve_indices.go

@@ -355,7 +355,7 @@ func (p *basePhysicalAgg) ResolveIndices() (err error) {
 
 // ResolveIndices implements Plan interface.
 func (p *PhysicalSort) ResolveIndices() (err error) {
-	err = p.basePhysicalPlan.ResolveIndices()
+	err = p.physicalSchemaProducer.ResolveIndices()
 	if err != nil {
 		return err
 	}

planner/core/rule_column_pruning.go

@@ -148,6 +148,7 @@ func (ls *LogicalSort) PruneColumns(parentUsedCols []*expression.Column) error {
 			parentUsedCols = append(parentUsedCols, cols...)
 		}
 	}
+	ls.inlineProjection(parentUsedCols)
 	return child.PruneColumns(parentUsedCols)
 }
 

planner/core/rule_max_min_eliminate.go

@@ -176,6 +176,7 @@ func (a *maxMinEliminator) eliminateSingleMaxMin(agg *LogicalAggregation) *Logic
 		// Compose Sort operator.
 		sort := LogicalSort{}.Init(ctx, agg.blockOffset)
 		sort.ByItems = append(sort.ByItems, &ByItems{f.Args[0], desc})
+		sort.SetSchema(child.Schema().Clone())
 		sort.SetChildren(child)
 		child = sort
 	}