func (r *FloatMovingAverageReducer) Emit() []FloatPoint {
	if len(r.buf) != cap(r.buf) {
		return []FloatPoint{}
	}
	return []FloatPoint{
		{
			Value:      r.sum / float64(len(r.buf)),
			Time:       r.time,
			Aggregated: uint32(len(r.buf)),
		},
	}
}
Emit emits the moving average of the current window. Emit should be called
after every call to AggregateFloat and it will produce one point if there
is enough data to fill a window, otherwise it will produce zero points.
package query
import (
	"container/heap"
	"math"
	"sort"
	"time"
	"github.com/influxdata/influxdb/query/neldermead"
	"github.com/influxdata/influxql"
)
type FloatMeanReducer struct {
	sum   float64
	count uint32
}
func NewFloatMeanReducer() *FloatMeanReducer {
	return &FloatMeanReducer{}
}
func (r *FloatMeanReducer) AggregateFloat(p *FloatPoint) {
	if p.Aggregated >= 2 {
		r.sum += p.Value * float64(p.Aggregated)
		r.count += p.Aggregated
	} else {
		r.sum += p.Value
		r.count++
	}
}
func (r *FloatMeanReducer) Emit() []FloatPoint {
	return []FloatPoint{{
		Time:       ZeroTime,
		Value:      r.sum / float64(r.count),
		Aggregated: r.count,
	}}
}
type IntegerMeanReducer struct {
	sum   int64
	count uint32
}
func NewIntegerMeanReducer() *IntegerMeanReducer {
	return &IntegerMeanReducer{}
}
func (r *IntegerMeanReducer) AggregateInteger(p *IntegerPoint) {
	if p.Aggregated >= 2 {
		r.sum += p.Value * int64(p.Aggregated)
		r.count += p.Aggregated
	} else {
		r.sum += p.Value
		r.count++
	}
}
func (r *IntegerMeanReducer) Emit() []FloatPoint {
	return []FloatPoint{{
		Time:       ZeroTime,
		Value:      float64(r.sum) / float64(r.count),
		Aggregated: r.count,
	}}
}
type UnsignedMeanReducer struct {
	sum   uint64
	count uint32
}
func NewUnsignedMeanReducer() *UnsignedMeanReducer {
	return &UnsignedMeanReducer{}
}
func (r *UnsignedMeanReducer) AggregateUnsigned(p *UnsignedPoint) {
	if p.Aggregated >= 2 {
		r.sum += p.Value * uint64(p.Aggregated)
		r.count += p.Aggregated
	} else {
		r.sum += p.Value
		r.count++
	}
}
func (r *UnsignedMeanReducer) Emit() []FloatPoint {
	return []FloatPoint{{
		Time:       ZeroTime,
		Value:      float64(r.sum) / float64(r.count),
		Aggregated: r.count,
	}}
}
type FloatDerivativeReducer struct {
	interval      Interval
	prev          FloatPoint
	curr          FloatPoint
	isNonNegative bool
	ascending     bool
}
func NewFloatDerivativeReducer(interval Interval, isNonNegative, ascending bool) *FloatDerivativeReducer {
	return &FloatDerivativeReducer{
		interval:      interval,
		isNonNegative: isNonNegative,
		ascending:     ascending,
		prev:          FloatPoint{Nil: true},
		curr:          FloatPoint{Nil: true},
	}
}
func (r *FloatDerivativeReducer) AggregateFloat(p *FloatPoint) {
	if !r.curr.Nil && r.curr.Time == p.Time {
		return
	}
	r.prev = r.curr
	r.curr = *p
}
func (r *FloatDerivativeReducer) Emit() []FloatPoint {
	if !r.prev.Nil {
		diff := r.curr.Value - r.prev.Value
		elapsed := r.curr.Time - r.prev.Time
		if !r.ascending {
			elapsed = -elapsed
		}
		value := diff / (float64(elapsed) / float64(r.interval.Duration))
		r.prev.Nil = true
		if r.isNonNegative && diff < 0 {
			return nil
		}
		return []FloatPoint{{Time: r.curr.Time, Value: value}}
	}
	return nil
}
type IntegerDerivativeReducer struct {
	interval      Interval
	prev          IntegerPoint
	curr          IntegerPoint
	isNonNegative bool
	ascending     bool
}
func NewIntegerDerivativeReducer(interval Interval, isNonNegative, ascending bool) *IntegerDerivativeReducer {
	return &IntegerDerivativeReducer{
		interval:      interval,
		isNonNegative: isNonNegative,
		ascending:     ascending,
		prev:          IntegerPoint{Nil: true},
		curr:          IntegerPoint{Nil: true},
	}
}
func (r *IntegerDerivativeReducer) AggregateInteger(p *IntegerPoint) {
	if !r.curr.Nil && r.curr.Time == p.Time {
		return
	}
	r.prev = r.curr
	r.curr = *p
}
func (r *IntegerDerivativeReducer) Emit() []FloatPoint {
	if !r.prev.Nil {
		diff := float64(r.curr.Value - r.prev.Value)
		elapsed := r.curr.Time - r.prev.Time
		if !r.ascending {
			elapsed = -elapsed
		}
		value := diff / (float64(elapsed) / float64(r.interval.Duration))
		r.prev.Nil = true
		if r.isNonNegative && diff < 0 {
			return nil
		}
		return []FloatPoint{{Time: r.curr.Time, Value: value}}
	}
	return nil
}
type UnsignedDerivativeReducer struct {
	interval      Interval
	prev          UnsignedPoint
	curr          UnsignedPoint
	isNonNegative bool
	ascending     bool
}
func NewUnsignedDerivativeReducer(interval Interval, isNonNegative, ascending bool) *UnsignedDerivativeReducer {
	return &UnsignedDerivativeReducer{
		interval:      interval,
		isNonNegative: isNonNegative,
		ascending:     ascending,
		prev:          UnsignedPoint{Nil: true},
		curr:          UnsignedPoint{Nil: true},
	}
}
func (r *UnsignedDerivativeReducer) AggregateUnsigned(p *UnsignedPoint) {
	if !r.curr.Nil && r.curr.Time == p.Time {
		return
	}
	r.prev = r.curr
	r.curr = *p
}
func (r *UnsignedDerivativeReducer) Emit() []FloatPoint {
	if !r.prev.Nil {
		var diff float64
		if r.curr.Value > r.prev.Value {
			diff = float64(r.curr.Value - r.prev.Value)
		} else {
			diff = -float64(r.prev.Value - r.curr.Value)
		}
		elapsed := r.curr.Time - r.prev.Time
		if !r.ascending {
			elapsed = -elapsed
		}
		value := diff / (float64(elapsed) / float64(r.interval.Duration))
		r.prev.Nil = true
		if r.isNonNegative && diff < 0 {
			return nil
		}
		return []FloatPoint{{Time: r.curr.Time, Value: value}}
	}
	return nil
}
type FloatDifferenceReducer struct {
	isNonNegative bool
	prev          FloatPoint
	curr          FloatPoint
}
func NewFloatDifferenceReducer(isNonNegative bool) *FloatDifferenceReducer {
	return &FloatDifferenceReducer{
		isNonNegative: isNonNegative,
		prev:          FloatPoint{Nil: true},
		curr:          FloatPoint{Nil: true},
	}
}
func (r *FloatDifferenceReducer) AggregateFloat(p *FloatPoint) {
	if !r.curr.Nil && r.curr.Time == p.Time {
		return
	}
	r.prev = r.curr
	r.curr = *p
}
func (r *FloatDifferenceReducer) Emit() []FloatPoint {
	if !r.prev.Nil {
		value := r.curr.Value - r.prev.Value
		if r.isNonNegative && value < 0 {
			return nil
		}
		r.prev.Nil = true
		return []FloatPoint{{Time: r.curr.Time, Value: value}}
	}
	return nil
}
type IntegerDifferenceReducer struct {
	isNonNegative bool
	prev          IntegerPoint
	curr          IntegerPoint
}
func NewIntegerDifferenceReducer(isNonNegative bool) *IntegerDifferenceReducer {
	return &IntegerDifferenceReducer{
		isNonNegative: isNonNegative,
		prev:          IntegerPoint{Nil: true},
		curr:          IntegerPoint{Nil: true},
	}
}
func (r *IntegerDifferenceReducer) AggregateInteger(p *IntegerPoint) {
	if !r.curr.Nil && r.curr.Time == p.Time {
		return
	}
	r.prev = r.curr
	r.curr = *p
}
func (r *IntegerDifferenceReducer) Emit() []IntegerPoint {
	if !r.prev.Nil {
		value := r.curr.Value - r.prev.Value
		if r.isNonNegative && value < 0 {
			return nil
		}
		r.prev.Nil = true
		return []IntegerPoint{{Time: r.curr.Time, Value: value}}
	}
	return nil
}
type UnsignedDifferenceReducer struct {
	isNonNegative bool
	prev          UnsignedPoint
	curr          UnsignedPoint
}
func NewUnsignedDifferenceReducer(isNonNegative bool) *UnsignedDifferenceReducer {
	return &UnsignedDifferenceReducer{
		isNonNegative: isNonNegative,
		prev:          UnsignedPoint{Nil: true},
		curr:          UnsignedPoint{Nil: true},
	}
}
func (r *UnsignedDifferenceReducer) AggregateUnsigned(p *UnsignedPoint) {
	if !r.curr.Nil && r.curr.Time == p.Time {
		return
	}
	r.prev = r.curr
	r.curr = *p
}
func (r *UnsignedDifferenceReducer) Emit() []UnsignedPoint {
	if !r.prev.Nil {
		if r.isNonNegative && r.curr.Value < r.prev.Value {
			return nil
		}
		value := r.curr.Value - r.prev.Value
		r.prev.Nil = true
		return []UnsignedPoint{{Time: r.curr.Time, Value: value}}
	}
	return nil
}
type FloatMovingAverageReducer struct {
	pos  int
	sum  float64
	time int64
	buf  []float64
}
func NewFloatMovingAverageReducer(n int) *FloatMovingAverageReducer {
	return &FloatMovingAverageReducer{
		buf: make([]float64, 0, n),
	}
}
func (r *FloatMovingAverageReducer) AggregateFloat(p *FloatPoint) {
	if len(r.buf) != cap(r.buf) {
		r.buf = append(r.buf, p.Value)
	} else {
		r.sum -= r.buf[r.pos]
		r.buf[r.pos] = p.Value
	}
	r.sum += p.Value
	r.time = p.Time
	r.pos++
	if r.pos >= cap(r.buf) {
		r.pos = 0
	}
}