func (s *Store) DeleteRetentionPolicy(database, name string) error {
	s.mu.RLock()
	if _, ok := s.databases[database]; !ok {
		s.mu.RUnlock()
		return nil
	}
	shards := s.filterShards(func(sh *Shard) bool {
		return sh.database == database && sh.retentionPolicy == name
	})
	s.mu.RUnlock()
	if err := s.walkShards(shards, func(sh *Shard) error {
		if sh.database != database || sh.retentionPolicy != name {
			return nil
		}
		return sh.Close()
	}); err != nil {
		return err
	}
	rpPath := filepath.Clean(filepath.Join(s.path, database, name))
	if filepath.Clean(s.path) != filepath.Dir(filepath.Dir(rpPath)) {
		return fmt.Errorf("invalid path for database '%s', retention policy '%s': %s", database, name, rpPath)
	}
	if err := os.RemoveAll(filepath.Join(s.path, database, name)); err != nil {
		return err
	}
	if err := os.RemoveAll(filepath.Join(s.EngineOptions.Config.WALDir, database, name)); err != nil {
		return err
	}
	s.mu.Lock()
	for _, sh := range shards {
		delete(s.shards, sh.id)
	}
	s.mu.Unlock()
	return nil
}
DeleteRetentionPolicy will close all shards associated with the
provided retention policy, remove the retention policy directories on
both the DB and WAL, and remove all shard files from disk.
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()
}
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
}
func (s *Store) CreateShardSnapshot(id uint64) (string, error) {
	sh := s.Shard(id)
	if sh == nil {
		return "", ErrShardNotFound
	}
	return sh.CreateSnapshot()
}
func (s *Store) SetShardEnabled(shardID uint64, enabled bool) error {
	sh := s.Shard(shardID)
	if sh == nil {
		return ErrShardNotFound
	}
	sh.SetEnabled(enabled)
	return nil
}
func (s *Store) DeleteShard(shardID uint64) error {
	sh := s.Shard(shardID)
	if sh == nil {
		return nil
	}
	s.mu.Lock()
	if _, ok := s.pendingShardDeletes[shardID]; ok {
		s.mu.Unlock()
		return nil
	}
	delete(s.shards, shardID)
	s.pendingShardDeletes[shardID] = struct{}{}
	s.mu.Unlock()
	defer func() {
		s.mu.Lock()
		defer s.mu.Unlock()
		delete(s.pendingShardDeletes, shardID)
	}()
	index, err := sh.Index()
	if err != nil {
		return err
	}
	var ss *SeriesIDSet
	if i, ok := index.(interface {
		SeriesIDSet() *SeriesIDSet
	}); ok {
		ss = i.SeriesIDSet()
	}
	db := sh.Database()
	if err := sh.Close(); err != nil {
		return err
	}
	shards := s.filterShards(byDatabase(db))
	s.walkShards(shards, func(sh *Shard) error {
		index, err := sh.Index()
		if err != nil {
			return err
		}
		if i, ok := index.(interface {
			SeriesIDSet() *SeriesIDSet
		}); ok {
			ss.Diff(i.SeriesIDSet())
		} else {
			return fmt.Errorf("unable to get series id set for index in shard at %s", sh.Path())
		}
		return nil
	})
	if ss.Cardinality() > 0 {
		sfile := s.seriesFile(db)
		if sfile != nil {
			ss.ForEach(func(id uint64) {
				sfile.DeleteSeriesID(id)
			})
		}
	}
	if err := os.RemoveAll(sh.path); err != nil {
		return err
	}
	return os.RemoveAll(sh.walPath)
}
func (s *Store) DeleteDatabase(name string) error {
	s.mu.RLock()
	if _, ok := s.databases[name]; !ok {
		s.mu.RUnlock()
		return nil
	}
	shards := s.filterShards(func(sh *Shard) bool {
		return sh.database == name
	})
	s.mu.RUnlock()
	if err := s.walkShards(shards, func(sh *Shard) error {
		if sh.database != name {
			return nil
		}
		return sh.Close()
	}); err != nil {
		return err
	}
	dbPath := filepath.Clean(filepath.Join(s.path, name))
	s.mu.Lock()
	defer s.mu.Unlock()
	sfile := s.sfiles[name]
	delete(s.sfiles, name)
	if sfile != nil {
		if err := sfile.Close(); err != nil {
			return err
		}
	}
	if filepath.Clean(s.path) != filepath.Dir(dbPath) {
		return fmt.Errorf("invalid database directory location for database '%s': %s", name, dbPath)
	}
	if err := os.RemoveAll(dbPath); err != nil {
		return err
	}
	if err := os.RemoveAll(filepath.Join(s.EngineOptions.Config.WALDir, name)); err != nil {
		return err
	}
	for _, sh := range shards {
		delete(s.shards, sh.id)
	}
	delete(s.databases, name)
	delete(s.indexes, name)
	return nil
}