executor: support group_concat under new aggregation evaluation framework

executor/aggfuncs/aggfuncs.go

@@ -38,6 +38,9 @@ var (
 	// All the AggFunc implementations for "MAX" are listed here.
 	// All the AggFunc implementations for "MIN" are listed here.
 	// All the AggFunc implementations for "GROUP_CONCAT" are listed here.
+	_ AggFunc = (*groupConcat4DistinctString)(nil)
+	_ AggFunc = (*groupConcat4String)(nil)
+
 	// All the AggFunc implementations for "BIT_OR" are listed here.
 	_ AggFunc = (*bitOrUint64)(nil)
 

executor/aggfuncs/builder.go

@@ -14,10 +14,13 @@
 package aggfuncs
 
 import (
+	"github.com/juju/errors"
 	"github.com/pingcap/tidb/ast"
+	"github.com/pingcap/tidb/expression"
 	"github.com/pingcap/tidb/expression/aggregation"
 	"github.com/pingcap/tidb/mysql"
 	"github.com/pingcap/tidb/types"
+	log "github.com/sirupsen/logrus"
 )
 
 // Build is used to build a specific AggFunc implementation according to the
@@ -116,7 +119,34 @@ func buildMin(aggFuncDesc *aggregation.AggFuncDesc, ordinal int) AggFunc {
 
 // buildCount builds the AggFunc implementation for function "GROUP_CONCAT".
 func buildGroupConcat(aggFuncDesc *aggregation.AggFuncDesc, ordinal int) AggFunc {
-	return nil
+	// TODO: There might be different kind of types of the args,
+	// we should add CastAsString upon every arg after cast can be pushed down to coprocessor.
+	// And this check can be removed at that time.
+	for _, arg := range aggFuncDesc.Args {
+		if arg.GetType().EvalType() != types.ETString {
+			return nil
+		}
+	}
+	base := baseAggFunc{
+		args:    aggFuncDesc.Args[:len(aggFuncDesc.Args)-1],
+		ordinal: ordinal,
+	}
+	switch aggFuncDesc.Mode {
+	case aggregation.DedupMode:
+		return nil
+	default:
+		// The last arg is promised to be a not-null string constant, so the error can be ignored.
+		c, _ := aggFuncDesc.Args[len(aggFuncDesc.Args)-1].(*expression.Constant)
+		sep, _, err := c.EvalString(nil, nil)
+		// This err will never happen, we check it here for passing the errcheck.
+		if err != nil {
+			log.Warning("Error happened when buildGroupConcat:", errors.Trace(err).Error())
+		}
+		if aggFuncDesc.HasDistinct {
+			return &groupConcat4DistinctString{baseGroupConcat4String{baseAggFunc: base, sep: sep}}
+		}
+		return &groupConcat4String{baseGroupConcat4String{baseAggFunc: base, sep: sep}}
+	}
 }
 
 // buildCount builds the AggFunc implementation for function "BIT_OR".

executor/aggfuncs/func_group_concat.go

@@ -0,0 +1,144 @@
+// Copyright 2018 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package aggfuncs
+
+import (
+	"bytes"
+	"strings"
+
+	"github.com/juju/errors"
+	"github.com/pingcap/tidb/sessionctx"
+	"github.com/pingcap/tidb/util/chunk"
+)
+
+type baseGroupConcat4String struct {
+	baseAggFunc
+
+	sep string
+}
+
+func (e *baseGroupConcat4String) AppendFinalResult2Chunk(sctx sessionctx.Context, pr PartialResult, chk *chunk.Chunk) error {
+	p := (*partialResult4ConcatString)(pr)
+	if p.buffer == nil {
+		chk.AppendNull(e.ordinal)
+		return nil
+	}
+	chk.AppendString(e.ordinal, p.buffer.String())
+	return nil
+}
+
+type basePartialResult4GroupConcat struct {
+	buffer *bytes.Buffer
+}
+
+type partialResult4ConcatString struct {
+	basePartialResult4GroupConcat
+}
+
+type groupConcat4String struct {
+	baseGroupConcat4String
+}
+
+func (e *groupConcat4String) AllocPartialResult() PartialResult {
+	return PartialResult(new(partialResult4ConcatString))
+}
+
+func (e *groupConcat4String) ResetPartialResult(pr PartialResult) {
+	p := (*partialResult4ConcatString)(pr)
+	p.buffer = nil
+}
+
+func (e *groupConcat4String) UpdatePartialResult(sctx sessionctx.Context, rowsInGroup []chunk.Row, pr PartialResult) (err error) {
+	p := (*partialResult4ConcatString)(pr)
+	v, isNull := "", false
+	for _, row := range rowsInGroup {
+		isWriteSep := false
+		for i, l := 0, len(e.args); i < l; i++ {
+			v, isNull, err = e.args[i].EvalString(sctx, row)
+			if err != nil {
+				return errors.Trace(err)
+			}
+			if isNull {
+				continue
+			}
+			isWriteSep = true
+			if p.buffer == nil {
+				p.buffer = &bytes.Buffer{}
+			}
+			p.buffer.WriteString(v)
+		}
+		if isWriteSep {
+			p.buffer.WriteString(e.sep)
+		}
+	}
+	p.buffer.Truncate(p.buffer.Len() - 1)
+	// TODO: if total length is greater than global var group_concat_max_len, truncate it.
+	// issue: #7034
+	return nil
+}
+
+type partialResult4ConcatDistinctString struct {
+	basePartialResult4GroupConcat
+	valSet stringSet
+}
+
+type groupConcat4DistinctString struct {
+	baseGroupConcat4String
+}
+
+func (e *groupConcat4DistinctString) AllocPartialResult() PartialResult {
+	p := new(partialResult4ConcatDistinctString)
+	p.valSet = newStringSet()
+	return PartialResult(p)
+}
+
+func (e *groupConcat4DistinctString) ResetPartialResult(pr PartialResult) {
+	p := (*partialResult4ConcatDistinctString)(pr)
+	p.buffer, p.valSet = nil, newStringSet()
+}
+
+func (e *groupConcat4DistinctString) UpdatePartialResult(sctx sessionctx.Context, rowsInGroup []chunk.Row, pr PartialResult) (err error) {
+	p := (*partialResult4ConcatDistinctString)(pr)
+	v, isNull, valsBuf, joinedVals := "", false, make([]string, len(e.args)), ""
+	for _, row := range rowsInGroup {
+		for i, l := 0, len(e.args); i < l; i++ {
+			v, isNull, err = e.args[i].EvalString(sctx, row)
+			if err != nil {
+				return errors.Trace(err)
+			}
+			if isNull {
+				continue
+			}
+			valsBuf[i] = v
+		}
+		joinedVals = strings.Join(valsBuf, "")
+		if p.valSet.exist(joinedVals) {
+			continue
+		}
+		p.valSet.insert(joinedVals)
+		// write separator
+		if p.buffer == nil {
+			p.buffer = &bytes.Buffer{}
+		} else {
+			p.buffer.WriteString(e.sep)
+		}
+		// write values
+		for _, s := range valsBuf {
+			p.buffer.WriteString(s)
+		}
+	}
+	// TODO: if total length is greater than global var group_concat_max_len, truncate it.
+	// issue: #7034
+	return nil
+}

executor/aggfuncs/sets.go

@@ -19,6 +19,7 @@ import (
 
 type decimalSet map[types.MyDecimal]struct{}
 type float64Set map[float64]struct{}
+type stringSet map[string]struct{}
 
 func newDecimalSet() decimalSet {
 	return make(map[types.MyDecimal]struct{})
@@ -45,3 +46,16 @@ func (s float64Set) exist(val float64) bool {
 func (s float64Set) insert(val float64) {
 	s[val] = struct{}{}
 }
+
+func newStringSet() stringSet {
+	return make(map[string]struct{})
+}
+
+func (s stringSet) exist(val string) bool {
+	_, ok := s[val]
+	return ok
+}
+
+func (s stringSet) insert(val string) {
+	s[val] = struct{}{}
+}