release-1.0: only set DefaultValues in agg push down. (#5712)

executor/builder.go

@@ -497,6 +497,14 @@ func (b *executorBuilder) buildUnionScanExec(v *plan.PhysicalUnionScan) Executor
 // TODO: Refactor against different join strategies by extracting common code base
 func (b *executorBuilder) buildMergeJoin(v *plan.PhysicalMergeJoin) Executor {
 	joinBuilder := &joinBuilder{}
+	defaultValues := v.DefaultValues
+	if defaultValues == nil {
+		if v.JoinType == plan.LeftOuterJoin {
+			defaultValues = make([]types.Datum, v.Children()[1].Schema().Len())
+		} else if v.JoinType == plan.RightOuterJoin {
+			defaultValues = make([]types.Datum, v.Children()[0].Schema().Len())
+		}
+	}
 	exec, err := joinBuilder.Context(b.ctx).
 		LeftChild(b.build(v.Children()[0])).
 		RightChild(b.build(v.Children()[1])).
@@ -506,7 +514,7 @@ func (b *executorBuilder) buildMergeJoin(v *plan.PhysicalMergeJoin) Executor {
 		OtherFilter(v.OtherConditions).
 		Schema(v.Schema()).
 		JoinType(v.JoinType).
-		DefaultVals(v.DefaultValues).
+		DefaultVals(defaultValues).
 		BuildMergeJoin(v.Desc)
 
 	if err != nil {
@@ -530,28 +538,34 @@ func (b *executorBuilder) buildHashJoin(v *plan.PhysicalHashJoin) Executor {
 		rightHashKey = append(rightHashKey, rn)
 	}
 	e := &HashJoinExec{
-		schema:        v.Schema(),
-		otherFilter:   v.OtherConditions,
-		prepared:      false,
-		ctx:           b.ctx,
-		concurrency:   v.Concurrency,
-		defaultValues: v.DefaultValues,
+		schema:      v.Schema(),
+		otherFilter: v.OtherConditions,
+		prepared:    false,
+		ctx:         b.ctx,
+		concurrency: v.Concurrency,
+	}
+	e.defaultValues = v.DefaultValues
+	if v.JoinType == plan.LeftOuterJoin || v.JoinType == plan.RightOuterJoin {
+		e.outer = true
 	}
 	if v.SmallTable == 1 {
 		e.smallFilter = v.RightConditions
 		e.bigFilter = v.LeftConditions
 		e.smallHashKey = rightHashKey
 		e.bigHashKey = leftHashKey
 		e.leftSmall = false
+		if e.defaultValues == nil && e.outer {
+			e.defaultValues = make([]types.Datum, v.Children()[1].Schema().Len())
+		}
 	} else {
 		e.leftSmall = true
 		e.smallFilter = v.LeftConditions
 		e.bigFilter = v.RightConditions
 		e.smallHashKey = leftHashKey
 		e.bigHashKey = rightHashKey
-	}
-	if v.JoinType == plan.LeftOuterJoin || v.JoinType == plan.RightOuterJoin {
-		e.outer = true
+		if e.defaultValues == nil && e.outer {
+			e.defaultValues = make([]types.Datum, v.Children()[0].Schema().Len())
+		}
 	}
 	if e.leftSmall {
 		e.smallExec = b.build(v.Children()[0])
@@ -1129,9 +1143,17 @@ func (b *executorBuilder) buildIndexJoin(v *plan.PhysicalIndexJoin) Executor {
 	}
 	outerConditions := v.LeftConditions
 	innerConditions := v.RightConditions
+	defaultValues := v.DefaultValues
 	if v.OuterIndex == 1 {
 		outerConditions, innerConditions = innerConditions, outerConditions
 	}
+	if defaultValues == nil && v.Outer {
+		if v.OuterIndex == 0 {
+			defaultValues = make([]types.Datum, v.Children()[1].Schema().Len())
+		} else {
+			defaultValues = make([]types.Datum, v.Children()[0].Schema().Len())
+		}
+	}
 	return &IndexLookUpJoin{
 		baseExecutor:    newBaseExecutor(v.Schema(), b.ctx, b.build(v.Children()[v.OuterIndex])),
 		innerExec:       b.build(v.Children()[1-v.OuterIndex]).(DataReader),
@@ -1142,7 +1164,7 @@ func (b *executorBuilder) buildIndexJoin(v *plan.PhysicalIndexJoin) Executor {
 		outerConditions: outerConditions,
 		innerConditions: innerConditions,
 		otherConditions: v.OtherConditions,
-		defaultValues:   v.DefaultValues,
+		defaultValues:   defaultValues,
 		batchSize:       batchSize,
 	}
 }

executor/join_test.go

@@ -767,3 +767,13 @@ func (s *testSuite) TestIssue5255(c *C) {
 	tk.MustExec("insert into t2 values(1)")
 	tk.MustQuery("select /*+ TIDB_INLJ(t2) */ * from t1 join t2 on t1.a=t2.a").Check(testkit.Rows("1 2017-11-29 2.2 1"))
 }
+
+func (s *testSuite) TestIssue5278(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t, tt")
+	tk.MustExec("create table t(a int, b int)")
+	tk.MustExec("create table tt(a varchar(10), b int)")
+	tk.MustExec("insert into t values(1, 1)")
+	tk.MustQuery("select * from t left join tt on t.a=tt.a left join t ttt on t.a=ttt.a").Check(testkit.Rows("1 1 <nil> <nil> 1 1"))
+}

plan/column_pruning.go

@@ -19,7 +19,6 @@ import (
 	"github.com/pingcap/tidb/context"
 	"github.com/pingcap/tidb/expression"
 	"github.com/pingcap/tidb/model"
-	"github.com/pingcap/tidb/util/types"
 )
 
 type columnPruner struct {
@@ -262,12 +261,6 @@ func (p *LogicalJoin) PruneColumns(parentUsedCols []*expression.Column) {
 	rChild := p.children[1].(LogicalPlan)
 	lChild.PruneColumns(leftCols)
 	rChild.PruneColumns(rightCols)
-	// After column pruning, the size of schema may change, so we should also change the len of default value.
-	if p.JoinType == LeftOuterJoin {
-		p.DefaultValues = make([]types.Datum, p.children[1].Schema().Len())
-	} else if p.JoinType == RightOuterJoin {
-		p.DefaultValues = make([]types.Datum, p.children[0].Schema().Len())
-	}
 	p.mergeSchema()
 }
 

plan/eliminate_projection.go

@@ -18,7 +18,6 @@ import (
 	"github.com/pingcap/tidb/context"
 	"github.com/pingcap/tidb/expression"
 	"github.com/pingcap/tidb/terror"
-	"github.com/pingcap/tidb/util/types"
 )
 
 // canProjectionBeEliminatedLoose checks whether a projection can be eliminated, returns true if
@@ -161,9 +160,6 @@ func (pe *projectionEliminater) eliminate(p LogicalPlan, replace map[string]*exp
 	if !(isProj && canEliminate && canProjectionBeEliminatedLoose(proj)) {
 		return p
 	}
-	if join, ok := p.Parents()[0].(*LogicalJoin); ok {
-		pe.resetDefaultValues(join, p)
-	}
 	exprs := proj.Exprs
 	for i, col := range proj.Schema().Columns {
 		replace[string(col.HashCode())] = exprs[i].(*expression.Column)
@@ -173,43 +169,6 @@ func (pe *projectionEliminater) eliminate(p LogicalPlan, replace map[string]*exp
 	return p.Children()[0].(LogicalPlan)
 }
 
-// If the inner child of a Join is a Projection which been eliminated,
-// and the schema of child plan of Projection is not consistent with
-// the schema of Projection, the default values of Join should be reset.
-func (pe *projectionEliminater) resetDefaultValues(join *LogicalJoin, prj Plan) {
-	prjChild := prj.Children()[0]
-	var joinInnerChild Plan
-	switch join.JoinType {
-	case LeftOuterJoin:
-		joinInnerChild = join.Children()[1]
-	case RightOuterJoin:
-		joinInnerChild = join.Children()[0]
-	default:
-		return
-	}
-	if joinInnerChild != prj {
-		return
-	}
-	var schemaIdxMap map[int]int
-	prjSchema := prj.Schema().Columns
-	childOfPrjSchema := prjChild.Schema().Columns
-	for i := 0; i < len(prjSchema); i++ {
-		for j := 0; j < len(childOfPrjSchema); j++ {
-			if prjSchema[i].Equal(childOfPrjSchema[j], nil) {
-				schemaIdxMap[i] = j
-			}
-		}
-	}
-	newDefaultValues := make([]types.Datum, len(childOfPrjSchema))
-	for i := range prjSchema {
-		if j, ok := schemaIdxMap[i]; ok {
-			newDefaultValues[j] = join.DefaultValues[i]
-		}
-	}
-	join.DefaultValues = newDefaultValues
-	return
-}
-
 func (p *LogicalJoin) replaceExprColumns(replace map[string]*expression.Column) {
 	for _, equalExpr := range p.EqualConditions {
 		resolveExprAndReplace(equalExpr, replace)

plan/logical_plan_builder.go

@@ -287,10 +287,8 @@ func (b *planBuilder) buildJoin(join *ast.Join) LogicalPlan {
 	}
 	if join.Tp == ast.LeftJoin {
 		joinPlan.JoinType = LeftOuterJoin
-		joinPlan.DefaultValues = make([]types.Datum, rightPlan.Schema().Len())
 	} else if join.Tp == ast.RightJoin {
 		joinPlan.JoinType = RightOuterJoin
-		joinPlan.DefaultValues = make([]types.Datum, leftPlan.Schema().Len())
 	} else {
 		joinPlan.JoinType = InnerJoin
 	}
@@ -1370,9 +1368,6 @@ func (b *planBuilder) buildApplyWithJoinType(outerPlan, innerPlan LogicalPlan, t
 	b.optFlag = b.optFlag | flagBuildKeyInfo
 	b.optFlag = b.optFlag | flagDecorrelate
 	ap := LogicalApply{LogicalJoin: LogicalJoin{JoinType: tp}}.init(b.allocator, b.ctx)
-	if tp == LeftOuterJoin {
-		ap.DefaultValues = make([]types.Datum, innerPlan.Schema().Len())
-	}
 	setParentAndChildren(ap, outerPlan, innerPlan)
 	ap.SetSchema(expression.MergeSchema(outerPlan.Schema(), innerPlan.Schema()))
 	for i := outerPlan.Schema().Len(); i < ap.Schema().Len(); i++ {