func (itr *floatSortedMergeIterator) pop() (*FloatPoint, error) {
	if !itr.init {
		items := itr.heap.items
		itr.heap.items = make([]*floatSortedMergeHeapItem, 0, len(items))
		for _, item := range items {
			var err error
			if item.point, err = item.itr.Next(); err != nil {
				return nil, err
			} else if item.point == nil {
				continue
			}
			itr.heap.items = append(itr.heap.items, item)
		}
		heap.Init(itr.heap)
		itr.init = true
	}
	if len(itr.heap.items) == 0 {
		return nil, nil
	}
	item := heap.Pop(itr.heap).(*floatSortedMergeHeapItem)
	if item.err != nil {
		return nil, item.err
	} else if item.point == nil {
		return nil, nil
	}
	p := item.point.Clone()
	if item.point, item.err = item.itr.Next(); item.point != nil {
		heap.Push(itr.heap, item)
	}
	return p, nil
}
pop returns the next point from the heap.
Reads the next point from item's cursor and puts it back on the heap.
package query
import (
	"container/heap"
	"context"
	"io"
	"sort"
	"sync"
	"time"
	"github.com/gogo/protobuf/proto"
	"github.com/influxdata/influxql"
)
const DefaultStatsInterval = time.Second
type FloatIterator interface {
	Iterator
	Next() (*FloatPoint, error)
}
func newFloatIterators(itrs []Iterator) []FloatIterator {
	a := make([]FloatIterator, 0, len(itrs))
	for _, itr := range itrs {
		switch itr := itr.(type) {
		case FloatIterator:
			a = append(a, itr)
		default:
			itr.Close()
		}
	}
	return a
}
type bufFloatIterator struct {
	itr FloatIterator
	buf *FloatPoint
}
func newBufFloatIterator(itr FloatIterator) *bufFloatIterator {
	return &bufFloatIterator{itr: itr}
}
func (itr *bufFloatIterator) Stats() IteratorStats { return itr.itr.Stats() }
func (itr *bufFloatIterator) Close() error { return itr.itr.Close() }
func (itr *bufFloatIterator) peek() (*FloatPoint, error) {
	p, err := itr.Next()
	if err != nil {
		return nil, err
	}
	itr.unread(p)
	return p, nil
}
func (itr *bufFloatIterator) peekTime() (int64, error) {
	p, err := itr.peek()
	if p == nil || err != nil {
		return ZeroTime, err
	}
	return p.Time, nil
}
func (itr *bufFloatIterator) Next() (*FloatPoint, error) {
	buf := itr.buf
	if buf != nil {
		itr.buf = nil
		return buf, nil
	}
	return itr.itr.Next()
}
func (itr *bufFloatIterator) NextInWindow(startTime, endTime int64) (*FloatPoint, error) {
	v, err := itr.Next()
	if v == nil || err != nil {
		return nil, err
	} else if t := v.Time; t >= endTime || t < startTime {
		itr.unread(v)
		return nil, nil
	}
	return v, nil
}
func (itr *bufFloatIterator) unread(v *FloatPoint) { itr.buf = v }
type floatMergeIterator struct {
	inputs []FloatIterator
	heap   *floatMergeHeap
	init   bool
	closed bool
	mu     sync.RWMutex
	curr   *floatMergeHeapItem
	window struct {
		name      string
		tags      string
		startTime int64
		endTime   int64
	}
}
func newFloatMergeIterator(inputs []FloatIterator, opt IteratorOptions) *floatMergeIterator {
	itr := &floatMergeIterator{
		inputs: inputs,
		heap: &floatMergeHeap{
			items: make([]*floatMergeHeapItem, 0, len(inputs)),
			opt:   opt,
		},
	}
	for _, input := range inputs {
		bufInput := newBufFloatIterator(input)
		itr.heap.items = append(itr.heap.items, &floatMergeHeapItem{itr: bufInput})
	}
	return itr
}
func (itr *floatMergeIterator) Stats() IteratorStats {
	var stats IteratorStats
	for _, input := range itr.inputs {
		stats.Add(input.Stats())
	}
	return stats
}
func (itr *floatMergeIterator) Close() error {
	itr.mu.Lock()
	defer itr.mu.Unlock()
	for _, input := range itr.inputs {
		input.Close()
	}
	itr.curr = nil
	itr.inputs = nil
	itr.heap.items = nil
	itr.closed = true
	return nil
}
func (itr *floatMergeIterator) Next() (*FloatPoint, error) {
	itr.mu.RLock()
	defer itr.mu.RUnlock()
	if itr.closed {
		return nil, nil
	}
	if !itr.init {
		items := itr.heap.items
		itr.heap.items = make([]*floatMergeHeapItem, 0, len(items))
		for _, item := range items {
			if p, err := item.itr.peek(); err != nil {
				return nil, err
			} else if p == nil {
				continue
			}
			itr.heap.items = append(itr.heap.items, item)
		}
		heap.Init(itr.heap)
		itr.init = true
	}
	for {
		if itr.curr == nil {
			if len(itr.heap.items) == 0 {
				return nil, nil
			}
			itr.curr = heap.Pop(itr.heap).(*floatMergeHeapItem)
			p, err := itr.curr.itr.Next()
			if err != nil {
				return nil, err
			}
			tags := p.Tags.Subset(itr.heap.opt.Dimensions)
			itr.window.name, itr.window.tags = p.Name, tags.ID()
			itr.window.startTime, itr.window.endTime = itr.heap.opt.Window(p.Time)
			return p, nil
		}
		p, err := itr.curr.itr.Next()
		if err != nil {
			return nil, err
		}
		if p == nil {
			itr.curr = nil
			continue
		}
		inWindow := true
		if window := itr.window; window.name != p.Name {
			inWindow = false
		} else if tags := p.Tags.Subset(itr.heap.opt.Dimensions); window.tags != tags.ID() {
			inWindow = false
		} else if opt := itr.heap.opt; opt.Ascending && p.Time >= window.endTime {
			inWindow = false
		} else if !opt.Ascending && p.Time < window.startTime {
			inWindow = false
		}
		if !inWindow {
			itr.curr.itr.unread(p)
			heap.Push(itr.heap, itr.curr)
			itr.curr = nil
			continue
		}
		return p, nil
	}
}
type floatMergeHeap struct {
	opt   IteratorOptions
	items []*floatMergeHeapItem
}
func (h *floatMergeHeap) Len() int      { return len(h.items) }
func (h *floatMergeHeap) Swap(i, j int) { h.items[i], h.items[j] = h.items[j], h.items[i] }
func (h *floatMergeHeap) Less(i, j int) bool {
	x, err := h.items[i].itr.peek()
	if err != nil {
		return true
	}
	y, err := h.items[j].itr.peek()
	if err != nil {
		return false
	}
	if h.opt.Ascending {
		if x.Name != y.Name {
			return x.Name < y.Name
		} else if xTags, yTags := x.Tags.Subset(h.opt.Dimensions), y.Tags.Subset(h.opt.Dimensions); xTags.ID() != yTags.ID() {
			return xTags.ID() < yTags.ID()
		}
	} else {
		if x.Name != y.Name {
			return x.Name > y.Name
		} else if xTags, yTags := x.Tags.Subset(h.opt.Dimensions), y.Tags.Subset(h.opt.Dimensions); xTags.ID() != yTags.ID() {
			return xTags.ID() > yTags.ID()
		}
	}
	xt, _ := h.opt.Window(x.Time)
	yt, _ := h.opt.Window(y.Time)
	if h.opt.Ascending {
		return xt < yt
	}
	return xt > yt
}
func (h *floatMergeHeap) Push(x interface{}) {
	h.items = append(h.items, x.(*floatMergeHeapItem))
}
func (h *floatMergeHeap) Pop() interface{} {
	old := h.items
	n := len(old)
	item := old[n-1]
	h.items = old[0 : n-1]
	return item
}
type floatMergeHeapItem struct {
	itr *bufFloatIterator
}
type floatSortedMergeIterator struct {
	inputs []FloatIterator
	heap   *floatSortedMergeHeap
	init   bool
}
func newFloatSortedMergeIterator(inputs []FloatIterator, opt IteratorOptions) Iterator {
	itr := &floatSortedMergeIterator{
		inputs: inputs,
		heap: &floatSortedMergeHeap{
			items: make([]*floatSortedMergeHeapItem, 0, len(inputs)),
			opt:   opt,
		},
	}
	for _, input := range inputs {
		itr.heap.items = append(itr.heap.items, &floatSortedMergeHeapItem{itr: input})
	}
	return itr
}
func (itr *floatSortedMergeIterator) Stats() IteratorStats {
	var stats IteratorStats
	for _, input := range itr.inputs {
		stats.Add(input.Stats())
	}
	return stats
}
func (itr *floatSortedMergeIterator) Close() error {
	for _, input := range itr.inputs {
		input.Close()
	}
	return nil
}
func (itr *floatSortedMergeIterator) Next() (*FloatPoint, error) { return itr.pop() }