vsphere-influxdb-go/vendor/github.com/influxdata/influxdb/influxql/select.go

1418 lines
37 KiB
Go

package influxql
import (
"errors"
"fmt"
"math"
"sort"
"time"
)
// SelectOptions are options that customize the select call.
type SelectOptions struct {
// Authorizer is used to limit access to data
Authorizer Authorizer
// The lower bound for a select call.
MinTime time.Time
// The upper bound for a select call.
MaxTime time.Time
// Node to exclusively read from.
// If zero, all nodes are used.
NodeID uint64
// An optional channel that, if closed, signals that the select should be
// interrupted.
InterruptCh <-chan struct{}
// Maximum number of concurrent series.
MaxSeriesN int
}
// Select executes stmt against ic and returns a list of iterators to stream from.
//
// Statements should have all rewriting performed before calling select(). This
// includes wildcard and source expansion.
func Select(stmt *SelectStatement, ic IteratorCreator, sopt *SelectOptions) ([]Iterator, error) {
// Determine base options for iterators.
opt, err := newIteratorOptionsStmt(stmt, sopt)
if err != nil {
return nil, err
}
return buildIterators(stmt, ic, opt)
}
func buildIterators(stmt *SelectStatement, ic IteratorCreator, opt IteratorOptions) ([]Iterator, error) {
// Retrieve refs for each call and var ref.
info := newSelectInfo(stmt)
if len(info.calls) > 1 && len(info.refs) > 0 {
return nil, errors.New("cannot select fields when selecting multiple aggregates")
}
// Determine auxiliary fields to be selected.
opt.Aux = make([]VarRef, 0, len(info.refs))
for ref := range info.refs {
opt.Aux = append(opt.Aux, *ref)
}
sort.Sort(VarRefs(opt.Aux))
// If there are multiple auxilary fields and no calls then construct an aux iterator.
if len(info.calls) == 0 && len(info.refs) > 0 {
return buildAuxIterators(stmt.Fields, ic, stmt.Sources, opt)
}
// Include auxiliary fields from top() and bottom() when not writing the results.
fields := stmt.Fields
if stmt.Target == nil {
extraFields := 0
for call := range info.calls {
if call.Name == "top" || call.Name == "bottom" {
for i := 1; i < len(call.Args)-1; i++ {
ref := call.Args[i].(*VarRef)
opt.Aux = append(opt.Aux, *ref)
extraFields++
}
}
}
if extraFields > 0 {
// Rebuild the list of fields if any extra fields are being implicitly added
fields = make([]*Field, 0, len(stmt.Fields)+extraFields)
for _, f := range stmt.Fields {
fields = append(fields, f)
switch expr := f.Expr.(type) {
case *Call:
if expr.Name == "top" || expr.Name == "bottom" {
for i := 1; i < len(expr.Args)-1; i++ {
fields = append(fields, &Field{Expr: expr.Args[i]})
}
}
}
}
}
}
// Determine if there is one call and it is a selector.
selector := false
if len(info.calls) == 1 {
for call := range info.calls {
selector = IsSelector(call)
}
}
return buildFieldIterators(fields, ic, stmt.Sources, opt, selector, stmt.Target != nil)
}
// buildAuxIterators creates a set of iterators from a single combined auxiliary iterator.
func buildAuxIterators(fields Fields, ic IteratorCreator, sources Sources, opt IteratorOptions) ([]Iterator, error) {
// Create the auxiliary iterators for each source.
inputs := make([]Iterator, 0, len(sources))
if err := func() error {
for _, source := range sources {
switch source := source.(type) {
case *Measurement:
input, err := ic.CreateIterator(source, opt)
if err != nil {
return err
}
inputs = append(inputs, input)
case *SubQuery:
b := subqueryBuilder{
ic: ic,
stmt: source.Statement,
}
input, err := b.buildAuxIterator(opt)
if err != nil {
return err
}
inputs = append(inputs, input)
}
}
return nil
}(); err != nil {
Iterators(inputs).Close()
return nil, err
}
// Merge iterators to read auxilary fields.
input, err := Iterators(inputs).Merge(opt)
if err != nil {
Iterators(inputs).Close()
return nil, err
} else if input == nil {
input = &nilFloatIterator{}
}
// Filter out duplicate rows, if required.
if opt.Dedupe {
// If there is no group by and it is a float iterator, see if we can use a fast dedupe.
if itr, ok := input.(FloatIterator); ok && len(opt.Dimensions) == 0 {
if sz := len(fields); sz > 0 && sz < 3 {
input = newFloatFastDedupeIterator(itr)
} else {
input = NewDedupeIterator(itr)
}
} else {
input = NewDedupeIterator(input)
}
}
// Apply limit & offset.
if opt.Limit > 0 || opt.Offset > 0 {
input = NewLimitIterator(input, opt)
}
// Wrap in an auxiliary iterator to separate the fields.
aitr := NewAuxIterator(input, opt)
// Generate iterators for each field.
itrs := make([]Iterator, len(fields))
if err := func() error {
for i, f := range fields {
expr := Reduce(f.Expr, nil)
itr, err := buildAuxIterator(expr, aitr, opt)
if err != nil {
return err
}
itrs[i] = itr
}
return nil
}(); err != nil {
Iterators(Iterators(itrs).filterNonNil()).Close()
aitr.Close()
return nil, err
}
// Background the primary iterator since there is no reader for it.
aitr.Background()
return itrs, nil
}
// buildAuxIterator constructs an Iterator for an expression from an AuxIterator.
func buildAuxIterator(expr Expr, aitr AuxIterator, opt IteratorOptions) (Iterator, error) {
switch expr := expr.(type) {
case *VarRef:
return aitr.Iterator(expr.Val, expr.Type), nil
case *BinaryExpr:
if rhs, ok := expr.RHS.(Literal); ok {
// The right hand side is a literal. It is more common to have the RHS be a literal,
// so we check that one first and have this be the happy path.
if lhs, ok := expr.LHS.(Literal); ok {
// We have two literals that couldn't be combined by Reduce.
return nil, fmt.Errorf("unable to construct an iterator from two literals: LHS: %T, RHS: %T", lhs, rhs)
}
lhs, err := buildAuxIterator(expr.LHS, aitr, opt)
if err != nil {
return nil, err
}
return buildRHSTransformIterator(lhs, rhs, expr.Op, opt)
} else if lhs, ok := expr.LHS.(Literal); ok {
rhs, err := buildAuxIterator(expr.RHS, aitr, opt)
if err != nil {
return nil, err
}
return buildLHSTransformIterator(lhs, rhs, expr.Op, opt)
} else {
// We have two iterators. Combine them into a single iterator.
lhs, err := buildAuxIterator(expr.LHS, aitr, opt)
if err != nil {
return nil, err
}
rhs, err := buildAuxIterator(expr.RHS, aitr, opt)
if err != nil {
return nil, err
}
return buildTransformIterator(lhs, rhs, expr.Op, opt)
}
case *ParenExpr:
return buildAuxIterator(expr.Expr, aitr, opt)
case *nilLiteral:
return &nilFloatIterator{}, nil
default:
return nil, fmt.Errorf("invalid expression type: %T", expr)
}
}
// buildFieldIterators creates an iterator for each field expression.
func buildFieldIterators(fields Fields, ic IteratorCreator, sources Sources, opt IteratorOptions, selector, writeMode bool) ([]Iterator, error) {
// Create iterators from fields against the iterator creator.
itrs := make([]Iterator, len(fields))
if err := func() error {
hasAuxFields := false
var input Iterator
for i, f := range fields {
// Build iterators for calls first and save the iterator.
// We do this so we can keep the ordering provided by the user, but
// still build the Call's iterator first.
if ContainsVarRef(f.Expr) {
hasAuxFields = true
continue
}
expr := Reduce(f.Expr, nil)
itr, err := buildExprIterator(expr, ic, sources, opt, selector, writeMode)
if err != nil {
return err
} else if itr == nil {
itr = &nilFloatIterator{}
}
// If there is a limit or offset then apply it.
if opt.Limit > 0 || opt.Offset > 0 {
itr = NewLimitIterator(itr, opt)
}
itrs[i] = itr
input = itr
}
if input == nil || !hasAuxFields {
return nil
}
// Build the aux iterators. Previous validation should ensure that only one
// call was present so we build an AuxIterator from that input.
aitr := NewAuxIterator(input, opt)
for i, f := range fields {
if itrs[i] != nil {
itrs[i] = aitr
continue
}
expr := Reduce(f.Expr, nil)
itr, err := buildAuxIterator(expr, aitr, opt)
if err != nil {
return err
} else if itr == nil {
itr = &nilFloatIterator{}
}
itrs[i] = itr
}
aitr.Start()
return nil
}(); err != nil {
Iterators(Iterators(itrs).filterNonNil()).Close()
return nil, err
}
return itrs, nil
}
// buildExprIterator creates an iterator for an expression.
func buildExprIterator(expr Expr, ic IteratorCreator, sources Sources, opt IteratorOptions, selector, writeMode bool) (Iterator, error) {
opt.Expr = expr
b := exprIteratorBuilder{
ic: ic,
sources: sources,
opt: opt,
selector: selector,
writeMode: writeMode,
}
switch expr := expr.(type) {
case *VarRef:
return b.buildVarRefIterator(expr)
case *Call:
return b.buildCallIterator(expr)
case *BinaryExpr:
return b.buildBinaryExprIterator(expr)
case *ParenExpr:
return buildExprIterator(expr.Expr, ic, sources, opt, selector, writeMode)
case *nilLiteral:
return &nilFloatIterator{}, nil
default:
return nil, fmt.Errorf("invalid expression type: %T", expr)
}
}
type exprIteratorBuilder struct {
ic IteratorCreator
sources Sources
opt IteratorOptions
selector bool
writeMode bool
}
func (b *exprIteratorBuilder) buildVarRefIterator(expr *VarRef) (Iterator, error) {
inputs := make([]Iterator, 0, len(b.sources))
if err := func() error {
for _, source := range b.sources {
switch source := source.(type) {
case *Measurement:
input, err := b.ic.CreateIterator(source, b.opt)
if err != nil {
return err
}
inputs = append(inputs, input)
case *SubQuery:
subquery := subqueryBuilder{
ic: b.ic,
stmt: source.Statement,
}
input, err := subquery.buildVarRefIterator(expr, b.opt)
if err != nil {
return err
}
inputs = append(inputs, input)
}
}
return nil
}(); err != nil {
Iterators(inputs).Close()
return nil, err
}
// Variable references in this section will always go into some call
// iterator. Combine it with a merge iterator.
itr := NewMergeIterator(inputs, b.opt)
if itr == nil {
itr = &nilFloatIterator{}
}
if b.opt.InterruptCh != nil {
itr = NewInterruptIterator(itr, b.opt.InterruptCh)
}
return itr, nil
}
func (b *exprIteratorBuilder) buildCallIterator(expr *Call) (Iterator, error) {
// TODO(jsternberg): Refactor this. This section needs to die in a fire.
opt := b.opt
// Eliminate limits and offsets if they were previously set. These are handled by the caller.
opt.Limit, opt.Offset = 0, 0
switch expr.Name {
case "distinct":
opt.Ordered = true
input, err := buildExprIterator(expr.Args[0].(*VarRef), b.ic, b.sources, opt, b.selector, false)
if err != nil {
return nil, err
}
input, err = NewDistinctIterator(input, opt)
if err != nil {
return nil, err
}
return NewIntervalIterator(input, opt), nil
case "sample":
opt.Ordered = true
input, err := buildExprIterator(expr.Args[0], b.ic, b.sources, opt, b.selector, false)
if err != nil {
return nil, err
}
size := expr.Args[1].(*IntegerLiteral)
return newSampleIterator(input, opt, int(size.Val))
case "holt_winters", "holt_winters_with_fit":
opt.Ordered = true
input, err := buildExprIterator(expr.Args[0], b.ic, b.sources, opt, b.selector, false)
if err != nil {
return nil, err
}
h := expr.Args[1].(*IntegerLiteral)
m := expr.Args[2].(*IntegerLiteral)
includeFitData := "holt_winters_with_fit" == expr.Name
interval := opt.Interval.Duration
// Redefine interval to be unbounded to capture all aggregate results
opt.StartTime = MinTime
opt.EndTime = MaxTime
opt.Interval = Interval{}
return newHoltWintersIterator(input, opt, int(h.Val), int(m.Val), includeFitData, interval)
case "derivative", "non_negative_derivative", "difference", "non_negative_difference", "moving_average", "elapsed":
if !opt.Interval.IsZero() {
if opt.Ascending {
opt.StartTime -= int64(opt.Interval.Duration)
} else {
opt.EndTime += int64(opt.Interval.Duration)
}
}
opt.Ordered = true
input, err := buildExprIterator(expr.Args[0], b.ic, b.sources, opt, b.selector, false)
if err != nil {
return nil, err
}
switch expr.Name {
case "derivative", "non_negative_derivative":
interval := opt.DerivativeInterval()
isNonNegative := (expr.Name == "non_negative_derivative")
return newDerivativeIterator(input, opt, interval, isNonNegative)
case "elapsed":
interval := opt.ElapsedInterval()
return newElapsedIterator(input, opt, interval)
case "difference", "non_negative_difference":
isNonNegative := (expr.Name == "non_negative_difference")
return newDifferenceIterator(input, opt, isNonNegative)
case "moving_average":
n := expr.Args[1].(*IntegerLiteral)
if n.Val > 1 && !opt.Interval.IsZero() {
if opt.Ascending {
opt.StartTime -= int64(opt.Interval.Duration) * (n.Val - 1)
} else {
opt.EndTime += int64(opt.Interval.Duration) * (n.Val - 1)
}
}
return newMovingAverageIterator(input, int(n.Val), opt)
}
panic(fmt.Sprintf("invalid series aggregate function: %s", expr.Name))
case "cumulative_sum":
opt.Ordered = true
input, err := buildExprIterator(expr.Args[0], b.ic, b.sources, opt, b.selector, false)
if err != nil {
return nil, err
}
return newCumulativeSumIterator(input, opt)
case "integral":
opt.Ordered = true
input, err := buildExprIterator(expr.Args[0].(*VarRef), b.ic, b.sources, opt, false, false)
if err != nil {
return nil, err
}
interval := opt.IntegralInterval()
return newIntegralIterator(input, opt, interval)
case "top":
if len(expr.Args) < 2 {
return nil, fmt.Errorf("top() requires 2 or more arguments, got %d", len(expr.Args))
}
var input Iterator
if len(expr.Args) > 2 {
// Create a max iterator using the groupings in the arguments.
dims := make(map[string]struct{}, len(expr.Args)-2+len(opt.GroupBy))
for i := 1; i < len(expr.Args)-1; i++ {
ref := expr.Args[i].(*VarRef)
dims[ref.Val] = struct{}{}
}
for dim := range opt.GroupBy {
dims[dim] = struct{}{}
}
call := &Call{
Name: "max",
Args: expr.Args[:1],
}
callOpt := opt
callOpt.Expr = call
callOpt.GroupBy = dims
callOpt.Fill = NoFill
builder := *b
builder.opt = callOpt
builder.selector = true
builder.writeMode = false
i, err := builder.callIterator(call, callOpt)
if err != nil {
return nil, err
}
input = i
} else {
// There are no arguments so do not organize the points by tags.
builder := *b
builder.opt.Expr = expr.Args[0]
builder.selector = true
builder.writeMode = false
ref := expr.Args[0].(*VarRef)
i, err := builder.buildVarRefIterator(ref)
if err != nil {
return nil, err
}
input = i
}
n := expr.Args[len(expr.Args)-1].(*IntegerLiteral)
return newTopIterator(input, opt, int(n.Val), b.writeMode)
case "bottom":
if len(expr.Args) < 2 {
return nil, fmt.Errorf("bottom() requires 2 or more arguments, got %d", len(expr.Args))
}
var input Iterator
if len(expr.Args) > 2 {
// Create a max iterator using the groupings in the arguments.
dims := make(map[string]struct{}, len(expr.Args)-2)
for i := 1; i < len(expr.Args)-1; i++ {
ref := expr.Args[i].(*VarRef)
dims[ref.Val] = struct{}{}
}
for dim := range opt.GroupBy {
dims[dim] = struct{}{}
}
call := &Call{
Name: "min",
Args: expr.Args[:1],
}
callOpt := opt
callOpt.Expr = call
callOpt.GroupBy = dims
callOpt.Fill = NoFill
builder := *b
builder.opt = callOpt
builder.selector = true
builder.writeMode = false
i, err := builder.callIterator(call, callOpt)
if err != nil {
return nil, err
}
input = i
} else {
// There are no arguments so do not organize the points by tags.
builder := *b
builder.opt.Expr = expr.Args[0]
builder.selector = true
builder.writeMode = false
ref := expr.Args[0].(*VarRef)
i, err := builder.buildVarRefIterator(ref)
if err != nil {
return nil, err
}
input = i
}
n := expr.Args[len(expr.Args)-1].(*IntegerLiteral)
return newBottomIterator(input, b.opt, int(n.Val), b.writeMode)
}
itr, err := func() (Iterator, error) {
switch expr.Name {
case "count":
switch arg0 := expr.Args[0].(type) {
case *Call:
if arg0.Name == "distinct" {
input, err := buildExprIterator(arg0, b.ic, b.sources, opt, b.selector, false)
if err != nil {
return nil, err
}
return newCountIterator(input, opt)
}
}
fallthrough
case "min", "max", "sum", "first", "last", "mean":
return b.callIterator(expr, opt)
case "median":
opt.Ordered = true
input, err := buildExprIterator(expr.Args[0].(*VarRef), b.ic, b.sources, opt, false, false)
if err != nil {
return nil, err
}
return newMedianIterator(input, opt)
case "mode":
input, err := buildExprIterator(expr.Args[0].(*VarRef), b.ic, b.sources, opt, false, false)
if err != nil {
return nil, err
}
return NewModeIterator(input, opt)
case "stddev":
input, err := buildExprIterator(expr.Args[0].(*VarRef), b.ic, b.sources, opt, false, false)
if err != nil {
return nil, err
}
return newStddevIterator(input, opt)
case "spread":
// OPTIMIZE(benbjohnson): convert to map/reduce
input, err := buildExprIterator(expr.Args[0].(*VarRef), b.ic, b.sources, opt, false, false)
if err != nil {
return nil, err
}
return newSpreadIterator(input, opt)
case "percentile":
opt.Ordered = true
input, err := buildExprIterator(expr.Args[0].(*VarRef), b.ic, b.sources, opt, false, false)
if err != nil {
return nil, err
}
var percentile float64
switch arg := expr.Args[1].(type) {
case *NumberLiteral:
percentile = arg.Val
case *IntegerLiteral:
percentile = float64(arg.Val)
}
return newPercentileIterator(input, opt, percentile)
default:
return nil, fmt.Errorf("unsupported call: %s", expr.Name)
}
}()
if err != nil {
return nil, err
}
if !b.selector || !opt.Interval.IsZero() {
itr = NewIntervalIterator(itr, opt)
if !opt.Interval.IsZero() && opt.Fill != NoFill {
itr = NewFillIterator(itr, expr, opt)
}
}
if opt.InterruptCh != nil {
itr = NewInterruptIterator(itr, opt.InterruptCh)
}
return itr, nil
}
func (b *exprIteratorBuilder) buildBinaryExprIterator(expr *BinaryExpr) (Iterator, error) {
if rhs, ok := expr.RHS.(Literal); ok {
// The right hand side is a literal. It is more common to have the RHS be a literal,
// so we check that one first and have this be the happy path.
if lhs, ok := expr.LHS.(Literal); ok {
// We have two literals that couldn't be combined by Reduce.
return nil, fmt.Errorf("unable to construct an iterator from two literals: LHS: %T, RHS: %T", lhs, rhs)
}
lhs, err := buildExprIterator(expr.LHS, b.ic, b.sources, b.opt, b.selector, false)
if err != nil {
return nil, err
}
return buildRHSTransformIterator(lhs, rhs, expr.Op, b.opt)
} else if lhs, ok := expr.LHS.(Literal); ok {
rhs, err := buildExprIterator(expr.RHS, b.ic, b.sources, b.opt, b.selector, false)
if err != nil {
return nil, err
}
return buildLHSTransformIterator(lhs, rhs, expr.Op, b.opt)
} else {
// We have two iterators. Combine them into a single iterator.
lhs, err := buildExprIterator(expr.LHS, b.ic, b.sources, b.opt, false, false)
if err != nil {
return nil, err
}
rhs, err := buildExprIterator(expr.RHS, b.ic, b.sources, b.opt, false, false)
if err != nil {
return nil, err
}
return buildTransformIterator(lhs, rhs, expr.Op, b.opt)
}
}
func (b *exprIteratorBuilder) callIterator(expr *Call, opt IteratorOptions) (Iterator, error) {
inputs := make([]Iterator, 0, len(b.sources))
if err := func() error {
for _, source := range b.sources {
switch source := source.(type) {
case *Measurement:
input, err := b.ic.CreateIterator(source, opt)
if err != nil {
return err
}
inputs = append(inputs, input)
case *SubQuery:
// Identify the name of the field we are using.
arg0 := expr.Args[0].(*VarRef)
input, err := buildExprIterator(arg0, b.ic, []Source{source}, opt, b.selector, false)
if err != nil {
return err
}
// Wrap the result in a call iterator.
i, err := NewCallIterator(input, opt)
if err != nil {
input.Close()
return err
}
inputs = append(inputs, i)
}
}
return nil
}(); err != nil {
Iterators(inputs).Close()
return nil, err
}
itr, err := Iterators(inputs).Merge(opt)
if err != nil {
Iterators(inputs).Close()
return nil, err
} else if itr == nil {
itr = &nilFloatIterator{}
}
return itr, nil
}
func buildRHSTransformIterator(lhs Iterator, rhs Literal, op Token, opt IteratorOptions) (Iterator, error) {
fn := binaryExprFunc(iteratorDataType(lhs), literalDataType(rhs), op)
switch fn := fn.(type) {
case func(float64, float64) float64:
var input FloatIterator
switch lhs := lhs.(type) {
case FloatIterator:
input = lhs
case IntegerIterator:
input = &integerFloatCastIterator{input: lhs}
default:
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as a FloatIterator", lhs)
}
var val float64
switch rhs := rhs.(type) {
case *NumberLiteral:
val = rhs.Val
case *IntegerLiteral:
val = float64(rhs.Val)
default:
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as a NumberLiteral", rhs)
}
return &floatTransformIterator{
input: input,
fn: func(p *FloatPoint) *FloatPoint {
if p == nil {
return nil
} else if p.Nil {
return p
}
p.Value = fn(p.Value, val)
return p
},
}, nil
case func(int64, int64) float64:
input, ok := lhs.(IntegerIterator)
if !ok {
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as a IntegerIterator", lhs)
}
var val int64
switch rhs := rhs.(type) {
case *IntegerLiteral:
val = rhs.Val
default:
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as a IntegerLiteral", rhs)
}
return &integerFloatTransformIterator{
input: input,
fn: func(p *IntegerPoint) *FloatPoint {
if p == nil {
return nil
}
fp := &FloatPoint{
Name: p.Name,
Tags: p.Tags,
Time: p.Time,
Aux: p.Aux,
}
if p.Nil {
fp.Nil = true
} else {
fp.Value = fn(p.Value, val)
}
return fp
},
}, nil
case func(float64, float64) bool:
var input FloatIterator
switch lhs := lhs.(type) {
case FloatIterator:
input = lhs
case IntegerIterator:
input = &integerFloatCastIterator{input: lhs}
default:
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as a FloatIterator", lhs)
}
var val float64
switch rhs := rhs.(type) {
case *NumberLiteral:
val = rhs.Val
case *IntegerLiteral:
val = float64(rhs.Val)
default:
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as a NumberLiteral", rhs)
}
return &floatBoolTransformIterator{
input: input,
fn: func(p *FloatPoint) *BooleanPoint {
if p == nil {
return nil
}
bp := &BooleanPoint{
Name: p.Name,
Tags: p.Tags,
Time: p.Time,
Aux: p.Aux,
}
if p.Nil {
bp.Nil = true
} else {
bp.Value = fn(p.Value, val)
}
return bp
},
}, nil
case func(int64, int64) int64:
var input IntegerIterator
switch lhs := lhs.(type) {
case IntegerIterator:
input = lhs
default:
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as an IntegerIterator", lhs)
}
var val int64
switch rhs := rhs.(type) {
case *IntegerLiteral:
val = rhs.Val
default:
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as an IntegerLiteral", rhs)
}
return &integerTransformIterator{
input: input,
fn: func(p *IntegerPoint) *IntegerPoint {
if p == nil {
return nil
} else if p.Nil {
return p
}
p.Value = fn(p.Value, val)
return p
},
}, nil
case func(int64, int64) bool:
var input IntegerIterator
switch lhs := lhs.(type) {
case IntegerIterator:
input = lhs
default:
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as an IntegerIterator", lhs)
}
var val int64
switch rhs := rhs.(type) {
case *IntegerLiteral:
val = rhs.Val
default:
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as an IntegerLiteral", rhs)
}
return &integerBoolTransformIterator{
input: input,
fn: func(p *IntegerPoint) *BooleanPoint {
if p == nil {
return nil
}
bp := &BooleanPoint{
Name: p.Name,
Tags: p.Tags,
Time: p.Time,
Aux: p.Aux,
}
if p.Nil {
bp.Nil = true
} else {
bp.Value = fn(p.Value, val)
}
return bp
},
}, nil
case func(bool, bool) bool:
var input BooleanIterator
switch lhs := lhs.(type) {
case BooleanIterator:
input = lhs
default:
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as an BooleanIterator", lhs)
}
var val bool
switch rhs := rhs.(type) {
case *BooleanLiteral:
val = rhs.Val
default:
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as an BooleanLiteral", rhs)
}
return &booleanTransformIterator{
input: input,
fn: func(p *BooleanPoint) *BooleanPoint {
if p == nil {
return nil
}
bp := &BooleanPoint{
Name: p.Name,
Tags: p.Tags,
Time: p.Time,
Aux: p.Aux,
}
if p.Nil {
bp.Nil = true
} else {
bp.Value = fn(p.Value, val)
}
return bp
},
}, nil
}
return nil, fmt.Errorf("unable to construct rhs transform iterator from %T and %T", lhs, rhs)
}
func buildLHSTransformIterator(lhs Literal, rhs Iterator, op Token, opt IteratorOptions) (Iterator, error) {
fn := binaryExprFunc(literalDataType(lhs), iteratorDataType(rhs), op)
switch fn := fn.(type) {
case func(float64, float64) float64:
var input FloatIterator
switch rhs := rhs.(type) {
case FloatIterator:
input = rhs
case IntegerIterator:
input = &integerFloatCastIterator{input: rhs}
default:
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as a FloatIterator", rhs)
}
var val float64
switch lhs := lhs.(type) {
case *NumberLiteral:
val = lhs.Val
case *IntegerLiteral:
val = float64(lhs.Val)
default:
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as a NumberLiteral", lhs)
}
return &floatTransformIterator{
input: input,
fn: func(p *FloatPoint) *FloatPoint {
if p == nil {
return nil
} else if p.Nil {
return p
}
p.Value = fn(val, p.Value)
return p
},
}, nil
case func(int64, int64) float64:
input, ok := rhs.(IntegerIterator)
if !ok {
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as a IntegerIterator", lhs)
}
var val int64
switch lhs := lhs.(type) {
case *IntegerLiteral:
val = lhs.Val
default:
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as a IntegerLiteral", rhs)
}
return &integerFloatTransformIterator{
input: input,
fn: func(p *IntegerPoint) *FloatPoint {
if p == nil {
return nil
}
fp := &FloatPoint{
Name: p.Name,
Tags: p.Tags,
Time: p.Time,
Aux: p.Aux,
}
if p.Nil {
fp.Nil = true
} else {
fp.Value = fn(val, p.Value)
}
return fp
},
}, nil
case func(float64, float64) bool:
var input FloatIterator
switch rhs := rhs.(type) {
case FloatIterator:
input = rhs
case IntegerIterator:
input = &integerFloatCastIterator{input: rhs}
default:
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as a FloatIterator", rhs)
}
var val float64
switch lhs := lhs.(type) {
case *NumberLiteral:
val = lhs.Val
case *IntegerLiteral:
val = float64(lhs.Val)
default:
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as a NumberLiteral", lhs)
}
return &floatBoolTransformIterator{
input: input,
fn: func(p *FloatPoint) *BooleanPoint {
if p == nil {
return nil
}
bp := &BooleanPoint{
Name: p.Name,
Tags: p.Tags,
Time: p.Time,
Aux: p.Aux,
}
if p.Nil {
bp.Nil = true
} else {
bp.Value = fn(val, p.Value)
}
return bp
},
}, nil
case func(int64, int64) int64:
var input IntegerIterator
switch rhs := rhs.(type) {
case IntegerIterator:
input = rhs
default:
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as an IntegerIterator", rhs)
}
var val int64
switch lhs := lhs.(type) {
case *IntegerLiteral:
val = lhs.Val
default:
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as an IntegerLiteral", lhs)
}
return &integerTransformIterator{
input: input,
fn: func(p *IntegerPoint) *IntegerPoint {
if p == nil {
return nil
} else if p.Nil {
return p
}
p.Value = fn(val, p.Value)
return p
},
}, nil
case func(int64, int64) bool:
var input IntegerIterator
switch rhs := rhs.(type) {
case IntegerIterator:
input = rhs
default:
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as an IntegerIterator", rhs)
}
var val int64
switch lhs := lhs.(type) {
case *IntegerLiteral:
val = lhs.Val
default:
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as an IntegerLiteral", lhs)
}
return &integerBoolTransformIterator{
input: input,
fn: func(p *IntegerPoint) *BooleanPoint {
if p == nil {
return nil
}
bp := &BooleanPoint{
Name: p.Name,
Tags: p.Tags,
Time: p.Time,
Aux: p.Aux,
}
if p.Nil {
bp.Nil = true
} else {
bp.Value = fn(val, p.Value)
}
return bp
},
}, nil
case func(bool, bool) bool:
var input BooleanIterator
switch rhs := rhs.(type) {
case BooleanIterator:
input = rhs
default:
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as an BooleanIterator", rhs)
}
var val bool
switch lhs := lhs.(type) {
case *BooleanLiteral:
val = lhs.Val
default:
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as a BooleanLiteral", lhs)
}
return &booleanTransformIterator{
input: input,
fn: func(p *BooleanPoint) *BooleanPoint {
if p == nil {
return nil
}
bp := &BooleanPoint{
Name: p.Name,
Tags: p.Tags,
Time: p.Time,
Aux: p.Aux,
}
if p.Nil {
bp.Nil = true
} else {
bp.Value = fn(val, p.Value)
}
return bp
},
}, nil
}
return nil, fmt.Errorf("unable to construct lhs transform iterator from %T and %T", lhs, rhs)
}
func buildTransformIterator(lhs Iterator, rhs Iterator, op Token, opt IteratorOptions) (Iterator, error) {
fn := binaryExprFunc(iteratorDataType(lhs), iteratorDataType(rhs), op)
switch fn := fn.(type) {
case func(float64, float64) float64:
var left FloatIterator
switch lhs := lhs.(type) {
case FloatIterator:
left = lhs
case IntegerIterator:
left = &integerFloatCastIterator{input: lhs}
default:
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as a FloatIterator", lhs)
}
var right FloatIterator
switch rhs := rhs.(type) {
case FloatIterator:
right = rhs
case IntegerIterator:
right = &integerFloatCastIterator{input: rhs}
default:
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as a FloatIterator", rhs)
}
return newFloatExprIterator(left, right, opt, fn), nil
case func(int64, int64) float64:
left, ok := lhs.(IntegerIterator)
if !ok {
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as a IntegerIterator", lhs)
}
right, ok := rhs.(IntegerIterator)
if !ok {
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as a IntegerIterator", rhs)
}
return newIntegerFloatExprIterator(left, right, opt, fn), nil
case func(int64, int64) int64:
left, ok := lhs.(IntegerIterator)
if !ok {
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as a IntegerIterator", lhs)
}
right, ok := rhs.(IntegerIterator)
if !ok {
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as a IntegerIterator", rhs)
}
return newIntegerExprIterator(left, right, opt, fn), nil
case func(float64, float64) bool:
var left FloatIterator
switch lhs := lhs.(type) {
case FloatIterator:
left = lhs
case IntegerIterator:
left = &integerFloatCastIterator{input: lhs}
default:
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as a FloatIterator", lhs)
}
var right FloatIterator
switch rhs := rhs.(type) {
case FloatIterator:
right = rhs
case IntegerIterator:
right = &integerFloatCastIterator{input: rhs}
default:
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as a FloatIterator", rhs)
}
return newFloatBooleanExprIterator(left, right, opt, fn), nil
case func(int64, int64) bool:
left, ok := lhs.(IntegerIterator)
if !ok {
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as a IntegerIterator", lhs)
}
right, ok := rhs.(IntegerIterator)
if !ok {
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as a IntegerIterator", rhs)
}
return newIntegerBooleanExprIterator(left, right, opt, fn), nil
case func(bool, bool) bool:
left, ok := lhs.(BooleanIterator)
if !ok {
return nil, fmt.Errorf("type mismatch on LHS, unable to use %T as a BooleanIterator", lhs)
}
right, ok := rhs.(BooleanIterator)
if !ok {
return nil, fmt.Errorf("type mismatch on RHS, unable to use %T as a BooleanIterator", rhs)
}
return newBooleanExprIterator(left, right, opt, fn), nil
}
return nil, fmt.Errorf("unable to construct transform iterator from %T and %T", lhs, rhs)
}
func iteratorDataType(itr Iterator) DataType {
switch itr.(type) {
case FloatIterator:
return Float
case IntegerIterator:
return Integer
case StringIterator:
return String
case BooleanIterator:
return Boolean
default:
return Unknown
}
}
func literalDataType(lit Literal) DataType {
switch lit.(type) {
case *NumberLiteral:
return Float
case *IntegerLiteral:
return Integer
case *StringLiteral:
return String
case *BooleanLiteral:
return Boolean
default:
return Unknown
}
}
func binaryExprFunc(typ1 DataType, typ2 DataType, op Token) interface{} {
var fn interface{}
switch typ1 {
case Float:
fn = floatBinaryExprFunc(op)
case Integer:
switch typ2 {
case Float:
fn = floatBinaryExprFunc(op)
default:
fn = integerBinaryExprFunc(op)
}
case Boolean:
fn = booleanBinaryExprFunc(op)
}
return fn
}
func floatBinaryExprFunc(op Token) interface{} {
switch op {
case ADD:
return func(lhs, rhs float64) float64 { return lhs + rhs }
case SUB:
return func(lhs, rhs float64) float64 { return lhs - rhs }
case MUL:
return func(lhs, rhs float64) float64 { return lhs * rhs }
case DIV:
return func(lhs, rhs float64) float64 {
if rhs == 0 {
return float64(0)
}
return lhs / rhs
}
case MOD:
return func(lhs, rhs float64) float64 { return math.Mod(lhs, rhs) }
case EQ:
return func(lhs, rhs float64) bool { return lhs == rhs }
case NEQ:
return func(lhs, rhs float64) bool { return lhs != rhs }
case LT:
return func(lhs, rhs float64) bool { return lhs < rhs }
case LTE:
return func(lhs, rhs float64) bool { return lhs <= rhs }
case GT:
return func(lhs, rhs float64) bool { return lhs > rhs }
case GTE:
return func(lhs, rhs float64) bool { return lhs >= rhs }
}
return nil
}
func integerBinaryExprFunc(op Token) interface{} {
switch op {
case ADD:
return func(lhs, rhs int64) int64 { return lhs + rhs }
case SUB:
return func(lhs, rhs int64) int64 { return lhs - rhs }
case MUL:
return func(lhs, rhs int64) int64 { return lhs * rhs }
case DIV:
return func(lhs, rhs int64) float64 {
if rhs == 0 {
return float64(0)
}
return float64(lhs) / float64(rhs)
}
case MOD:
return func(lhs, rhs int64) int64 {
if rhs == 0 {
return int64(0)
}
return lhs % rhs
}
case BITWISE_AND:
return func(lhs, rhs int64) int64 { return lhs & rhs }
case BITWISE_OR:
return func(lhs, rhs int64) int64 { return lhs | rhs }
case BITWISE_XOR:
return func(lhs, rhs int64) int64 { return lhs ^ rhs }
case EQ:
return func(lhs, rhs int64) bool { return lhs == rhs }
case NEQ:
return func(lhs, rhs int64) bool { return lhs != rhs }
case LT:
return func(lhs, rhs int64) bool { return lhs < rhs }
case LTE:
return func(lhs, rhs int64) bool { return lhs <= rhs }
case GT:
return func(lhs, rhs int64) bool { return lhs > rhs }
case GTE:
return func(lhs, rhs int64) bool { return lhs >= rhs }
}
return nil
}
func booleanBinaryExprFunc(op Token) interface{} {
switch op {
case BITWISE_AND:
return func(lhs, rhs bool) bool { return lhs && rhs }
case BITWISE_OR:
return func(lhs, rhs bool) bool { return lhs || rhs }
case BITWISE_XOR:
return func(lhs, rhs bool) bool { return lhs != rhs }
}
return nil
}
// stringSetSlice returns a sorted slice of keys from a string set.
func stringSetSlice(m map[string]struct{}) []string {
if m == nil {
return nil
}
a := make([]string, 0, len(m))
for k := range m {
a = append(a, k)
}
sort.Strings(a)
return a
}