// 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 }