executor: introduce a new execution framework for aggregate functions #6852

zz-jason · 2018-06-17T12:58:42Z

What have you changed? (mandatory)

Introduce a new interface named AggFunc defined in executor/aggfuncs/aggfuncs.go to refactor the execution framework of aggregate functions. The main usage of the new execution framework is:

use AllocPartialResult() to allocate the struct to store the partial result for every aggregate function
use UpdatePartialResult() to update the partial result for every aggregate function, no mater whether the input is the original or partial data. The input partialBytes will be converted to the specific partial result struct before update.
use ResetPartialResult() to reset or reinitialize the partial result for every aggregate function. The input partialBytes will be converted to the specific partial result struct before reinitialization.
use AppendFinalResult2Chunk() to finalize the partial result to the input chk. The input partialBytes will be converted to the specific partial result before finalization every group.

The main improvements are:

by calling UpdatePartialResult() with []chunk.Row, we can reduce the total function calls, which saves a lot of time. And for stream aggregate, the input data for a aggregate function are stored sequentially in the input []chunk.Row, which can further improve the CPU cache performance.
by calling AllocPartialResult() to allocate the specific struct to store the partial result for every aggregate function, we can reduce the redundant memory usage in the old struct AggEvaluateContext.

Use aggfuncs.Build to create a AggFunc according to the AggFuncDesc. For now:

only partially supported some implementations of AVG
the new execution framework is only supported in the StreamAggExec if possiable

What are the type of the changes (mandatory)?

Improvement (non-breaking change which is an improvement to an existing feature)

How has this PR been tested (mandatory)?

unit test
explain test

Does this PR affect documentation (docs/docs-cn) update? (optional)

No

Benchmark result if necessary (optional)

test sql:

mysql root@172.16.10.112:tpch> desc select avg(L_QUANTITY) from (select * from lineitem union all select * from lineitem) tmp;
+----------------+--------------+-------------------------------+------+-----------------------------------------------------+--------------+
| id             | parents      | children                      | task | operator info                                       | count        |
+----------------+--------------+-------------------------------+------+-----------------------------------------------------+--------------+
| TableScan_23   |              |                               | cop  | table:lineitem, range:[-inf,+inf], keep order:false | 59986052.00  |
| TableReader_24 | Union_21     |                               | root | data:TableScan_23                                   | 59986052.00  |
| TableScan_26   |              |                               | cop  | table:lineitem, range:[-inf,+inf], keep order:false | 59986052.00  |
| TableReader_27 | Union_21     |                               | root | data:TableScan_26                                   | 59986052.00  |
| Union_21       | StreamAgg_13 | TableReader_24,TableReader_27 | root |                                                     | 119972104.00 |
| StreamAgg_13   |              | Union_21                      | root | funcs:avg(tmp.l_quantity)                           | 1.00         |
+----------------+--------------+-------------------------------+------+-----------------------------------------------------+--------------+
6 rows in set
Time: 0.017s

Before this PR:

mysql root@172.16.10.112:tpch> select avg(L_QUANTITY) from (select * from lineitem union all select * from lineitem) tmp;
+-----------------+
| avg(L_QUANTITY) |
+-----------------+
| 25.501562       |
+-----------------+
1 row in set
Time: 54.508s

After this PR:

mysql root@172.16.10.112:tpch> select avg(L_QUANTITY) from (select * from lineitem union all select * from lineitem) tmp;
+-----------------+
| avg(L_QUANTITY) |
+-----------------+
| 25.501562       |
+-----------------+
1 row in set
Time: 27.767s

The performance gain is about 96%

zz-jason · 2018-06-17T12:58:54Z

/run-all-tests

winoros · 2018-06-20T03:49:50Z

Before's result and After's is placed in the wrong position?

zz-jason · 2018-06-20T04:47:55Z

@winoros updated

zz-jason · 2018-06-21T04:27:11Z

/run-all-tests

zz-jason · 2018-06-21T07:17:41Z

/run-all-tests

zz-jason · 2018-06-21T07:46:23Z

/run-all-tests

zz-jason · 2018-06-21T11:35:46Z

/run-all-tests

zz-jason · 2018-06-21T14:06:46Z

/run-all-tests

zz-jason · 2018-06-21T14:51:57Z

/run-all-tests

zz-jason · 2018-06-21T15:25:30Z

@XuHuaiyu @winoros @lamxTyler PTAL

zz-jason · 2018-06-22T02:56:15Z

@lysu PTAL

alivxxx · 2018-06-22T05:53:36Z

executor/aggregate.go

-func (e *StreamAggExec) appendResult2Chunk(chk *chunk.Chunk) {
+func (e *StreamAggExec) appendResult2Chunk(chk *chunk.Chunk) error {
+	if e.newAggFuncs != nil {
+		fmt.Printf("StreamAggExec.appendResult2Chunk: use new aggfunc\n")


Remove the debug log.

…nto dev/refactor-agg

XuHuaiyu · 2018-06-25T02:53:30Z

executor/aggfuncs/aggfuncs.go

+type AggFunc interface {
+	// AllocPartialResult allocates a specific data structure to store the
+	// partial result, initializes it, and converts it to a bype slice to return
+	// back. Aggregate operator implementations, no mater whether it's a hash or


s/ mater/ matter

XuHuaiyu · 2018-06-25T02:59:00Z

executor/aggfuncs/builder.go

+	"github.com/pingcap/tidb/expression/aggregation"
+)
+
+// Build is used to build a specific AggFunc implementation according to the


add . at the end of this comment.
so as the other comments

XuHuaiyu · 2018-06-25T03:03:21Z

executor/builder.go

+			newAggFuncs = append(newAggFuncs, newAggFunc)
+		}
+	}
+	if len(newAggFuncs) == len(v.AggFuncs) {


Add a comment for this check.

XuHuaiyu · 2018-06-25T03:03:23Z

executor/builder.go

+			newAggFuncs = append(newAggFuncs, newAggFunc)
+		}
+	}
+	if len(newAggFuncs) == len(v.AggFuncs) {


Add a comment for this check.

XuHuaiyu · 2018-06-25T03:44:39Z

executor/aggfuncs/aggfuncs.go

+
+type baseAggFunc struct {
+	input  []expression.Expression
+	output []int


add comments for these two args.

XuHuaiyu · 2018-06-25T04:48:42Z

executor/aggregate.go

 func (e *StreamAggExec) fetchChildIfNecessary(ctx context.Context, chk *chunk.Chunk) error {
 	if e.inputRow != e.inputIter.End() {
 		return nil
 	}

+	if e.newAggFuncs != nil {


why we need to consumeGroupRows here?

before calling fetchChildIfNecessary, we may have some unconsumed rows stored in e.childrenResults[0], we should consume them before calling e.children[0].Next, which will reset e.childrenResults[0] before execution.

put this check between line 279 and line 280 may be better?

No, if we put this check to that position, we have to call consumeGroupRows() for every input row.

XuHuaiyu · 2018-06-25T04:55:02Z

executor/aggfuncs/aggfuncs.go

+	// input byte slice to the specific data structure which stores the partial
+	// result and then calculates the final result and append that final result
+	// to the chunk provided.
+	AppendFinalResult2Chunk(sctx sessionctx.Context, partialBytes []byte, chk *chunk.Chunk) error


s/ AppendFinalResult2Chunk/ GetFinalResult

I prefer the original name, which indicates the result is appended to the output chunk

XuHuaiyu · 2018-06-25T05:04:57Z

executor/aggfuncs/aggfuncs.go

+	// back. Aggregate operator implementations, no mater whether it's a hash or
+	// stream implementation, should hold this byte slice for further operations
+	// like: "ResetPartialResult", "UpdatePartialResult".
+	AllocPartialResult() []byte


It seems that we need another struct which contains partialResultBytes to handle the hashagg evaluation and aggfunc with distinct?

no need for now, we can just add a map field in a specific aggregate function implementation, during the execution of UpdatePartialResult we use that map to deduplicate the input, when ResetPartialResult, we reset that map.

XuHuaiyu · 2018-06-25T05:11:17Z

executor/aggfuncs/func_avg.go

@@ -100,6 +100,7 @@ func (e *avgDedup4Decimal) UpdatePartialResult(sctx sessionctx.Context, rowsInGr

 type avgOriginal4Decimal struct {
 	baseAvgDecimal
+	deDuper map[types.MyDecimal]bool


deDuper should be initialized

XuHuaiyu · 2018-06-25T05:12:21Z

executor/aggfuncs/builder.go

@@ -80,11 +81,20 @@ func buildAvg(aggFuncDesc *aggregation.AggFuncDesc, output []int) AggFunc {
 	case aggregation.CompleteMode, aggregation.Partial1Mode:
 		switch aggFuncDesc.Args[0].GetType().Tp {


we should consider all the input types,
use EvalType here may be better?

XuHuaiyu · 2018-06-26T09:05:07Z

executor/builder.go

 		e.AggFuncs = append(e.AggFuncs, aggDesc.GetAggFunc())
+		newAggFunc := aggfuncs.Build(aggDesc, []int{i})


why do we need to pass a slice
since there is only one element in the slice?

XuHuaiyu · 2018-06-26T09:07:59Z

executor/aggfuncs/aggfuncs.go

+	// input PartialResult to the specific data structure which stores the
+	// partial result and then calculates the final result and append that
+	// final result to the chunk provided.
+	AppendFinalResult2Chunk(sctx sessionctx.Context, pr PartialResult, chk *chunk.Chunk) error


s/ AppendFinalResult2Chunk/ GetFinalResult

I prefer the original name, which indicates the result is appended to the output chunk

XuHuaiyu · 2018-06-26T09:12:50Z

executor/aggfuncs/aggfuncs.go

+type baseAggFunc struct {
+	// input stores the input arguments for an aggregate function, we should
+	// call input.EvalXXX to get the actual input data for this function.
+	input []expression.Expression


s/ input/ args may be clearer.

we do not need to define output as a slice,
since we only use it to append the final result to a chunk.

XuHuaiyu · 2018-06-26T09:15:08Z

Do we need a GetPartialResult func which may used by mocktikv.

zz-jason · 2018-06-26T09:17:20Z

@XuHuaiyu If we only decide to use it in the final or complete mode, we don't need to add the GetPartialResult, mocktikv can just use the origin old aggregate funcs.

XuHuaiyu · 2018-06-26T12:26:19Z

PTAL @coocood

lysu · 2018-06-28T08:13:20Z

/run-all-tests tidb-test=pr/559

coocood · 2018-06-29T05:50:05Z

executor/aggregate.go

+
+	// for the new execution framework of aggregate functions
+	newAggFuncs    []aggfuncs.AggFunc
+	partialResults []aggfuncs.PartialResult


Why do we need to hold partialResults here instead of in each AggFunc?

It'e better to let aggregate function implementations to be stateless. If not so, we have to allocate an aggregate function for every group, this is worse when we use it in the hash aggregate operator.

For example, ClickHouse also has the same aggregate function framework: https://github.com/yandex/ClickHouse/blob/master/dbms/src/AggregateFunctions/AggregateFunctionAvg.h, and so does the Impala: https://github.com/cloudera/Impala/blob/cdh5-trunk/be/src/udf/udf.h

XuHuaiyu · 2018-06-29T05:51:21Z

LGTM

coocood · 2018-06-29T08:34:40Z

LGTM

executor: refactor aggregate functions

182658b

change type name

01ec369

zz-jason added sig/execution SIG execution type/enhancement The issue or PR belongs to an enhancement. labels Jun 21, 2018

Merge branch 'master' into dev/refactor-agg

faa3f7d

Merge branch 'master' into dev/refactor-agg

1ebf4d5

Merge branch 'master' into dev/refactor-agg

038135a

handle distinct in partial1 and complete mode

7f1fd3a

handle 0 count

e5acefb

zz-jason added the status/all tests passed label Jun 21, 2018

Merge branch 'master' into dev/refactor-agg

e25d273

alivxxx reviewed Jun 22, 2018

View reviewed changes

zz-jason added 6 commits June 23, 2018 20:41

add implementation of SUM

dca132a

Merge branch 'dev/refactor-agg' of https://github.com/zz-jason/tidb i…

81c812b

…nto dev/refactor-agg

add missing file

cacad7c

only introduce the framework in this PR

d3a1ced

remove useless code

cf34a16

remove debug log

b1335c3

XuHuaiyu reviewed Jun 25, 2018

View reviewed changes

zz-jason added 2 commits June 25, 2018 16:39

address comment

5f17822

address comment

aa71782

XuHuaiyu reviewed Jun 26, 2018

View reviewed changes

address comment

f3b005f

Merge branch 'master' into dev/refactor-agg

c7f7144

zz-jason added 2 commits June 28, 2018 20:27

Merge branch 'master' into dev/refactor-agg

66801be

addres comment

0d859db

coocood reviewed Jun 29, 2018

View reviewed changes

zz-jason added the status/LGT1 Indicates that a PR has LGTM 1. label Jun 29, 2018

coocood added status/LGT2 Indicates that a PR has LGTM 2. and removed status/LGT1 Indicates that a PR has LGTM 1. labels Jun 29, 2018

coocood approved these changes Jun 29, 2018

View reviewed changes

Merge branch 'master' into dev/refactor-agg

7f8134c

zz-jason merged commit 3c05d77 into pingcap:master Jun 29, 2018

zz-jason deleted the dev/refactor-agg branch June 29, 2018 08:52

This was referenced Jul 2, 2018

aggfuncs: partially implement "AVG" #6951

Merged

implement aggregate functions under the new framework #6952

Closed

XuHuaiyu mentioned this pull request Jul 3, 2018

executor: support MAX/MIN in new evaluation framework partially #6971

Merged

crazycs520 mentioned this pull request Jul 4, 2018

aggfuncs: implement bit-or with new aggregation framework #6975

Merged

mccxj mentioned this pull request Sep 2, 2018

slice bounds out of range #7578

Closed

		@@ -80,11 +81,20 @@ func buildAvg(aggFuncDesc *aggregation.AggFuncDesc, output []int) AggFunc {
		case aggregation.CompleteMode, aggregation.Partial1Mode:
		switch aggFuncDesc.Args[0].GetType().Tp {

		e.AggFuncs = append(e.AggFuncs, aggDesc.GetAggFunc())
		newAggFunc := aggfuncs.Build(aggDesc, []int{i})

executor: introduce a new execution framework for aggregate functions #6852

executor: introduce a new execution framework for aggregate functions #6852

Conversation

zz-jason commented Jun 17, 2018 • edited Loading

What have you changed? (mandatory)

What are the type of the changes (mandatory)?

How has this PR been tested (mandatory)?

Does this PR affect documentation (docs/docs-cn) update? (optional)

Benchmark result if necessary (optional)

zz-jason commented Jun 17, 2018

winoros commented Jun 20, 2018

zz-jason commented Jun 20, 2018

zz-jason commented Jun 21, 2018

zz-jason commented Jun 21, 2018

zz-jason commented Jun 21, 2018

zz-jason commented Jun 21, 2018

zz-jason commented Jun 21, 2018

zz-jason commented Jun 21, 2018

zz-jason commented Jun 21, 2018

zz-jason commented Jun 22, 2018

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

XuHuaiyu commented Jun 26, 2018

zz-jason commented Jun 26, 2018

XuHuaiyu commented Jun 26, 2018

lysu commented Jun 28, 2018

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

XuHuaiyu commented Jun 29, 2018

coocood commented Jun 29, 2018

zz-jason commented Jun 17, 2018 •

edited

Loading