executor: Improve the performance of aggFuncSum by using sliding window

executor/aggfuncs/builder.go

@@ -34,7 +34,7 @@ func Build(ctx sessionctx.Context, aggFuncDesc *aggregation.AggFuncDesc, ordinal
 	case ast.AggFuncCount:
 		return buildCount(aggFuncDesc, ordinal)
 	case ast.AggFuncSum:
-		return buildSum(aggFuncDesc, ordinal)
+		return buildSum(ctx, aggFuncDesc, ordinal)
 	case ast.AggFuncAvg:
 		return buildAvg(aggFuncDesc, ordinal)
 	case ast.AggFuncFirstRow:
@@ -134,7 +134,7 @@ func buildCount(aggFuncDesc *aggregation.AggFuncDesc, ordinal int) AggFunc {
 }
 
 // buildSum builds the AggFunc implementation for function "SUM".
-func buildSum(aggFuncDesc *aggregation.AggFuncDesc, ordinal int) AggFunc {
+func buildSum(ctx sessionctx.Context, aggFuncDesc *aggregation.AggFuncDesc, ordinal int) AggFunc {
 	base := baseSumAggFunc{
 		baseAggFunc: baseAggFunc{
 			args:    aggFuncDesc.Args,
@@ -155,7 +155,10 @@ func buildSum(aggFuncDesc *aggregation.AggFuncDesc, ordinal int) AggFunc {
 			if aggFuncDesc.HasDistinct {
 				return &sum4DistinctFloat64{base}
 			}
-			return &sum4Float64{base}
+			if ctx.GetSessionVars().WindowingUseHighPrecision {
+				return &sum4Float64HighPrecision{baseSum4Float64{base}}
+			}
+			return &sum4Float64{baseSum4Float64{base}}
 		}
 	}
 }

executor/aggfuncs/func_sum.go

@@ -22,56 +22,57 @@ import (
 )
 
 type partialResult4SumFloat64 struct {
-	val    float64
-	isNull bool
+	val             float64
+	notNullRowCount int64
 }
 
 type partialResult4SumDecimal struct {
-	val    types.MyDecimal
-	isNull bool
+	val             types.MyDecimal
+	notNullRowCount int64
 }
 
 type partialResult4SumDistinctFloat64 struct {
-	partialResult4SumFloat64
+	val    float64
+	isNull bool
 	valSet set.Float64Set
 }
 
 type partialResult4SumDistinctDecimal struct {
-	partialResult4SumDecimal
+	val    types.MyDecimal
+	isNull bool
 	valSet set.StringSet
 }
 
 type baseSumAggFunc struct {
 	baseAggFunc
 }
 
-type sum4Float64 struct {
+type baseSum4Float64 struct {
 	baseSumAggFunc
 }
 
-func (e *sum4Float64) AllocPartialResult() PartialResult {
+func (e *baseSum4Float64) AllocPartialResult() PartialResult {
 	p := new(partialResult4SumFloat64)
-	p.isNull = true
 	return PartialResult(p)
 }
 
-func (e *sum4Float64) ResetPartialResult(pr PartialResult) {
+func (e *baseSum4Float64) ResetPartialResult(pr PartialResult) {
 	p := (*partialResult4SumFloat64)(pr)
 	p.val = 0
-	p.isNull = true
+	p.notNullRowCount = 0
 }
 
-func (e *sum4Float64) AppendFinalResult2Chunk(sctx sessionctx.Context, pr PartialResult, chk *chunk.Chunk) error {
+func (e *baseSum4Float64) AppendFinalResult2Chunk(sctx sessionctx.Context, pr PartialResult, chk *chunk.Chunk) error {
 	p := (*partialResult4SumFloat64)(pr)
-	if p.isNull {
+	if p.notNullRowCount == 0 {
 		chk.AppendNull(e.ordinal)
 		return nil
 	}
 	chk.AppendFloat64(e.ordinal, p.val)
 	return nil
 }
 
-func (e *sum4Float64) UpdatePartialResult(sctx sessionctx.Context, rowsInGroup []chunk.Row, pr PartialResult) error {
+func (e *baseSum4Float64) UpdatePartialResult(sctx sessionctx.Context, rowsInGroup []chunk.Row, pr PartialResult) error {
 	p := (*partialResult4SumFloat64)(pr)
 	for _, row := range rowsInGroup {
 		input, isNull, err := e.args[0].EvalReal(sctx, row)
@@ -81,44 +82,74 @@ func (e *sum4Float64) UpdatePartialResult(sctx sessionctx.Context, rowsInGroup [
 		if isNull {
 			continue
 		}
-		if p.isNull {
-			p.val = input
-			p.isNull = false
-			continue
-		}
 		p.val += input
+		p.notNullRowCount++
 	}
 	return nil
 }
 
-func (e *sum4Float64) MergePartialResult(sctx sessionctx.Context, src, dst PartialResult) error {
+func (e *baseSum4Float64) MergePartialResult(sctx sessionctx.Context, src, dst PartialResult) error {
 	p1, p2 := (*partialResult4SumFloat64)(src), (*partialResult4SumFloat64)(dst)
-	if p1.isNull {
+	if p1.notNullRowCount == 0 {
 		return nil
 	}
 	p2.val += p1.val
-	p2.isNull = false
+	p2.notNullRowCount += p1.notNullRowCount
+	return nil
+}
+
+type sum4Float64 struct {
+	baseSum4Float64
+}
+
+func (e *sum4Float64) Slide(sctx sessionctx.Context, rows []chunk.Row, lastStart, lastEnd uint64, shiftStart, shiftEnd uint64, pr PartialResult) error {
+	p := (*partialResult4SumFloat64)(pr)
+	for i := uint64(0); i < shiftEnd; i++ {
+		input, isNull, err := e.args[0].EvalReal(sctx, rows[lastEnd+i])
+		if err != nil {
+			return err
+		}
+		if isNull {
+			continue
+		}
+		p.val += input
+		p.notNullRowCount++
+	}
+	for i := uint64(0); i < shiftStart; i++ {
+		input, isNull, err := e.args[0].EvalReal(sctx, rows[lastStart+i])
+		if err != nil {
+			return err
+		}
+		if isNull {
+			continue
+		}
+		p.val -= input
+		p.notNullRowCount--
+	}
 	return nil
 }
 
+type sum4Float64HighPrecision struct {
+	baseSum4Float64
+}
+
 type sum4Decimal struct {
 	baseSumAggFunc
 }
 
 func (e *sum4Decimal) AllocPartialResult() PartialResult {
 	p := new(partialResult4SumDecimal)
-	p.isNull = true
 	return PartialResult(p)
 }
 
 func (e *sum4Decimal) ResetPartialResult(pr PartialResult) {
 	p := (*partialResult4SumDecimal)(pr)
-	p.isNull = true
+	p.notNullRowCount = 0
 }
 
 func (e *sum4Decimal) AppendFinalResult2Chunk(sctx sessionctx.Context, pr PartialResult, chk *chunk.Chunk) error {
 	p := (*partialResult4SumDecimal)(pr)
-	if p.isNull {
+	if p.notNullRowCount == 0 {
 		chk.AppendNull(e.ordinal)
 		return nil
 	}
@@ -136,9 +167,9 @@ func (e *sum4Decimal) UpdatePartialResult(sctx sessionctx.Context, rowsInGroup [
 		if isNull {
 			continue
 		}
-		if p.isNull {
+		if p.notNullRowCount == 0 {
 			p.val = *input
-			p.isNull = false
+			p.notNullRowCount = 1
 			continue
 		}
 
@@ -148,13 +179,56 @@ func (e *sum4Decimal) UpdatePartialResult(sctx sessionctx.Context, rowsInGroup [
 			return err
 		}
 		p.val = *newSum
+		p.notNullRowCount++
+	}
+	return nil
+}
+
+func (e *sum4Decimal) Slide(sctx sessionctx.Context, rows []chunk.Row, lastStart, lastEnd uint64, shiftStart, shiftEnd uint64, pr PartialResult) error {
+	p := (*partialResult4SumDecimal)(pr)
+	for i := uint64(0); i < shiftEnd; i++ {
+		input, isNull, err := e.args[0].EvalDecimal(sctx, rows[lastEnd+i])
+		if err != nil {
+			return err
+		}
+		if isNull {
+			continue
+		}
+		if p.notNullRowCount == 0 {
+			p.val = *input
+			p.notNullRowCount = 1
+			continue
+		}
+		newSum := new(types.MyDecimal)
+		err = types.DecimalAdd(&p.val, input, newSum)
+		if err != nil {
+			return err
+		}
+		p.val = *newSum
+		p.notNullRowCount++
+	}
+	for i := uint64(0); i < shiftStart; i++ {
+		input, isNull, err := e.args[0].EvalDecimal(sctx, rows[lastStart+i])
+		if err != nil {
+			return err
+		}
+		if isNull {
+			continue
+		}
+		newSum := new(types.MyDecimal)
+		err = types.DecimalSub(&p.val, input, newSum)
+		if err != nil {
+			return err
+		}
+		p.val = *newSum
+		p.notNullRowCount--
 	}
 	return nil
 }
 
 func (e *sum4Decimal) MergePartialResult(sctx sessionctx.Context, src, dst PartialResult) error {
 	p1, p2 := (*partialResult4SumDecimal)(src), (*partialResult4SumDecimal)(dst)
-	if p1.isNull {
+	if p1.notNullRowCount == 0 {
 		return nil
 	}
 	newSum := new(types.MyDecimal)
@@ -163,7 +237,7 @@ func (e *sum4Decimal) MergePartialResult(sctx sessionctx.Context, src, dst Parti
 		return err
 	}
 	p2.val = *newSum
-	p2.isNull = false
+	p2.notNullRowCount += p1.notNullRowCount
 	return nil
 }