planner/core: support union all for mpp. #24287

hanfei1991 · 2021-04-26T07:13:35Z

What problem does this PR solve?

Problem Summary:
Support union all operator push down

What is changed and how it works?

Proposal: proposal of union all

What's Changed:
multiplying tasks during buidling mpp tasks.

Related changes

Check List

Tests

Unit test
Manual test (add detailed scripts or steps below)

Release note

planner/core: support union all for mpp.

eurekaka · 2021-05-17T06:47:46Z

planner/core/exhaust_physical_plans.go

 	}
-	ua := PhysicalUnionAll{}.Init(p.ctx, p.stats.ScaleByExpectCnt(prop.ExpectedCnt), p.blockOffset, chReqProps...)
+	ua := PhysicalUnionAll{
+		mpp: prop.TaskTp == property.MppTaskType,


should be canUseMpp && prop.TaskTp == property.MppTaskType?

That's right, I will fix it

eurekaka · 2021-05-17T07:09:49Z

planner/core/fragment.go

+		}
+		*forest = append(*forest, p.(*PhysicalExchangeSender))
+		for i := 1; i < len(stack); i++ {
+			if _, ok := stack[i].(*PhysicalUnionAll); ok {


Do we need to adjust the columns below and above this removed UnionAll to keep same UniqueID? or replace this UnionAll with a Projection?

I think it's unnecessary, because TiFlash won't use UniqueID.

fzhedu · 2021-05-17T08:21:02Z

planner/core/exhaust_physical_plans.go

+		if canUseMpp && prop.TaskTp == property.MppTaskType {
+			chReqProps = append(chReqProps, &property.PhysicalProperty{
+				ExpectedCnt: prop.ExpectedCnt,
+				TaskTp:      property.MppTaskType,
+			})
+		} else {
+			chReqProps = append(chReqProps, &property.PhysicalProperty{ExpectedCnt: prop.ExpectedCnt})
+		}
 	}
-	ua := PhysicalUnionAll{}.Init(p.ctx, p.stats.ScaleByExpectCnt(prop.ExpectedCnt), p.blockOffset, chReqProps...)
+	ua := PhysicalUnionAll{
+		mpp: prop.TaskTp == property.MppTaskType,
+	}.Init(p.ctx, p.stats.ScaleByExpectCnt(prop.ExpectedCnt), p.blockOffset, chReqProps...)
 	ua.SetSchema(p.Schema())
+	if canUseMpp && prop.TaskTp == property.RootTaskType {
+		chReqProps = make([]*property.PhysicalProperty, 0, len(p.children))
+		for range p.children {
+			chReqProps = append(chReqProps, &property.PhysicalProperty{
+				ExpectedCnt: prop.ExpectedCnt,
+				TaskTp:      property.MppTaskType,
+			})
+		}
+		mppUA := PhysicalUnionAll{mpp: true}.Init(p.ctx, p.stats.ScaleByExpectCnt(prop.ExpectedCnt), p.blockOffset, chReqProps...)
+		mppUA.SetSchema(p.Schema())
+		return []PhysicalPlan{ua, mppUA}, true
+	}


it is better to clear which case will generate mpp task, and which cases will generate root task. This will avoid some errors like pointed above by kenan.

if(canUseMpp && (prop.TaskTp == property.MppTaskType|| prop.TaskTp == property.RootTaskType)) { generate mpp task } if (!(canUseMpp && prop.TaskTp == property.MppTaskType)) { generate root task } ...

It's a little troublesome and confusing by using a negative condition (meaning condition with "!").
Although it aligns code beautifully, I cannot figure out the meanings clearly by only reading the conditions.

fzhedu · 2021-05-17T08:32:31Z

planner/core/fragment.go

-	s.Fragment = f
-	err := f.init(s)
-	return f, errors.Trace(err)
+func flattenPlanForUnionAll(stack []PhysicalPlan, forest *[]*PhysicalExchangeSender) error {


pls add some comments

fzhedu · 2021-05-17T08:49:44Z

planner/core/task.go

 func (p *PhysicalUnionAll) attach2Task(tasks ...task) task {
+	if _, ok := tasks[0].(*mppTask); ok {


even if their children are mpp tasks, the unionAll may also return root task

in what case?

in TiDBMPP mode, union all is at tidb, then its children run in MPP.

fzhedu · 2021-05-17T08:58:46Z

planner/core/rule_inject_extra_projection.go

+			srcCol := ch.Schema().Columns[i]
+			srcType := srcCol.RetType
+			if !srcType.Equal(dstType) || !(mysql.HasNotNullFlag(dstType.Flag) == mysql.HasNotNullFlag(srcType.Flag)) {
+				exprs[i] = expression.BuildCastFunction4Union(un.ctx, srcCol, dstType)


we should do cast like in the join, such as decima wich different precision and (uint8 = int 32), besides, set the null property for these cast functions.

When comparing types, It will take "length" and "decimal" of types into consideration.

fzhedu · 2021-05-17T09:52:43Z

planner/core/fragment.go

+			p = ch
+		}
+	case *PhysicalHashJoin:
+		stack = append(stack, x.children[1-x.InnerChildIdx])


x.children[1-x.InnerChildIdx] means probe side?

what about other side where union may also occur?

yes

the other side can never have a union-all.

fzhedu · 2021-05-17T09:58:14Z

executor/tiflash_test.go

+	tk.MustQuery("select x1.a, x.a from x1 left join (select x2.b a, x1.b from x1 join x2 on x1.a = x2.b union all select * from x1 ) x on x1.a = x.a order by x1.a").Check(testkit.Rows("1 1", "1 1", "2 2", "2 2", "3 3", "3 3", "4 4", "4 4"))
+	tk.MustQuery("select x1.a, x.a from x1 left join (select count(*) a, sum(b) b from x1 group by a union all select * from x2 ) x on x1.a = x.a order by x1.a").Check(testkit.Rows("1 1", "1 1", "1 1", "1 1", "2 2", "3 3", "4 4"))
+
+	tk.MustExec("drop table if exists x3")
+	tk.MustExec("create table x3(a int , b int);")
+	tk.MustExec("alter table x3 set tiflash replica 1")
+	tb = testGetTableByName(c, tk.Se, "test", "x3")
+	err = domain.GetDomain(tk.Se).DDL().UpdateTableReplicaInfo(tk.Se, tb.Meta().ID, true)
+	c.Assert(err, IsNil)
+
+	tk.MustExec("insert into x3 values (2, 2), (2, 3), (2, 4)")
+	// test nested union all
+	tk.MustQuery("select count(*) from (select a, b from x1 union all select a, b from x3 union all (select x1.a, x3.b from (select * from x3 union all select * from x2) x3 left join x1 on x3.a = x1.b))").Check(testkit.Rows("14"))
+	// test union all join union all
+	tk.MustQuery("select count(*) from (select * from x1 union all select * from x2 union all select * from x3) x join (select * from x1 union all select * from x2 union all select * from x3) y on x.a = y.b").Check(testkit.Rows("29"))
+	tk.MustExec("set @@session.tidb_broadcast_join_threshold_count=100000")
+	failpoint.Enable("github.com/pingcap/tidb/executor/checkTotalMPPTasks", `return(6)`)
+	tk.MustQuery("select count(*) from (select * from x1 union all select * from x2 union all select * from x3) x join (select * from x1 union all select * from x2 union all select * from x3) y on x.a = y.b").Check(testkit.Rows("29"))
+	failpoint.Disable("github.com/pingcap/tidb/executor/checkTotalMPPTasks")
+
+}


please also check plans

genereate more complex queries, like agg, agg+join, partition key is a subset of join/group by keys.

use randgen to test more complex cases.

We have checked plan results in plan tests. I agree with adding more tests but It needs time.

can the final plan be printed? as it is important to debug according to explain [analyze].

The plan that has eliminated union-all can not be showed.

can it support explain analyze now?

fzhedu · 2021-05-17T11:07:27Z

planner/core/exhaust_physical_plans.go

 		return nil, true
 	}
+	canUseMpp := p.ctx.GetSessionVars().AllowMPPExecution && p.canPushToCop(kv.TiFlash)


this function does not deliver the partition properity, which introduce more network overhead during exchange.
It is better to deliver the partition properity, then the unionAll becomes a exchange reciver, and set different exchange sender such as pashthrough or partition. for example:
group by R.a
|
(group by S.a, S.b union scan C) as R

if use the current code, the plan becomes:

group by R.a
|
exchange receiver
|----exchange sender1[partition by S.a] - group by S.a, S.b[partition by S.a,S.B]
|----exchange sender2[partition by C.a] - Scan C

the optimal one is:

group by R.a
|
exchange receiver
|----exchange sender1[pashthrough] - group by S.a, S.b[partition by S.a]
|----exchange sender2[partition by C.a] - Scan C

here exchange sender1 will send to local node and reduce network IO.

That's too complicated and only is beneficial in a few situations.

eliminating shuffle is an important optimization for distributed plans. If the users' sql is slow due to more unnecessary exchange, how to support this case in the future?
besides, it is not too complicated, its main idea is:

delivery data distribution properity to its child,

push up unionAll until the data distribution properity does not satisfy needs, then

enforce exchange.

Chould you detail the pass through exchanger? How could we granrantee the same task number between source tasks and target tasks?

planner/core/fragment.go

lzmhhh123 · 2021-05-18T09:00:01Z

planner/core/rule_inject_extra_projection.go

 	}
 	return plan
 }

+func injectProjBelowUnion(un *PhysicalUnionAll) *PhysicalUnionAll {


Is it necessary? In the logical plan building stage, field cast projection has already been appended.

It is! The check in logical plan is looser, such as lack of "NotNull" flag check.

Co-authored-by: Zhuomin Liu <lzmhhh123@gmail.com>

ti-chi-bot · 2021-06-01T07:03:27Z

@eurekaka: /merge is only allowed for the committers, you can assign this pull request to the committer in list by filling /assign @committer in the comment to help merge this pull request.

In response to this:

/merge

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the ti-community-infra/tichi repository.

eurekaka · 2021-06-01T07:04:11Z

/lgtm

ti-chi-bot · 2021-06-01T07:04:14Z

@eurekaka: Please use GitHub review feature instead of /lgtm [cancel] when you want to submit review to the pull request.
For how to use GitHub review feature, see also this document provided by GitHub.

For the reason we drop support to the commands, see also this page.
This reply is being used as a temporary reply during the migration of review process and will be removed on July 1st.

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the ti-community-infra/tichi repository.

ti-chi-bot · 2021-06-01T07:04:43Z

[REVIEW NOTIFICATION]

This pull request has been approved by:

eurekaka
lzmhhh123

To complete the pull request process, please ask the reviewers in the list to review by filling /cc @reviewer in the comment.
After your PR has acquired the required number of LGTMs, you can assign this pull request to the committer in the list by filling /assign @committer in the comment to help you merge this pull request.

The full list of commands accepted by this bot can be found here.

Reviewer can indicate their review by submitting an approval review.
Reviewer can cancel approval by submitting a request changes review.

eurekaka · 2021-06-01T07:05:01Z

/merge

ti-chi-bot · 2021-06-01T07:05:05Z

This pull request has been accepted and is ready to merge.

Commit hash: 9a7b2a7

fzhedu · 2021-06-01T11:51:52Z

planner/core/fragment.go

+// after untwist, there will be two plans in `forest` slice:
+// - ExchangeSender -> Projection (c1) -> TableScan(t)
+// - ExchangeSender -> Projection (c2) -> TableScan(s)
+func untwistPlanAndRemoveUnionAll(stack []PhysicalPlan, forest *[]*PhysicalExchangeSender) error {


can this change be seen by explain?

No, this can only be seen in the log.

hanfei1991 · 2021-06-02T06:01:53Z

/run-all-tests

Signed-off-by: ti-srebot <ti-srebot@pingcap.com>

ti-srebot · 2021-06-02T06:16:11Z

cherry pick to release-5.0 in PR #25051

hanfei1991 added 5 commits April 25, 2021 13:59

planner: support union all

93eb074

fix test

fc24eb7

add some test

1d3582e

add some test

14229f9

add some tests

ebcff6d

hanfei1991 requested review from a team as code owners April 26, 2021 07:13

hanfei1991 requested review from XuHuaiyu and removed request for a team April 26, 2021 07:13

ti-chi-bot added the size/XL Denotes a PR that changes 500-999 lines, ignoring generated files. label Apr 26, 2021

github-actions bot added component/expression sig/execution SIG execution labels Apr 26, 2021

hanfei1991 added 2 commits April 27, 2021 12:29

fix a type bug

e38d895

fix go fmt

859273b

hanfei1991 force-pushed the hanfei/union-all branch from ec3593c to 859273b Compare April 28, 2021 07:43

hanfei1991 added 2 commits April 28, 2021 17:10

Merge branch 'master' into hanfei/union-all

24cd66e

Merge branch 'master' into hanfei/union-all

924db6e

hanfei1991 added the needs-cherry-pick-release-5.0 label May 8, 2021

Merge branch 'master' into hanfei/union-all

19bff5d

eurekaka reviewed May 17, 2021

View reviewed changes

fzhedu reviewed May 17, 2021

View reviewed changes

address comments

1cf85b4

fzhedu reviewed May 17, 2021

View reviewed changes

lzmhhh123 reviewed May 18, 2021

View reviewed changes

Update planner/core/fragment.go

429d01a

Co-authored-by: Zhuomin Liu <lzmhhh123@gmail.com>

eurekaka added the sig/planner SIG: Planner label Jun 1, 2021

eurekaka approved these changes Jun 1, 2021

View reviewed changes

ti-chi-bot added status/LGT2 Indicates that a PR has LGTM 2. and removed status/LGT1 Indicates that a PR has LGTM 1. labels Jun 1, 2021

ti-chi-bot added the status/can-merge Indicates a PR has been approved by a committer. label Jun 1, 2021

ti-chi-bot added 4 commits June 1, 2021 15:05

Merge branch 'master' into hanfei/union-all

fe75b8f

Merge branch 'master' into hanfei/union-all

236d5c7

Merge branch 'master' into hanfei/union-all

b14e9d5

Merge branch 'master' into hanfei/union-all

b10fec0

fzhedu reviewed Jun 1, 2021

View reviewed changes

ti-chi-bot added 6 commits June 2, 2021 11:29

Merge branch 'master' into hanfei/union-all

d2a7b48

Merge branch 'master' into hanfei/union-all

f2e34db

Merge branch 'master' into hanfei/union-all

74a4ddf

Merge branch 'master' into hanfei/union-all

23e7868

Merge branch 'master' into hanfei/union-all

7379a67

Merge branch 'master' into hanfei/union-all

5e6d5b1

ti-chi-bot merged commit 52e89cb into pingcap:master Jun 2, 2021

ti-srebot pushed a commit to ti-srebot/tidb that referenced this pull request Jun 2, 2021

cherry pick pingcap#24287 to release-5.0

bc676ea

Signed-off-by: ti-srebot <ti-srebot@pingcap.com>

ti-srebot mentioned this pull request Jun 2, 2021

planner/core: support union all for mpp. (#24287) #25051

Merged

hanfei1991 deleted the hanfei/union-all branch June 2, 2021 08:54

winoros mentioned this pull request Jun 9, 2021

executor: select result with truncated values is not compatible with MySQL #24893

Closed

ti-chi-bot pushed a commit that referenced this pull request Jun 21, 2021

planner/core: support union all for mpp. (#24287) (#25051)

8fa3686

		func (p *PhysicalUnionAll) attach2Task(tasks ...task) task {
		if _, ok := tasks[0].(*mppTask); ok {

planner/core: support union all for mpp. #24287

planner/core: support union all for mpp. #24287

Conversation

hanfei1991 commented Apr 26, 2021

What problem does this PR solve?

What is changed and how it works?

Related changes

Check List

Release note

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ti-chi-bot commented Jun 1, 2021

eurekaka commented Jun 1, 2021

ti-chi-bot commented Jun 1, 2021

ti-chi-bot commented Jun 1, 2021

eurekaka commented Jun 1, 2021

ti-chi-bot commented Jun 1, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hanfei1991 commented Jun 2, 2021

ti-srebot commented Jun 2, 2021