func NewFillIterator(input Iterator, expr influxql.Expr, opt IteratorOptions) Iterator {
	switch input := input.(type) {
	case FloatIterator:
		return newFloatFillIterator(input, expr, opt)
	case IntegerIterator:
		return newIntegerFillIterator(input, expr, opt)
	case UnsignedIterator:
		return newUnsignedFillIterator(input, expr, opt)
	case StringIterator:
		return newStringFillIterator(input, expr, opt)
	case BooleanIterator:
		return newBooleanFillIterator(input, expr, opt)
	default:
		panic(fmt.Sprintf("unsupported fill iterator type: %T", input))
	}
}
NewFillIterator returns an iterator that fills in missing points in an aggregate.
package query
import (
	"context"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"regexp"
	"sync"
	"time"
	"github.com/gogo/protobuf/proto"
	"github.com/influxdata/influxdb/pkg/tracing"
	internal "github.com/influxdata/influxdb/query/internal"
	"github.com/influxdata/influxql"
)
var ErrUnknownCall = errors.New("unknown call")
const (
	secToNs = int64(time.Second)
)
type Iterator interface {
	Stats() IteratorStats
	Close() error
}
type Iterators []Iterator
func (a Iterators) Stats() IteratorStats {
	var stats IteratorStats
	for _, itr := range a {
		stats.Add(itr.Stats())
	}
	return stats
}
func (a Iterators) Close() error {
	for _, itr := range a {
		itr.Close()
	}
	return nil
}
func (a Iterators) filterNonNil() []Iterator {
	other := make([]Iterator, 0, len(a))
	for _, itr := range a {
		if itr == nil {
			continue
		}
		other = append(other, itr)
	}
	return other
}
func (a Iterators) dataType() influxql.DataType {
	if len(a) == 0 {
		return influxql.Unknown
	}
	switch a[0].(type) {
	case FloatIterator:
		return influxql.Float
	case IntegerIterator:
		return influxql.Integer
	case UnsignedIterator:
		return influxql.Unsigned
	case StringIterator:
		return influxql.String
	case BooleanIterator:
		return influxql.Boolean
	default:
		return influxql.Unknown
	}
}
func (a Iterators) coerce() interface{} {
	typ := a.dataType()
	switch typ {
	case influxql.Float:
		return newFloatIterators(a)
	case influxql.Integer:
		return newIntegerIterators(a)
	case influxql.Unsigned:
		return newUnsignedIterators(a)
	case influxql.String:
		return newStringIterators(a)
	case influxql.Boolean:
		return newBooleanIterators(a)
	}
	return a
}
func (a Iterators) Merge(opt IteratorOptions) (Iterator, error) {
	call, ok := opt.Expr.(*influxql.Call)
	if !ok && opt.MergeSorted() {
		itr := NewSortedMergeIterator(a, opt)
		if itr != nil && opt.InterruptCh != nil {
			itr = NewInterruptIterator(itr, opt.InterruptCh)
		}
		return itr, nil
	}
	itr := NewMergeIterator(a, opt)
	if itr == nil {
		return nil, nil
	}
	if opt.InterruptCh != nil {
		itr = NewInterruptIterator(itr, opt.InterruptCh)
	}
	if !ok {
		return itr, nil
	}
	if call.Name == "count" {
		opt.Expr = &influxql.Call{
			Name: "sum",
			Args: call.Args,
		}
	}
	return NewCallIterator(itr, opt)
}
func NewMergeIterator(inputs []Iterator, opt IteratorOptions) Iterator {
	inputs = Iterators(inputs).filterNonNil()
	if n := len(inputs); n == 0 {
		return nil
	} else if n == 1 {
		return inputs[0]
	}
	switch inputs := Iterators(inputs).coerce().(type) {
	case []FloatIterator:
		return newFloatMergeIterator(inputs, opt)
	case []IntegerIterator:
		return newIntegerMergeIterator(inputs, opt)
	case []UnsignedIterator:
		return newUnsignedMergeIterator(inputs, opt)
	case []StringIterator:
		return newStringMergeIterator(inputs, opt)
	case []BooleanIterator:
		return newBooleanMergeIterator(inputs, opt)
	default:
		panic(fmt.Sprintf("unsupported merge iterator type: %T", inputs))
	}
}
func NewParallelMergeIterator(inputs []Iterator, opt IteratorOptions, parallelism int) Iterator {
	inputs = Iterators(inputs).filterNonNil()
	if len(inputs) == 0 {
		return nil
	} else if len(inputs) == 1 {
		return inputs[0]
	}
	if len(inputs) < parallelism {
		parallelism = len(inputs)
	}
	n := len(inputs) / parallelism
	outputs := make([]Iterator, parallelism)
	for i := range outputs {
		var slice []Iterator
		if i < len(outputs)-1 {
			slice = inputs[i*n : (i+1)*n]
		} else {
			slice = inputs[i*n:]
		}
		outputs[i] = newParallelIterator(NewMergeIterator(slice, opt))
	}
	return NewMergeIterator(outputs, opt)
}
func NewSortedMergeIterator(inputs []Iterator, opt IteratorOptions) Iterator {
	inputs = Iterators(inputs).filterNonNil()
	if len(inputs) == 0 {
		return nil
	} else if len(inputs) == 1 {
		return inputs[0]
	}
	switch inputs := Iterators(inputs).coerce().(type) {
	case []FloatIterator:
		return newFloatSortedMergeIterator(inputs, opt)
	case []IntegerIterator:
		return newIntegerSortedMergeIterator(inputs, opt)
	case []UnsignedIterator:
		return newUnsignedSortedMergeIterator(inputs, opt)
	case []StringIterator:
		return newStringSortedMergeIterator(inputs, opt)
	case []BooleanIterator:
		return newBooleanSortedMergeIterator(inputs, opt)
	default:
		panic(fmt.Sprintf("unsupported sorted merge iterator type: %T", inputs))
	}
}
func newParallelIterator(input Iterator) Iterator {
	if input == nil {
		return nil
	}
	switch itr := input.(type) {
	case FloatIterator:
		return newFloatParallelIterator(itr)
	case IntegerIterator:
		return newIntegerParallelIterator(itr)
	case UnsignedIterator:
		return newUnsignedParallelIterator(itr)
	case StringIterator:
		return newStringParallelIterator(itr)
	case BooleanIterator:
		return newBooleanParallelIterator(itr)
	default:
		panic(fmt.Sprintf("unsupported parallel iterator type: %T", itr))
	}
}
func NewLimitIterator(input Iterator, opt IteratorOptions) Iterator {
	switch input := input.(type) {
	case FloatIterator:
		return newFloatLimitIterator(input, opt)
	case IntegerIterator:
		return newIntegerLimitIterator(input, opt)
	case UnsignedIterator:
		return newUnsignedLimitIterator(input, opt)
	case StringIterator:
		return newStringLimitIterator(input, opt)
	case BooleanIterator:
		return newBooleanLimitIterator(input, opt)
	default:
		panic(fmt.Sprintf("unsupported limit iterator type: %T", input))
	}
}
func NewFilterIterator(input Iterator, cond influxql.Expr, opt IteratorOptions) Iterator {
	if input == nil {
		return nil
	}
	switch input := input.(type) {
	case FloatIterator:
		return newFloatFilterIterator(input, cond, opt)
	case IntegerIterator:
		return newIntegerFilterIterator(input, cond, opt)
	case UnsignedIterator:
		return newUnsignedFilterIterator(input, cond, opt)
	case StringIterator:
		return newStringFilterIterator(input, cond, opt)
	case BooleanIterator:
		return newBooleanFilterIterator(input, cond, opt)
	default:
		panic(fmt.Sprintf("unsupported filter iterator type: %T", input))
	}
}
func NewDedupeIterator(input Iterator) Iterator {
	if input == nil {
		return nil
	}
	switch input := input.(type) {
	case FloatIterator:
		return newFloatDedupeIterator(input)
	case IntegerIterator:
		return newIntegerDedupeIterator(input)
	case UnsignedIterator:
		return newUnsignedDedupeIterator(input)
	case StringIterator:
		return newStringDedupeIterator(input)
	case BooleanIterator:
		return newBooleanDedupeIterator(input)
	default:
		panic(fmt.Sprintf("unsupported dedupe iterator type: %T", input))
	}
}