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.
350 lines
8.2 KiB
350 lines
8.2 KiB
// Copyright 2019 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 (
|
|
"github.com/pingcap/parser/mysql"
|
|
"github.com/pingcap/tidb/expression"
|
|
"github.com/pingcap/tidb/sessionctx"
|
|
"github.com/pingcap/tidb/types"
|
|
"github.com/pingcap/tidb/types/json"
|
|
"github.com/pingcap/tidb/util/chunk"
|
|
)
|
|
|
|
// valueEvaluator is used to evaluate values for `first_value`, `last_value`, `nth_value`,
|
|
// `lead` and `lag`.
|
|
type valueEvaluator interface {
|
|
// evaluateRow evaluates the expression using row and stores the result inside.
|
|
evaluateRow(ctx sessionctx.Context, expr expression.Expression, row chunk.Row) error
|
|
// appendResult appends the result to chunk.
|
|
appendResult(chk *chunk.Chunk, colIdx int)
|
|
}
|
|
|
|
type value4Int struct {
|
|
val int64
|
|
isNull bool
|
|
}
|
|
|
|
func (v *value4Int) evaluateRow(ctx sessionctx.Context, expr expression.Expression, row chunk.Row) error {
|
|
var err error
|
|
v.val, v.isNull, err = expr.EvalInt(ctx, row)
|
|
return err
|
|
}
|
|
|
|
func (v *value4Int) appendResult(chk *chunk.Chunk, colIdx int) {
|
|
if v.isNull {
|
|
chk.AppendNull(colIdx)
|
|
} else {
|
|
chk.AppendInt64(colIdx, v.val)
|
|
}
|
|
}
|
|
|
|
type value4Float32 struct {
|
|
val float32
|
|
isNull bool
|
|
}
|
|
|
|
func (v *value4Float32) evaluateRow(ctx sessionctx.Context, expr expression.Expression, row chunk.Row) error {
|
|
var err error
|
|
var val float64
|
|
val, v.isNull, err = expr.EvalReal(ctx, row)
|
|
v.val = float32(val)
|
|
return err
|
|
}
|
|
|
|
func (v *value4Float32) appendResult(chk *chunk.Chunk, colIdx int) {
|
|
if v.isNull {
|
|
chk.AppendNull(colIdx)
|
|
} else {
|
|
chk.AppendFloat32(colIdx, v.val)
|
|
}
|
|
}
|
|
|
|
type value4Decimal struct {
|
|
val *types.MyDecimal
|
|
isNull bool
|
|
}
|
|
|
|
func (v *value4Decimal) evaluateRow(ctx sessionctx.Context, expr expression.Expression, row chunk.Row) error {
|
|
var err error
|
|
v.val, v.isNull, err = expr.EvalDecimal(ctx, row)
|
|
return err
|
|
}
|
|
|
|
func (v *value4Decimal) appendResult(chk *chunk.Chunk, colIdx int) {
|
|
if v.isNull {
|
|
chk.AppendNull(colIdx)
|
|
} else {
|
|
chk.AppendMyDecimal(colIdx, v.val)
|
|
}
|
|
}
|
|
|
|
type value4Float64 struct {
|
|
val float64
|
|
isNull bool
|
|
}
|
|
|
|
func (v *value4Float64) evaluateRow(ctx sessionctx.Context, expr expression.Expression, row chunk.Row) error {
|
|
var err error
|
|
v.val, v.isNull, err = expr.EvalReal(ctx, row)
|
|
return err
|
|
}
|
|
|
|
func (v *value4Float64) appendResult(chk *chunk.Chunk, colIdx int) {
|
|
if v.isNull {
|
|
chk.AppendNull(colIdx)
|
|
} else {
|
|
chk.AppendFloat64(colIdx, v.val)
|
|
}
|
|
}
|
|
|
|
type value4String struct {
|
|
val string
|
|
isNull bool
|
|
}
|
|
|
|
func (v *value4String) evaluateRow(ctx sessionctx.Context, expr expression.Expression, row chunk.Row) error {
|
|
var err error
|
|
v.val, v.isNull, err = expr.EvalString(ctx, row)
|
|
return err
|
|
}
|
|
|
|
func (v *value4String) appendResult(chk *chunk.Chunk, colIdx int) {
|
|
if v.isNull {
|
|
chk.AppendNull(colIdx)
|
|
} else {
|
|
chk.AppendString(colIdx, v.val)
|
|
}
|
|
}
|
|
|
|
type value4Time struct {
|
|
val types.Time
|
|
isNull bool
|
|
}
|
|
|
|
func (v *value4Time) evaluateRow(ctx sessionctx.Context, expr expression.Expression, row chunk.Row) error {
|
|
var err error
|
|
v.val, v.isNull, err = expr.EvalTime(ctx, row)
|
|
return err
|
|
}
|
|
|
|
func (v *value4Time) appendResult(chk *chunk.Chunk, colIdx int) {
|
|
if v.isNull {
|
|
chk.AppendNull(colIdx)
|
|
} else {
|
|
chk.AppendTime(colIdx, v.val)
|
|
}
|
|
}
|
|
|
|
type value4Duration struct {
|
|
val types.Duration
|
|
isNull bool
|
|
}
|
|
|
|
func (v *value4Duration) evaluateRow(ctx sessionctx.Context, expr expression.Expression, row chunk.Row) error {
|
|
var err error
|
|
v.val, v.isNull, err = expr.EvalDuration(ctx, row)
|
|
return err
|
|
}
|
|
|
|
func (v *value4Duration) appendResult(chk *chunk.Chunk, colIdx int) {
|
|
if v.isNull {
|
|
chk.AppendNull(colIdx)
|
|
} else {
|
|
chk.AppendDuration(colIdx, v.val)
|
|
}
|
|
}
|
|
|
|
type value4JSON struct {
|
|
val json.BinaryJSON
|
|
isNull bool
|
|
}
|
|
|
|
func (v *value4JSON) evaluateRow(ctx sessionctx.Context, expr expression.Expression, row chunk.Row) error {
|
|
var err error
|
|
v.val, v.isNull, err = expr.EvalJSON(ctx, row)
|
|
v.val = v.val.Copy() // deep copy to avoid content change.
|
|
return err
|
|
}
|
|
|
|
func (v *value4JSON) appendResult(chk *chunk.Chunk, colIdx int) {
|
|
if v.isNull {
|
|
chk.AppendNull(colIdx)
|
|
} else {
|
|
chk.AppendJSON(colIdx, v.val)
|
|
}
|
|
}
|
|
|
|
func buildValueEvaluator(tp *types.FieldType) valueEvaluator {
|
|
evalType := tp.EvalType()
|
|
if tp.Tp == mysql.TypeBit {
|
|
evalType = types.ETString
|
|
}
|
|
switch evalType {
|
|
case types.ETInt:
|
|
return &value4Int{}
|
|
case types.ETReal:
|
|
switch tp.Tp {
|
|
case mysql.TypeFloat:
|
|
return &value4Float32{}
|
|
case mysql.TypeDouble:
|
|
return &value4Float64{}
|
|
}
|
|
case types.ETDecimal:
|
|
return &value4Decimal{}
|
|
case types.ETDatetime, types.ETTimestamp:
|
|
return &value4Time{}
|
|
case types.ETDuration:
|
|
return &value4Duration{}
|
|
case types.ETString:
|
|
return &value4String{}
|
|
case types.ETJson:
|
|
return &value4JSON{}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type firstValue struct {
|
|
baseAggFunc
|
|
|
|
tp *types.FieldType
|
|
}
|
|
|
|
type partialResult4FirstValue struct {
|
|
gotFirstValue bool
|
|
evaluator valueEvaluator
|
|
}
|
|
|
|
func (v *firstValue) AllocPartialResult() PartialResult {
|
|
return PartialResult(&partialResult4FirstValue{evaluator: buildValueEvaluator(v.tp)})
|
|
}
|
|
|
|
func (v *firstValue) ResetPartialResult(pr PartialResult) {
|
|
p := (*partialResult4FirstValue)(pr)
|
|
p.gotFirstValue = false
|
|
}
|
|
|
|
func (v *firstValue) UpdatePartialResult(sctx sessionctx.Context, rowsInGroup []chunk.Row, pr PartialResult) error {
|
|
p := (*partialResult4FirstValue)(pr)
|
|
if p.gotFirstValue {
|
|
return nil
|
|
}
|
|
if len(rowsInGroup) > 0 {
|
|
p.gotFirstValue = true
|
|
err := p.evaluator.evaluateRow(sctx, v.args[0], rowsInGroup[0])
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (v *firstValue) AppendFinalResult2Chunk(sctx sessionctx.Context, pr PartialResult, chk *chunk.Chunk) error {
|
|
p := (*partialResult4FirstValue)(pr)
|
|
if !p.gotFirstValue {
|
|
chk.AppendNull(v.ordinal)
|
|
} else {
|
|
p.evaluator.appendResult(chk, v.ordinal)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type lastValue struct {
|
|
baseAggFunc
|
|
|
|
tp *types.FieldType
|
|
}
|
|
|
|
type partialResult4LastValue struct {
|
|
gotLastValue bool
|
|
evaluator valueEvaluator
|
|
}
|
|
|
|
func (v *lastValue) AllocPartialResult() PartialResult {
|
|
return PartialResult(&partialResult4LastValue{evaluator: buildValueEvaluator(v.tp)})
|
|
}
|
|
|
|
func (v *lastValue) ResetPartialResult(pr PartialResult) {
|
|
p := (*partialResult4LastValue)(pr)
|
|
p.gotLastValue = false
|
|
}
|
|
|
|
func (v *lastValue) UpdatePartialResult(sctx sessionctx.Context, rowsInGroup []chunk.Row, pr PartialResult) error {
|
|
p := (*partialResult4LastValue)(pr)
|
|
if len(rowsInGroup) > 0 {
|
|
p.gotLastValue = true
|
|
err := p.evaluator.evaluateRow(sctx, v.args[0], rowsInGroup[len(rowsInGroup)-1])
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (v *lastValue) AppendFinalResult2Chunk(sctx sessionctx.Context, pr PartialResult, chk *chunk.Chunk) error {
|
|
p := (*partialResult4LastValue)(pr)
|
|
if !p.gotLastValue {
|
|
chk.AppendNull(v.ordinal)
|
|
} else {
|
|
p.evaluator.appendResult(chk, v.ordinal)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type nthValue struct {
|
|
baseAggFunc
|
|
|
|
tp *types.FieldType
|
|
nth uint64
|
|
}
|
|
|
|
type partialResult4NthValue struct {
|
|
seenRows uint64
|
|
evaluator valueEvaluator
|
|
}
|
|
|
|
func (v *nthValue) AllocPartialResult() PartialResult {
|
|
return PartialResult(&partialResult4NthValue{evaluator: buildValueEvaluator(v.tp)})
|
|
}
|
|
|
|
func (v *nthValue) ResetPartialResult(pr PartialResult) {
|
|
p := (*partialResult4NthValue)(pr)
|
|
p.seenRows = 0
|
|
}
|
|
|
|
func (v *nthValue) UpdatePartialResult(sctx sessionctx.Context, rowsInGroup []chunk.Row, pr PartialResult) error {
|
|
if v.nth == 0 {
|
|
return nil
|
|
}
|
|
p := (*partialResult4NthValue)(pr)
|
|
numRows := uint64(len(rowsInGroup))
|
|
if v.nth > p.seenRows && v.nth-p.seenRows <= numRows {
|
|
err := p.evaluator.evaluateRow(sctx, v.args[0], rowsInGroup[v.nth-p.seenRows-1])
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
p.seenRows += numRows
|
|
return nil
|
|
}
|
|
|
|
func (v *nthValue) AppendFinalResult2Chunk(sctx sessionctx.Context, pr PartialResult, chk *chunk.Chunk) error {
|
|
p := (*partialResult4NthValue)(pr)
|
|
if v.nth == 0 || p.seenRows < v.nth {
|
|
chk.AppendNull(v.ordinal)
|
|
} else {
|
|
p.evaluator.appendResult(chk, v.ordinal)
|
|
}
|
|
return nil
|
|
}
|
|
|