func (s *Store) CreateShard(database, retentionPolicy string, shardID uint64, enabled bool) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	select {
	case <-s.closing:
		return ErrStoreClosed
	default:
	}
	if _, ok := s.shards[shardID]; ok {
		return nil
	}
	if _, ok := s.pendingShardDeletes[shardID]; ok {
		return ErrShardDeletion
	}
	if err := os.MkdirAll(filepath.Join(s.path, database, retentionPolicy), 0700); err != nil {
		return err
	}
	walPath := filepath.Join(s.EngineOptions.Config.WALDir, database, retentionPolicy, fmt.Sprintf("%d", shardID))
	if err := os.MkdirAll(walPath, 0700); err != nil {
		return err
	}
	sfile, err := s.openSeriesFile(database)
	if err != nil {
		return err
	}
	idx, err := s.createIndexIfNotExists(database)
	if err != nil {
		return err
	}
	opt := s.EngineOptions
	opt.InmemIndex = idx
	opt.SeriesIDSets = shardSet{store: s, db: database}
	path := filepath.Join(s.path, database, retentionPolicy, strconv.FormatUint(shardID, 10))
	shard := NewShard(shardID, path, walPath, sfile, opt)
	shard.WithLogger(s.baseLogger)
	shard.EnableOnOpen = enabled
	if err := shard.Open(); err != nil {
		return err
	}
	s.shards[shardID] = shard
	s.databases[database] = struct{}{}  
	return nil
}
CreateShard creates a shard with the given id and retention policy on a database.
package tsdb  
import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"path/filepath"
	"runtime"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
	"github.com/influxdata/influxdb/logger"
	"github.com/influxdata/influxdb/pkg/estimator/hll"
	"github.com/influxdata/influxdb/models"
	"github.com/influxdata/influxdb/pkg/estimator"
	"github.com/influxdata/influxdb/pkg/limiter"
	"github.com/influxdata/influxdb/query"
	"github.com/influxdata/influxql"
	"go.uber.org/zap"
)
var (
	ErrShardNotFound = fmt.Errorf("shard not found")
	ErrStoreClosed = fmt.Errorf("store is closed")
	ErrShardDeletion = errors.New("shard is being deleted")
)
const (
	statDatabaseSeries       = "numSeries"        
	statDatabaseMeasurements = "numMeasurements"  
)
const SeriesFileDirectory = "_series"
type Store struct {
	mu                sync.RWMutex
	shards            map[uint64]*Shard
	databases         map[string]struct{}
	sfiles            map[string]*SeriesFile
	SeriesFileMaxSize int64  
	path              string
	indexes map[string]interface{}
	pendingShardDeletes map[uint64]struct{}
	EngineOptions EngineOptions
	baseLogger *zap.Logger
	Logger     *zap.Logger
	closing chan struct{}
	wg      sync.WaitGroup
	opened  bool
}
func NewStore(path string) *Store {
	logger := zap.NewNop()
	return &Store{
		databases:           make(map[string]struct{}),
		path:                path,
		sfiles:              make(map[string]*SeriesFile),
		indexes:             make(map[string]interface{}),
		pendingShardDeletes: make(map[uint64]struct{}),
		EngineOptions:       NewEngineOptions(),
		Logger:              logger,
		baseLogger:          logger,
	}
}
func (s *Store) WithLogger(log *zap.Logger) {
	s.baseLogger = log
	s.Logger = log.With(zap.String("service", "store"))
	for _, sh := range s.shards {
		sh.WithLogger(s.baseLogger)
	}
}
func (s *Store) Statistics(tags map[string]string) []models.Statistic {
	s.mu.RLock()
	shards := s.shardsSlice()
	s.mu.RUnlock()
	databases := s.Databases()
	statistics := make([]models.Statistic, 0, len(databases))
	for _, database := range databases {
		sc, err := s.SeriesCardinality(database)
		if err != nil {
			s.Logger.Info("Cannot retrieve series cardinality", zap.Error(err))
			continue
		}
		mc, err := s.MeasurementsCardinality(database)
		if err != nil {
			s.Logger.Info("Cannot retrieve measurement cardinality", zap.Error(err))
			continue
		}
		statistics = append(statistics, models.Statistic{
			Name: "database",
			Tags: models.StatisticTags{"database": database}.Merge(tags),
			Values: map[string]interface{}{
				statDatabaseSeries:       sc,
				statDatabaseMeasurements: mc,
			},
		})
	}
	for _, shard := range shards {
		statistics = append(statistics, shard.Statistics(tags)...)
	}
	return statistics
}
func (s *Store) Path() string { return s.path }
func (s *Store) Open() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.opened {
		return nil
	}
	s.closing = make(chan struct{})
	s.shards = map[uint64]*Shard{}
	s.Logger.Info("Using data dir", zap.String("path", s.Path()))
	if err := os.MkdirAll(s.path, 0777); err != nil {
		return err
	}
	if err := s.loadShards(); err != nil {
		return err
	}
	s.opened = true
	s.wg.Add(1)
	go s.monitorShards()
	return nil
}
func (s *Store) loadShards() error {
	type res struct {
		s   *Shard
		err error
	}
	lim := s.EngineOptions.Config.MaxConcurrentCompactions
	if lim == 0 {
		lim = runtime.GOMAXPROCS(0) / 2  
		if lim > 4 {
			lim = 4
		}
		if lim < 1 {
			lim = 1
		}
	}
	if lim > runtime.GOMAXPROCS(0) {
		lim = runtime.GOMAXPROCS(0)
	}
	s.EngineOptions.CompactionLimiter = limiter.NewFixed(lim)
	if os.Getenv("INFLUXDB_DATA_COMPACTION_THROUGHPUT") == "" {
		s.EngineOptions.CompactionThroughputLimiter = limiter.NewRate(48*1024*1024, 48*1024*1024)
	} else {
		s.Logger.Info("Compaction throughput limit disabled")
	}
	log, logEnd := logger.NewOperation(s.Logger, "Open store", "tsdb_open")
	defer logEnd()
	t := limiter.NewFixed(runtime.GOMAXPROCS(0))
	resC := make(chan *res)
	var n int
	dbDirs, err := ioutil.ReadDir(s.path)
	if err != nil {
		return err
	}
	for _, db := range dbDirs {
		dbPath := filepath.Join(s.path, db.Name())
		if !db.IsDir() {
			log.Info("Skipping database dir", zap.String("name", db.Name()), zap.String("reason", "not a directory"))
			continue
		}
		sfile, err := s.openSeriesFile(db.Name())
		if err != nil {
			return err
		}
		idx, err := s.createIndexIfNotExists(db.Name())
		if err != nil {
			return err
		}
		rpDirs, err := ioutil.ReadDir(dbPath)
		if err != nil {
			return err
		}
		for _, rp := range rpDirs {
			rpPath := filepath.Join(s.path, db.Name(), rp.Name())
			if !rp.IsDir() {
				log.Info("Skipping retention policy dir", zap.String("name", rp.Name()), zap.String("reason", "not a directory"))
				continue
			}
			if rp.Name() == SeriesFileDirectory {
				continue
			}
			shardDirs, err := ioutil.ReadDir(rpPath)
			if err != nil {
				return err
			}
			for _, sh := range shardDirs {
				n++
				go func(db, rp, sh string) {
					t.Take()
					defer t.Release()
					start := time.Now()
					path := filepath.Join(s.path, db, rp, sh)
					walPath := filepath.Join(s.EngineOptions.Config.WALDir, db, rp, sh)
					shardID, err := strconv.ParseUint(sh, 10, 64)
					if err != nil {
						log.Info("invalid shard ID found at path", zap.String("path", path))
						resC <- &res{err: fmt.Errorf("%s is not a valid ID. Skipping shard.", sh)}
						return
					}
					opt := s.EngineOptions
					opt.InmemIndex = idx
					opt.SeriesIDSets = shardSet{store: s, db: db}
					if _, err := os.Stat(filepath.Join(path, "index")); os.IsNotExist(err) {
						opt.IndexVersion = "inmem"
					}
					shard := NewShard(shardID, path, walPath, sfile, opt)
					shard.EnableOnOpen = false
					shard.WithLogger(s.baseLogger)
					err = shard.Open()
					if err != nil {
						log.Info("Failed to open shard", logger.Shard(shardID), zap.Error(err))
						resC <- &res{err: fmt.Errorf("Failed to open shard: %d: %s", shardID, err)}
						return
					}
					resC <- &res{s: shard}
					log.Info("Opened shard", zap.String("path", path), zap.Duration("duration", time.Since(start)))
				}(db.Name(), rp.Name(), sh.Name())
			}
		}
	}
	for i := 0; i < n; i++ {
		res := <-resC
		if res.err != nil {
			continue
		}
		s.shards[res.s.id] = res.s
		s.databases[res.s.database] = struct{}{}
	}
	close(resC)
	for _, sh := range s.shards {
		sh.SetEnabled(true)
		if sh.IsIdle() {
			if err := sh.Free(); err != nil {
				return err
			}
		}
	}
	return nil
}
func (s *Store) Close() error {
	s.mu.Lock()
	if s.opened {
		close(s.closing)
	}
	s.mu.Unlock()
	s.wg.Wait()
	if err := s.walkShards(s.shardsSlice(), func(sh *Shard) error {
		return sh.Close()
	}); err != nil {
		return err
	}
	s.mu.Lock()
	for _, sfile := range s.sfiles {
		if err := sfile.Close(); err != nil {
			s.mu.Unlock()
			return err
		}
	}
	s.shards = nil
	s.sfiles = map[string]*SeriesFile{}
	s.opened = false  
	s.mu.Unlock()
	return nil
}
func (s *Store) openSeriesFile(database string) (*SeriesFile, error) {
	if sfile := s.sfiles[database]; sfile != nil {
		return sfile, nil
	}
	sfile := NewSeriesFile(filepath.Join(s.path, database, SeriesFileDirectory))
	sfile.Logger = s.baseLogger
	if err := sfile.Open(); err != nil {
		return nil, err
	}
	s.sfiles[database] = sfile
	return sfile, nil
}
func (s *Store) seriesFile(database string) *SeriesFile {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.sfiles[database]
}
func (s *Store) createIndexIfNotExists(name string) (interface{}, error) {
	if idx := s.indexes[name]; idx != nil {
		return idx, nil
	}
	sfile, err := s.openSeriesFile(name)
	if err != nil {
		return nil, err
	}
	idx, err := NewInmemIndex(name, sfile)
	if err != nil {
		return nil, err
	}
	s.indexes[name] = idx
	return idx, nil
}
func (s *Store) Shard(id uint64) *Shard {
	s.mu.RLock()
	defer s.mu.RUnlock()
	sh, ok := s.shards[id]
	if !ok {
		return nil
	}
	return sh
}
func (s *Store) Shards(ids []uint64) []*Shard {
	s.mu.RLock()
	defer s.mu.RUnlock()
	a := make([]*Shard, 0, len(ids))
	for _, id := range ids {
		sh, ok := s.shards[id]
		if !ok {
			continue
		}
		a = append(a, sh)
	}
	return a
}
func (s *Store) ShardGroup(ids []uint64) ShardGroup {
	return Shards(s.Shards(ids))
}
func (s *Store) ShardN() int {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return len(s.shards)
}
func (s *Store) ShardDigest(id uint64) (io.ReadCloser, int64, error) {
	sh := s.Shard(id)
	if sh == nil {
		return nil, 0, ErrShardNotFound
	}
	return sh.Digest()
}