You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
352 lines
10 KiB
352 lines
10 KiB
// Copyright 2020 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 (
|
|
"math"
|
|
"sort"
|
|
|
|
"github.com/pingcap/tidb/sessionctx"
|
|
"github.com/pingcap/tidb/types"
|
|
"github.com/pingcap/tidb/util/chunk"
|
|
"github.com/pingcap/tidb/util/selection"
|
|
)
|
|
|
|
var (
|
|
_ partialResult4Percentile = partialResult4PercentileInt{}
|
|
_ partialResult4Percentile = partialResult4PercentileReal{}
|
|
_ partialResult4Percentile = partialResult4PercentileDecimal{}
|
|
_ partialResult4Percentile = partialResult4PercentileTime{}
|
|
_ partialResult4Percentile = partialResult4PercentileDuration{}
|
|
)
|
|
|
|
func percentile(data sort.Interface, percent int) int {
|
|
// Ordinal rank k = Ceil(P / 100 * N)
|
|
k := int(math.Ceil(float64(data.Len()) / 100 * float64(percent)))
|
|
return selection.Select(data, k)
|
|
}
|
|
|
|
type basePercentile struct {
|
|
percent int
|
|
|
|
baseAggFunc
|
|
}
|
|
|
|
func (e *basePercentile) AllocPartialResult() (pr PartialResult) {
|
|
return
|
|
}
|
|
|
|
func (e *basePercentile) ResetPartialResult(pr PartialResult) {
|
|
return
|
|
}
|
|
|
|
func (e *basePercentile) UpdatePartialResult(sctx sessionctx.Context, rowsInGroup []chunk.Row, pr PartialResult) (err error) {
|
|
return
|
|
}
|
|
|
|
func (e *basePercentile) AppendFinalResult2Chunk(sctx sessionctx.Context, pr PartialResult, chk *chunk.Chunk) error {
|
|
chk.AppendNull(e.ordinal)
|
|
return nil
|
|
}
|
|
|
|
type partialResult4Percentile interface {
|
|
sort.Interface
|
|
}
|
|
|
|
type partialResult4PercentileInt []int64
|
|
|
|
func (p partialResult4PercentileInt) Len() int { return len(p) }
|
|
func (p partialResult4PercentileInt) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
|
|
func (p partialResult4PercentileInt) Less(i, j int) bool { return p[i] < p[j] }
|
|
|
|
type partialResult4PercentileReal []float64
|
|
|
|
func (p partialResult4PercentileReal) Len() int { return len(p) }
|
|
func (p partialResult4PercentileReal) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
|
|
func (p partialResult4PercentileReal) Less(i, j int) bool { return p[i] < p[j] }
|
|
|
|
// TODO: use []*types.MyDecimal to prevent massive value copy
|
|
type partialResult4PercentileDecimal []types.MyDecimal
|
|
|
|
func (p partialResult4PercentileDecimal) Len() int { return len(p) }
|
|
func (p partialResult4PercentileDecimal) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
|
|
func (p partialResult4PercentileDecimal) Less(i, j int) bool { return p[i].Compare(&p[j]) < 0 }
|
|
|
|
type partialResult4PercentileTime []types.Time
|
|
|
|
func (p partialResult4PercentileTime) Len() int { return len(p) }
|
|
func (p partialResult4PercentileTime) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
|
|
func (p partialResult4PercentileTime) Less(i, j int) bool { return p[i].Compare(p[j]) < 0 }
|
|
|
|
type partialResult4PercentileDuration []types.Duration
|
|
|
|
func (p partialResult4PercentileDuration) Len() int { return len(p) }
|
|
func (p partialResult4PercentileDuration) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
|
|
func (p partialResult4PercentileDuration) Less(i, j int) bool { return p[i].Compare(p[j]) < 0 }
|
|
|
|
type percentileOriginal4Int struct {
|
|
basePercentile
|
|
}
|
|
|
|
func (e *percentileOriginal4Int) AllocPartialResult() (pr PartialResult) {
|
|
// TODO: Preserve appropriate capacity for data
|
|
pr = PartialResult(&partialResult4PercentileInt{})
|
|
return pr
|
|
}
|
|
|
|
func (e *percentileOriginal4Int) ResetPartialResult(pr PartialResult) {
|
|
p := (*partialResult4PercentileInt)(pr)
|
|
*p = partialResult4PercentileInt{}
|
|
}
|
|
|
|
func (e *percentileOriginal4Int) UpdatePartialResult(sctx sessionctx.Context, rowsInGroup []chunk.Row, pr PartialResult) (err error) {
|
|
p := (*partialResult4PercentileInt)(pr)
|
|
for _, row := range rowsInGroup {
|
|
v, isNull, err := e.args[0].EvalInt(sctx, row)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if isNull {
|
|
continue
|
|
}
|
|
*p = append(*p, v)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (e *percentileOriginal4Int) MergePartialResult(sctx sessionctx.Context, src, dst PartialResult) (err error) {
|
|
p1, p2 := (*partialResult4PercentileInt)(src), (*partialResult4PercentileInt)(dst)
|
|
mergeBuff := make([]int64, len(*p1)+len(*p2))
|
|
copy(mergeBuff, *p2)
|
|
copy(mergeBuff[len(*p2):], *p1)
|
|
*p1 = nil
|
|
*p2 = mergeBuff
|
|
return nil
|
|
}
|
|
|
|
func (e *percentileOriginal4Int) AppendFinalResult2Chunk(sctx sessionctx.Context, pr PartialResult, chk *chunk.Chunk) error {
|
|
p := (*partialResult4PercentileInt)(pr)
|
|
if len(*p) == 0 {
|
|
chk.AppendNull(e.ordinal)
|
|
return nil
|
|
}
|
|
index := percentile(p, e.percent)
|
|
chk.AppendInt64(e.ordinal, (*p)[index])
|
|
return nil
|
|
}
|
|
|
|
type percentileOriginal4Real struct {
|
|
basePercentile
|
|
}
|
|
|
|
func (e *percentileOriginal4Real) AllocPartialResult() (pr PartialResult) {
|
|
// TODO: Preserve appropriate capacity for data
|
|
pr = PartialResult(&partialResult4PercentileReal{})
|
|
return pr
|
|
}
|
|
|
|
func (e *percentileOriginal4Real) ResetPartialResult(pr PartialResult) {
|
|
p := (*partialResult4PercentileReal)(pr)
|
|
*p = partialResult4PercentileReal{}
|
|
}
|
|
|
|
func (e *percentileOriginal4Real) UpdatePartialResult(sctx sessionctx.Context, rowsInGroup []chunk.Row, pr PartialResult) (err error) {
|
|
p := (*partialResult4PercentileReal)(pr)
|
|
for _, row := range rowsInGroup {
|
|
v, isNull, err := e.args[0].EvalReal(sctx, row)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if isNull {
|
|
continue
|
|
}
|
|
*p = append(*p, v)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (e *percentileOriginal4Real) MergePartialResult(sctx sessionctx.Context, src, dst PartialResult) (err error) {
|
|
p1, p2 := (*partialResult4PercentileReal)(src), (*partialResult4PercentileReal)(dst)
|
|
mergeBuff := make([]float64, len(*p1)+len(*p2))
|
|
copy(mergeBuff, *p2)
|
|
copy(mergeBuff[len(*p2):], *p1)
|
|
*p1 = nil
|
|
*p2 = mergeBuff
|
|
return nil
|
|
}
|
|
|
|
func (e *percentileOriginal4Real) AppendFinalResult2Chunk(sctx sessionctx.Context, pr PartialResult, chk *chunk.Chunk) error {
|
|
p := (*partialResult4PercentileReal)(pr)
|
|
if len(*p) == 0 {
|
|
chk.AppendNull(e.ordinal)
|
|
return nil
|
|
}
|
|
index := percentile(*p, e.percent)
|
|
chk.AppendFloat64(e.ordinal, (*p)[index])
|
|
return nil
|
|
}
|
|
|
|
type percentileOriginal4Decimal struct {
|
|
basePercentile
|
|
}
|
|
|
|
func (e *percentileOriginal4Decimal) AllocPartialResult() (pr PartialResult) {
|
|
// TODO: Preserve appropriate capacity for data
|
|
pr = PartialResult(&partialResult4PercentileDecimal{})
|
|
return pr
|
|
}
|
|
|
|
func (e *percentileOriginal4Decimal) ResetPartialResult(pr PartialResult) {
|
|
p := (*partialResult4PercentileDecimal)(pr)
|
|
*p = partialResult4PercentileDecimal{}
|
|
}
|
|
|
|
func (e *percentileOriginal4Decimal) UpdatePartialResult(sctx sessionctx.Context, rowsInGroup []chunk.Row, pr PartialResult) (err error) {
|
|
p := (*partialResult4PercentileDecimal)(pr)
|
|
for _, row := range rowsInGroup {
|
|
v, isNull, err := e.args[0].EvalDecimal(sctx, row)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if isNull {
|
|
continue
|
|
}
|
|
*p = append(*p, *v)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (e *percentileOriginal4Decimal) MergePartialResult(sctx sessionctx.Context, src, dst PartialResult) (err error) {
|
|
p1, p2 := (*partialResult4PercentileDecimal)(src), (*partialResult4PercentileDecimal)(dst)
|
|
mergeBuff := make([]types.MyDecimal, len(*p1)+len(*p2))
|
|
copy(mergeBuff, *p2)
|
|
copy(mergeBuff[len(*p2):], *p1)
|
|
*p1 = nil
|
|
*p2 = mergeBuff
|
|
return nil
|
|
}
|
|
|
|
func (e *percentileOriginal4Decimal) AppendFinalResult2Chunk(sctx sessionctx.Context, pr PartialResult, chk *chunk.Chunk) error {
|
|
p := (*partialResult4PercentileDecimal)(pr)
|
|
if len(*p) == 0 {
|
|
chk.AppendNull(e.ordinal)
|
|
return nil
|
|
}
|
|
index := percentile(*p, e.percent)
|
|
chk.AppendMyDecimal(e.ordinal, &(*p)[index])
|
|
return nil
|
|
}
|
|
|
|
type percentileOriginal4Time struct {
|
|
basePercentile
|
|
}
|
|
|
|
func (e *percentileOriginal4Time) AllocPartialResult() (pr PartialResult) {
|
|
// TODO: Preserve appropriate capacity for data
|
|
pr = PartialResult(&partialResult4PercentileTime{})
|
|
return pr
|
|
}
|
|
|
|
func (e *percentileOriginal4Time) ResetPartialResult(pr PartialResult) {
|
|
p := (*partialResult4PercentileTime)(pr)
|
|
*p = partialResult4PercentileTime{}
|
|
}
|
|
|
|
func (e *percentileOriginal4Time) UpdatePartialResult(sctx sessionctx.Context, rowsInGroup []chunk.Row, pr PartialResult) (err error) {
|
|
p := (*partialResult4PercentileTime)(pr)
|
|
for _, row := range rowsInGroup {
|
|
v, isNull, err := e.args[0].EvalTime(sctx, row)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if isNull {
|
|
continue
|
|
}
|
|
*p = append(*p, v)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (e *percentileOriginal4Time) MergePartialResult(sctx sessionctx.Context, src, dst PartialResult) (err error) {
|
|
p1, p2 := (*partialResult4PercentileTime)(src), (*partialResult4PercentileTime)(dst)
|
|
mergeBuff := make(partialResult4PercentileTime, len(*p1)+len(*p2))
|
|
copy(mergeBuff, *p2)
|
|
copy(mergeBuff[len(*p2):], *p1)
|
|
*p1 = nil
|
|
*p2 = mergeBuff
|
|
return nil
|
|
}
|
|
|
|
func (e *percentileOriginal4Time) AppendFinalResult2Chunk(sctx sessionctx.Context, pr PartialResult, chk *chunk.Chunk) error {
|
|
p := (*partialResult4PercentileTime)(pr)
|
|
if len(*p) == 0 {
|
|
chk.AppendNull(e.ordinal)
|
|
return nil
|
|
}
|
|
index := percentile(p, e.percent)
|
|
chk.AppendTime(e.ordinal, (*p)[index])
|
|
return nil
|
|
}
|
|
|
|
type percentileOriginal4Duration struct {
|
|
basePercentile
|
|
}
|
|
|
|
func (e *percentileOriginal4Duration) AllocPartialResult() (pr PartialResult) {
|
|
// TODO: Preserve appropriate capacity for data
|
|
pr = PartialResult(&partialResult4PercentileTime{})
|
|
return pr
|
|
}
|
|
|
|
func (e *percentileOriginal4Duration) ResetPartialResult(pr PartialResult) {
|
|
p := (*partialResult4PercentileDuration)(pr)
|
|
*p = partialResult4PercentileDuration{}
|
|
}
|
|
|
|
func (e *percentileOriginal4Duration) UpdatePartialResult(sctx sessionctx.Context, rowsInGroup []chunk.Row, pr PartialResult) (err error) {
|
|
p := (*partialResult4PercentileDuration)(pr)
|
|
for _, row := range rowsInGroup {
|
|
v, isNull, err := e.args[0].EvalDuration(sctx, row)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if isNull {
|
|
continue
|
|
}
|
|
*p = append(*p, v)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (e *percentileOriginal4Duration) MergePartialResult(sctx sessionctx.Context, src, dst PartialResult) (err error) {
|
|
p1, p2 := (*partialResult4PercentileDuration)(src), (*partialResult4PercentileDuration)(dst)
|
|
mergeBuff := make(partialResult4PercentileDuration, len(*p1)+len(*p2))
|
|
copy(mergeBuff, *p2)
|
|
copy(mergeBuff[len(*p2):], *p1)
|
|
*p1 = nil
|
|
*p2 = mergeBuff
|
|
return nil
|
|
}
|
|
|
|
func (e *percentileOriginal4Duration) AppendFinalResult2Chunk(sctx sessionctx.Context, pr PartialResult, chk *chunk.Chunk) error {
|
|
p := (*partialResult4PercentileDuration)(pr)
|
|
if len(*p) == 0 {
|
|
chk.AppendNull(e.ordinal)
|
|
return nil
|
|
}
|
|
index := percentile(*p, e.percent)
|
|
|
|
chk.AppendDuration(e.ordinal, (*p)[index])
|
|
return nil
|
|
}
|
|
|