func (s *Shard) MeasurementExists(name []byte) (bool, error) {
	engine, err := s.engine()
	if err != nil {
		return false, err
	}
	return engine.MeasurementExists(name)
}
MeasurementExists returns true if the shard contains name.
TODO(edd): This method is currently only being called from tests; do we
really need it?
package tsdb
import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"path/filepath"
	"regexp"
	"runtime"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/gogo/protobuf/proto"
	"github.com/influxdata/influxdb/models"
	"github.com/influxdata/influxdb/pkg/bytesutil"
	"github.com/influxdata/influxdb/pkg/estimator"
	"github.com/influxdata/influxdb/pkg/file"
	"github.com/influxdata/influxdb/pkg/limiter"
	"github.com/influxdata/influxdb/query"
	internal "github.com/influxdata/influxdb/tsdb/internal"
	"github.com/influxdata/influxql"
	"go.uber.org/zap"
)
const (
	statWriteReq           = "writeReq"
	statWriteReqOK         = "writeReqOk"
	statWriteReqErr        = "writeReqErr"
	statSeriesCreate       = "seriesCreate"
	statFieldsCreate       = "fieldsCreate"
	statWritePointsErr     = "writePointsErr"
	statWritePointsDropped = "writePointsDropped"
	statWritePointsOK      = "writePointsOk"
	statWriteBytes         = "writeBytes"
	statDiskBytes          = "diskBytes"
)
var (
	ErrFieldOverflow = errors.New("field overflow")
	ErrFieldTypeConflict = errors.New("field type conflict")
	ErrFieldNotFound = errors.New("field not found")
	ErrFieldUnmappedID = errors.New("field ID not mapped")
	ErrEngineClosed = errors.New("engine is closed")
	ErrShardDisabled = errors.New("shard is disabled")
	ErrUnknownFieldsFormat = errors.New("unknown field index format")
	ErrShardNotIdle = errors.New("shard not idle")
	fieldsIndexMagicNumber = []byte{0, 6, 1, 3}
)
var (
	timeBytes = []byte("time")
)
type ShardError struct {
	id  uint64
	Err error
}
func NewShardError(id uint64, err error) error {
	if err == nil {
		return nil
	}
	return ShardError{id: id, Err: err}
}
func (e ShardError) Error() string {
	return fmt.Sprintf("[shard %d] %s", e.id, e.Err)
}
type PartialWriteError struct {
	Reason  string
	Dropped int
	DroppedKeys [][]byte
}
func (e PartialWriteError) Error() string {
	return fmt.Sprintf("partial write: %s dropped=%d", e.Reason, e.Dropped)
}
type Shard struct {
	path    string
	walPath string
	id      uint64
	database        string
	retentionPolicy string
	sfile   *SeriesFile
	options EngineOptions
	mu      sync.RWMutex
	_engine Engine
	index   Index
	enabled bool
	stats       *ShardStatistics
	defaultTags models.StatisticTags
	baseLogger *zap.Logger
	logger     *zap.Logger
	EnableOnOpen bool
}
func NewShard(id uint64, path string, walPath string, sfile *SeriesFile, opt EngineOptions) *Shard {
	db, rp := decodeStorePath(path)
	logger := zap.NewNop()
	s := &Shard{
		id:      id,
		path:    path,
		walPath: walPath,
		sfile:   sfile,
		options: opt,
		stats: &ShardStatistics{},
		defaultTags: models.StatisticTags{
			"path":            path,
			"walPath":         walPath,
			"id":              fmt.Sprintf("%d", id),
			"database":        db,
			"retentionPolicy": rp,
			"engine":          opt.EngineVersion,
		},
		database:        db,
		retentionPolicy: rp,
		logger:       logger,
		baseLogger:   logger,
		EnableOnOpen: true,
	}
	return s
}
func (s *Shard) WithLogger(log *zap.Logger) {
	s.baseLogger = log
	engine, err := s.engine()
	if err == nil {
		engine.WithLogger(s.baseLogger)
		s.index.WithLogger(s.baseLogger)
	}
	s.logger = s.baseLogger.With(zap.String("service", "shard"))
}
func (s *Shard) SetEnabled(enabled bool) {
	s.mu.Lock()
	s.enabled = enabled
	if s._engine != nil {
		s._engine.SetEnabled(enabled)
	}
	s.mu.Unlock()
}
func (s *Shard) ScheduleFullCompaction() error {
	engine, err := s.engine()
	if err != nil {
		return err
	}
	return engine.ScheduleFullCompaction()
}
func (s *Shard) ID() uint64 {
	return s.id
}
func (s *Shard) Database() string {
	return s.database
}
func (s *Shard) RetentionPolicy() string {
	return s.retentionPolicy
}
type ShardStatistics struct {
	WriteReq           int64
	WriteReqOK         int64
	WriteReqErr        int64
	FieldsCreated      int64
	WritePointsErr     int64
	WritePointsDropped int64
	WritePointsOK      int64
	BytesWritten       int64
	DiskBytes          int64
}
func (s *Shard) Statistics(tags map[string]string) []models.Statistic {
	engine, err := s.engine()
	if err != nil {
		return nil
	}
	if _, err := s.DiskSize(); err != nil {
		return nil
	}
	seriesN := engine.SeriesN()
	tags = s.defaultTags.Merge(tags)
	statistics := []models.Statistic{{
		Name: "shard",
		Tags: tags,
		Values: map[string]interface{}{
			statWriteReq:           atomic.LoadInt64(&s.stats.WriteReq),
			statWriteReqOK:         atomic.LoadInt64(&s.stats.WriteReqOK),
			statWriteReqErr:        atomic.LoadInt64(&s.stats.WriteReqErr),
			statSeriesCreate:       seriesN,
			statFieldsCreate:       atomic.LoadInt64(&s.stats.FieldsCreated),
			statWritePointsErr:     atomic.LoadInt64(&s.stats.WritePointsErr),
			statWritePointsDropped: atomic.LoadInt64(&s.stats.WritePointsDropped),
			statWritePointsOK:      atomic.LoadInt64(&s.stats.WritePointsOK),
			statWriteBytes:         atomic.LoadInt64(&s.stats.BytesWritten),
			statDiskBytes:          atomic.LoadInt64(&s.stats.DiskBytes),
		},
	}}
	statistics = append(statistics, engine.Statistics(tags)...)
	return statistics
}
func (s *Shard) Path() string { return s.path }
func (s *Shard) Open() error {
	if err := func() error {
		s.mu.Lock()
		defer s.mu.Unlock()
		if s._engine != nil {
			return nil
		}
		seriesIDSet := NewSeriesIDSet()
		ipath := filepath.Join(s.path, "index")
		idx, err := NewIndex(s.id, s.database, ipath, seriesIDSet, s.sfile, s.options)
		if err != nil {
			return err
		}
		if err := idx.Open(); err != nil {
			return err
		}
		s.index = idx
		idx.WithLogger(s.baseLogger)
		e, err := NewEngine(s.id, idx, s.database, s.path, s.walPath, s.sfile, s.options)
		if err != nil {
			return err
		}
		e.WithLogger(s.baseLogger)
		e.SetEnabled(false)
		if err := e.Open(); err != nil {
			return err
		}
		if err := e.LoadMetadataIndex(s.id, s.index); err != nil {
			return err
		}
		s._engine = e
		return nil
	}(); err != nil {
		s.close()
		return NewShardError(s.id, err)
	}
	if s.EnableOnOpen {
		s.SetEnabled(true)
	}
	return nil
}
func (s *Shard) Close() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.close()
}
func (s *Shard) close() error {
	if s._engine == nil {
		return nil
	}
	err := s._engine.Close()
	if err == nil {
		s._engine = nil
	}
	if e := s.index.Close(); e == nil {
		s.index = nil
	}
	return err
}
func (s *Shard) IndexType() string {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if err := s.ready(); err != nil {
		return ""
	}
	return s.index.Type()
}
func (s *Shard) ready() error {
	var err error
	if s._engine == nil {
		err = ErrEngineClosed
	} else if !s.enabled {
		err = ErrShardDisabled
	}
	return err
}
func (s *Shard) LastModified() time.Time {
	engine, err := s.engine()
	if err != nil {
		return time.Time{}
	}
	return engine.LastModified()
}
func (s *Shard) Index() (Index, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if err := s.ready(); err != nil {
		return nil, err
	}
	return s.index, nil
}
func (s *Shard) seriesFile() (*SeriesFile, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if err := s.ready(); err != nil {
		return nil, err
	}
	return s.sfile, nil
}
func (s *Shard) IsIdle() bool {
	engine, err := s.engine()
	if err != nil {
		return true
	}
	return engine.IsIdle()
}
func (s *Shard) Free() error {
	engine, err := s.engine()
	if err != nil {
		return err
	}
	s.SetCompactionsEnabled(false)
	return engine.Free()
}
func (s *Shard) SetCompactionsEnabled(enabled bool) {
	engine, err := s.engine()
	if err != nil {
		return
	}
	engine.SetCompactionsEnabled(enabled)
}
func (s *Shard) DiskSize() (int64, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if s._engine == nil {
		return 0, ErrEngineClosed
	}
	size := s._engine.DiskSize()
	atomic.StoreInt64(&s.stats.DiskBytes, size)
	return size, nil
}
type FieldCreate struct {
	Measurement []byte
	Field       *Field
}
func (s *Shard) WritePoints(points []models.Point) error {
	s.mu.RLock()
	defer s.mu.RUnlock()
	engine, err := s.engineNoLock()
	if err != nil {
		return err
	}
	var writeError error
	atomic.AddInt64(&s.stats.WriteReq, 1)
	points, fieldsToCreate, err := s.validateSeriesAndFields(points)
	if err != nil {
		if _, ok := err.(PartialWriteError); !ok {
			return err
		}
		writeError = err
	}
	atomic.AddInt64(&s.stats.FieldsCreated, int64(len(fieldsToCreate)))
	if err := s.createFieldsAndMeasurements(fieldsToCreate); err != nil {
		return err
	}
	if err := engine.WritePoints(points); err != nil {
		atomic.AddInt64(&s.stats.WritePointsErr, int64(len(points)))
		atomic.AddInt64(&s.stats.WriteReqErr, 1)
		return fmt.Errorf("engine: %s", err)
	}
	atomic.AddInt64(&s.stats.WritePointsOK, int64(len(points)))
	atomic.AddInt64(&s.stats.WriteReqOK, 1)
	return writeError
}
func (s *Shard) validateSeriesAndFields(points []models.Point) ([]models.Point, []*FieldCreate, error) {
	var (
		fieldsToCreate []*FieldCreate
		err            error
		dropped        int
		reason         string  
	)
	keys := make([][]byte, len(points))
	names := make([][]byte, len(points))
	tagsSlice := make([]models.Tags, len(points))
	var j int
	for i, p := range points {
		tags := p.Tags()
		if v := tags.Get(timeBytes); v != nil {
			dropped++
			if reason == "" {
				reason = fmt.Sprintf("invalid tag key: input tag \"%s\" on measurement \"%s\" is invalid", "time", string(p.Name()))
			}
			continue
		}
		keys[j] = p.Key()
		names[j] = p.Name()
		tagsSlice[j] = tags
		points[j] = points[i]
		j++
	}
	points, keys, names, tagsSlice = points[:j], keys[:j], names[:j], tagsSlice[:j]
	engine, err := s.engineNoLock()
	if err != nil {
		return nil, nil, err
	}
	var droppedKeys [][]byte
	if err := engine.CreateSeriesListIfNotExists(keys, names, tagsSlice); err != nil {
		switch err := err.(type) {
		case *PartialWriteError:
			reason = err.Reason
			dropped += err.Dropped
			droppedKeys = err.DroppedKeys
			atomic.AddInt64(&s.stats.WritePointsDropped, int64(err.Dropped))
		default:
			return nil, nil, err
		}
	}
	n := 0
	mfCache := make(map[string]*MeasurementFields, 16)
	for i, p := range points {
		var skip bool
		var validField bool
		iter := p.FieldIterator()
		for iter.Next() {
			if bytes.Equal(iter.FieldKey(), timeBytes) {
				continue
			}
			validField = true
			break
		}
		if !validField {
			dropped++
			if reason == "" {
				reason = fmt.Sprintf("invalid field name: input field \"%s\" on measurement \"%s\" is invalid", "time", string(p.Name()))
			}
			continue
		}
		iter.Reset()
		if len(droppedKeys) > 0 && bytesutil.Contains(droppedKeys, keys[i]) {
			continue
		}
		name := p.Name()
		mf := mfCache[string(name)]
		if mf == nil {
			mf = engine.MeasurementFields(name).Clone()
			mfCache[string(name)] = mf
		}
		iter.Reset()
		for iter.Next() {
			if bytes.Equal(iter.FieldKey(), timeBytes) {
				continue
			}
			var fieldType influxql.DataType
			switch iter.Type() {
			case models.Float:
				fieldType = influxql.Float
			case models.Integer:
				fieldType = influxql.Integer
			case models.Unsigned:
				fieldType = influxql.Unsigned
			case models.Boolean:
				fieldType = influxql.Boolean
			case models.String:
				fieldType = influxql.String
			default:
				continue
			}
			if f := mf.FieldBytes(iter.FieldKey()); f != nil {
				if f.Type != fieldType {
					atomic.AddInt64(&s.stats.WritePointsDropped, 1)
					dropped++
					if reason == "" {
						reason = fmt.Sprintf("%s: input field \"%s\" on measurement \"%s\" is type %s, already exists as type %s", ErrFieldTypeConflict, iter.FieldKey(), name, fieldType, f.Type)
					}
					skip = true
				} else {
					continue  
				}
			}
			if !skip {
				fieldsToCreate = append(fieldsToCreate, &FieldCreate{p.Name(), &Field{Name: string(iter.FieldKey()), Type: fieldType}})
			}
		}
		if !skip {
			points[n] = points[i]
			n++
		}
	}
	points = points[:n]
	if dropped > 0 {
		err = PartialWriteError{Reason: reason, Dropped: dropped}
	}
	return points, fieldsToCreate, err
}
func (s *Shard) createFieldsAndMeasurements(fieldsToCreate []*FieldCreate) error {
	if len(fieldsToCreate) == 0 {
		return nil
	}
	engine, err := s.engineNoLock()
	if err != nil {
		return err
	}
	for _, f := range fieldsToCreate {
		mf := engine.MeasurementFields(f.Measurement)
		if err := mf.CreateFieldIfNotExists([]byte(f.Field.Name), f.Field.Type); err != nil {
			return err
		}
		s.index.SetFieldName(f.Measurement, f.Field.Name)
	}
	if len(fieldsToCreate) > 0 {
		return engine.MeasurementFieldSet().Save()
	}
	return nil
}
func (s *Shard) DeleteSeriesRange(itr SeriesIterator, min, max int64) error {
	engine, err := s.engine()
	if err != nil {
		return err
	}
	return engine.DeleteSeriesRange(itr, min, max)
}
func (s *Shard) DeleteMeasurement(name []byte) error {
	engine, err := s.engine()
	if err != nil {
		return err
	}
	return engine.DeleteMeasurement(name)
}
func (s *Shard) SeriesN() int64 {
	engine, err := s.engine()
	if err != nil {
		return 0
	}
	return engine.SeriesN()
}
func (s *Shard) SeriesSketches() (estimator.Sketch, estimator.Sketch, error) {
	engine, err := s.engine()
	if err != nil {
		return nil, nil, err
	}
	return engine.SeriesSketches()
}
func (s *Shard) MeasurementsSketches() (estimator.Sketch, estimator.Sketch, error) {
	engine, err := s.engine()
	if err != nil {
		return nil, nil, err
	}
	return engine.MeasurementsSketches()
}
func (s *Shard) MeasurementNamesByRegex(re *regexp.Regexp) ([][]byte, error) {
	engine, err := s.engine()
	if err != nil {
		return nil, err
	}
	return engine.MeasurementNamesByRegex(re)
}
func (s *Shard) MeasurementTagKeysByExpr(name []byte, expr influxql.Expr) (map[string]struct{}, error) {
	engine, err := s.engine()
	if err != nil {
		return nil, err
	}
	return engine.MeasurementTagKeysByExpr(name, expr)
}
func (s *Shard) MeasurementTagKeyValuesByExpr(auth query.Authorizer, name []byte, key []string, expr influxql.Expr, keysSorted bool) ([][]string, error) {
	index, err := s.Index()
	if err != nil {
		return nil, err
	}
	indexSet := IndexSet{Indexes: []Index{index}, SeriesFile: s.sfile}
	return indexSet.MeasurementTagKeyValuesByExpr(auth, name, key, expr, keysSorted)
}
func (s *Shard) MeasurementFields(name []byte) *MeasurementFields {
	engine, err := s.engine()
	if err != nil {
		return nil
	}
	return engine.MeasurementFields(name)
}