func (i *Partition) compactToLevel(files []*IndexFile, level int, interrupt <-chan struct{}) {
	assert(len(files) >= 2, "at least two index files are required for compaction")
	assert(level > 0, "cannot compact level zero")
	log, logEnd := logger.NewOperation(i.logger, "TSI level compaction", "tsi1_compact_to_level", zap.Int("tsi1_level", level))
	defer logEnd()
	select {
	case <-interrupt:
		log.Error("Cannot begin compaction", zap.Error(ErrCompactionInterrupted))
		return
	default:
	}
	var once sync.Once
	defer once.Do(func() { IndexFiles(files).Release() })
	start := time.Now()
	path := filepath.Join(i.path, FormatIndexFileName(i.NextSequence(), level))
	f, err := os.Create(path)
	if err != nil {
		log.Error("Cannot create compaction files", zap.Error(err))
		return
	}
	defer f.Close()
	log.Info("Performing full compaction",
		zap.String("src", joinIntSlice(IndexFiles(files).IDs(), ",")),
		zap.String("dst", path),
	)
	lvl := i.levels[level]
	n, err := IndexFiles(files).CompactTo(f, i.sfile, lvl.M, lvl.K, interrupt)
	if err != nil {
		log.Error("Cannot compact index files", zap.Error(err))
		return
	}
	if err := f.Close(); err != nil {
		log.Error("Error closing index file", zap.Error(err))
		return
	}
	file := NewIndexFile(i.sfile)
	file.SetPath(path)
	if err := file.Open(); err != nil {
		log.Error("Cannot open new index file", zap.Error(err))
		return
	}
	if err := func() error {
		i.mu.Lock()
		defer i.mu.Unlock()
		i.fileSet = i.fileSet.MustReplace(IndexFiles(files).Files(), file)
		manifestSize, err := i.Manifest().Write()
		if err != nil {
			return err
		}
		i.manifestSize = manifestSize
		return nil
	}(); err != nil {
		log.Error("Cannot write manifest", zap.Error(err))
		return
	}
	elapsed := time.Since(start)
	log.Info("Full compaction complete",
		zap.String("path", path),
		logger.DurationLiteral("elapsed", elapsed),
		zap.Int64("bytes", n),
		zap.Int("kb_per_sec", int(float64(n)/elapsed.Seconds())/1024),
	)
	once.Do(func() { IndexFiles(files).Release() })
	for _, f := range files {
		log.Info("Removing index file", zap.String("path", f.Path()))
		if err := f.Close(); err != nil {
			log.Error("Cannot close index file", zap.Error(err))
			return
		} else if err := os.Remove(f.Path()); err != nil {
			log.Error("Cannot remove index file", zap.Error(err))
			return
		}
	}
}
compactToLevel compacts a set of files into a new file. Replaces old files with
compacted file on successful completion. This runs in a separate goroutine.
package tsi1
import (
	"encoding/json"
	"errors"
	"fmt"
	"io/ioutil"
	"os"
	"path/filepath"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"
	"github.com/influxdata/influxdb/logger"
	"github.com/influxdata/influxdb/models"
	"github.com/influxdata/influxdb/pkg/bytesutil"
	"github.com/influxdata/influxdb/pkg/estimator"
	"github.com/influxdata/influxdb/tsdb"
	"github.com/influxdata/influxql"
	"go.uber.org/zap"
)
const Version = 1
const (
	LogFileExt   = ".tsl"
	IndexFileExt = ".tsi"
	CompactingExt = ".compacting"
)
const ManifestFileName = "MANIFEST"
type Partition struct {
	mu     sync.RWMutex
	opened bool
	sfile         *tsdb.SeriesFile  
	activeLogFile *LogFile          
	fileSet       *FileSet          
	seq           int               
	seriesIDSet *tsdb.SeriesIDSet
	levels          []CompactionLevel  
	levelCompacting []bool             
	once    sync.Once
	closing chan struct{}  
	wg      sync.WaitGroup
	fieldset *tsdb.MeasurementFieldSet
	Database string
	path string
	id   string  
	MaxLogFileSize int64
	compactionInterrupt chan struct{}
	compactionsDisabled int
	logger *zap.Logger
	manifestSize int64
	version int
}
func NewPartition(sfile *tsdb.SeriesFile, path string) *Partition {
	return &Partition{
		closing:     make(chan struct{}),
		path:        path,
		sfile:       sfile,
		seriesIDSet: tsdb.NewSeriesIDSet(),
		MaxLogFileSize: tsdb.DefaultMaxIndexLogFileSize,
		compactionInterrupt: make(chan struct{}),
		logger:  zap.NewNop(),
		version: Version,
	}
}
var ErrIncompatibleVersion = errors.New("incompatible tsi1 index MANIFEST")
func (i *Partition) Open() error {
	i.mu.Lock()
	defer i.mu.Unlock()
	i.closing = make(chan struct{})
	if i.opened {
		return errors.New("index partition already open")
	}
	i.id = filepath.Base(i.path)
	_, err := strconv.Atoi(i.id)
	if err != nil {
		return err
	}
	if err := os.MkdirAll(i.path, 0777); err != nil {
		return err
	}
	m, manifestSize, err := ReadManifestFile(filepath.Join(i.path, ManifestFileName))
	if os.IsNotExist(err) {
		m = NewManifest(i.ManifestPath())
	} else if err != nil {
		return err
	}
	i.manifestSize = manifestSize
	if err := m.Validate(); err != nil {
		return err
	}
	i.levels = make([]CompactionLevel, len(m.Levels))
	copy(i.levels, m.Levels)
	i.levelCompacting = make([]bool, len(i.levels))
	var files []File
	for _, filename := range m.Files {
		switch filepath.Ext(filename) {
		case LogFileExt:
			f, err := i.openLogFile(filepath.Join(i.path, filename))
			if err != nil {
				return err
			}
			files = append(files, f)
			sz, _ := f.Stat()
			if i.activeLogFile == nil && sz < i.MaxLogFileSize {
				i.activeLogFile = f
			}
		case IndexFileExt:
			f, err := i.openIndexFile(filepath.Join(i.path, filename))
			if err != nil {
				return err
			}
			files = append(files, f)
		}
	}
	fs, err := NewFileSet(i.Database, i.levels, i.sfile, files)
	if err != nil {
		return err
	}
	i.fileSet = fs
	i.seq = i.fileSet.MaxID()
	if err := i.deleteNonManifestFiles(m); err != nil {
		return err
	}
	if i.activeLogFile == nil {
		if err := i.prependActiveLogFile(); err != nil {
			return err
		}
	}
	if err := i.buildSeriesSet(); err != nil {
		return err
	}
	i.opened = true
	i.compact()
	return nil
}
func (i *Partition) openLogFile(path string) (*LogFile, error) {
	f := NewLogFile(i.sfile, path)
	if err := f.Open(); err != nil {
		return nil, err
	}
	return f, nil
}
func (i *Partition) openIndexFile(path string) (*IndexFile, error) {
	f := NewIndexFile(i.sfile)
	f.SetPath(path)
	if err := f.Open(); err != nil {
		return nil, err
	}
	return f, nil
}
func (i *Partition) deleteNonManifestFiles(m *Manifest) error {
	dir, err := os.Open(i.path)
	if err != nil {
		return err
	}
	defer dir.Close()
	fis, err := dir.Readdir(-1)
	if err != nil {
		return err
	}
	for _, fi := range fis {
		filename := filepath.Base(fi.Name())
		if filename == ManifestFileName || m.HasFile(filename) {
			continue
		}
		if err := os.RemoveAll(filename); err != nil {
			return err
		}
	}
	return nil
}
func (p *Partition) buildSeriesSet() error {
	fs := p.retainFileSet()
	defer fs.Release()
	p.seriesIDSet = tsdb.NewSeriesIDSet()
	for i := len(fs.files) - 1; i >= 0; i-- {
		f := fs.files[i]
		ts, err := f.TombstoneSeriesIDSet()
		if err != nil {
			return err
		}
		p.seriesIDSet.Diff(ts)
		ss, err := f.SeriesIDSet()
		if err != nil {
			return err
		}
		p.seriesIDSet.Merge(ss)
	}
	return nil
}
func (i *Partition) Wait() {
	i.wg.Wait()
}
func (i *Partition) Close() error {
	i.once.Do(func() {
		close(i.closing)
		close(i.compactionInterrupt)
	})
	i.wg.Wait()
	i.mu.Lock()
	defer i.mu.Unlock()
	for _, f := range i.fileSet.files {
		f.Close()
	}
	i.fileSet.files = nil
	return nil
}
func (p *Partition) isClosing() bool {
	select {
	case <-p.closing:
		return true
	default:
		return false
	}
}
func (i *Partition) Path() string { return i.path }
func (i *Partition) SeriesFile() *tsdb.SeriesFile { return i.sfile }
func (i *Partition) NextSequence() int {
	i.mu.Lock()
	defer i.mu.Unlock()
	return i.nextSequence()
}
func (i *Partition) nextSequence() int {
	i.seq++
	return i.seq
}
func (i *Partition) ManifestPath() string {
	return filepath.Join(i.path, ManifestFileName)
}
func (i *Partition) Manifest() *Manifest {
	m := &Manifest{
		Levels:  i.levels,
		Files:   make([]string, len(i.fileSet.files)),
		Version: i.version,
		path:    i.ManifestPath(),
	}
	for j, f := range i.fileSet.files {
		m.Files[j] = filepath.Base(f.Path())
	}
	return m
}
func (i *Partition) WithLogger(logger *zap.Logger) {
	i.logger = logger.With(zap.String("index", "tsi"))
}
func (i *Partition) SetFieldSet(fs *tsdb.MeasurementFieldSet) {
	i.mu.Lock()
	i.fieldset = fs
	i.mu.Unlock()
}
func (i *Partition) FieldSet() *tsdb.MeasurementFieldSet {
	i.mu.Lock()
	fs := i.fieldset
	i.mu.Unlock()
	return fs
}
func (i *Partition) RetainFileSet() (*FileSet, error) {
	select {
	case <-i.closing:
		return nil, errors.New("index is closing")
	default:
		i.mu.RLock()
		defer i.mu.RUnlock()
		return i.retainFileSet(), nil
	}
}
func (i *Partition) retainFileSet() *FileSet {
	fs := i.fileSet
	fs.Retain()
	return fs
}
func (i *Partition) FileN() int { return len(i.fileSet.files) }
func (i *Partition) prependActiveLogFile() error {
	f, err := i.openLogFile(filepath.Join(i.path, FormatLogFileName(i.nextSequence())))
	if err != nil {
		return err
	}
	i.activeLogFile = f
	i.fileSet = i.fileSet.PrependLogFile(f)
	manifestSize, err := i.Manifest().Write()
	if err != nil {
		return err
	}
	i.manifestSize = manifestSize
	return nil
}
func (i *Partition) ForEachMeasurementName(fn func(name []byte) error) error {
	fs, err := i.RetainFileSet()
	if err != nil {
		return err
	}
	defer fs.Release()
	itr := fs.MeasurementIterator()
	if itr == nil {
		return nil
	}
	for e := itr.Next(); e != nil; e = itr.Next() {
		if err := fn(e.Name()); err != nil {
			return err
		}
	}
	return nil
}
func (p *Partition) MeasurementHasSeries(name []byte) (bool, error) {
	fs, err := p.RetainFileSet()
	if err != nil {
		return false, err
	}
	defer fs.Release()
	for _, f := range fs.files {
		if f.MeasurementHasSeries(p.seriesIDSet, name) {
			return true, nil
		}
	}
	return false, nil
}
func (i *Partition) MeasurementIterator() (tsdb.MeasurementIterator, error) {
	fs, err := i.RetainFileSet()
	if err != nil {
		return nil, err
	}
	itr := fs.MeasurementIterator()
	if itr == nil {
		fs.Release()
		return nil, nil
	}
	return newFileSetMeasurementIterator(fs, NewTSDBMeasurementIteratorAdapter(itr)), nil
}
func (i *Partition) MeasurementExists(name []byte) (bool, error) {
	fs, err := i.RetainFileSet()
	if err != nil {
		return false, err
	}
	defer fs.Release()
	m := fs.Measurement(name)
	return m != nil && !m.Deleted(), nil
}
func (i *Partition) MeasurementNamesByRegex(re *regexp.Regexp) ([][]byte, error) {
	fs, err := i.RetainFileSet()
	if err != nil {
		return nil, err
	}
	defer fs.Release()
	itr := fs.MeasurementIterator()
	if itr == nil {
		return nil, nil
	}
	var a [][]byte
	for e := itr.Next(); e != nil; e = itr.Next() {
		if re.Match(e.Name()) {
			a = append(a, bytesutil.Clone(e.Name()))
		}
	}
	return a, nil
}
func (i *Partition) MeasurementSeriesIDIterator(name []byte) (tsdb.SeriesIDIterator, error) {
	fs, err := i.RetainFileSet()
	if err != nil {
		return nil, err
	}
	return newFileSetSeriesIDIterator(fs, fs.MeasurementSeriesIDIterator(name)), nil
}
func (i *Partition) DropMeasurement(name []byte) error {
	fs, err := i.RetainFileSet()
	if err != nil {
		return err
	}
	defer fs.Release()
	if kitr := fs.TagKeyIterator(name); kitr != nil {
		for k := kitr.Next(); k != nil; k = kitr.Next() {
			if !k.Deleted() {
				if err := func() error {
					i.mu.RLock()
					defer i.mu.RUnlock()
					return i.activeLogFile.DeleteTagKey(name, k.Key())
				}(); err != nil {
					return err
				}
			}
			if vitr := k.TagValueIterator(); vitr != nil {
				for v := vitr.Next(); v != nil; v = vitr.Next() {
					if !v.Deleted() {
						if err := func() error {
							i.mu.RLock()
							defer i.mu.RUnlock()
							return i.activeLogFile.DeleteTagValue(name, k.Key(), v.Value())
						}(); err != nil {
							return err
						}
					}
				}
			}
		}
	}
	if itr := fs.MeasurementSeriesIDIterator(name); itr != nil {
		defer itr.Close()
		for {
			elem, err := itr.Next()
			if err != nil {
				return err
			} else if elem.SeriesID == 0 {
				break
			}
			if err := i.activeLogFile.DeleteSeriesID(elem.SeriesID); err != nil {
				return err
			}
		}
	}
	if err := func() error {
		i.mu.RLock()
		defer i.mu.RUnlock()
		return i.activeLogFile.DeleteMeasurement(name)
	}(); err != nil {
		return err
	}
	if err := i.CheckLogFile(); err != nil {
		return err
	}
	return nil
}
func (i *Partition) createSeriesListIfNotExists(names [][]byte, tagsSlice []models.Tags) error {
	if len(names) == 0 {
		return nil
	} else if len(names) != len(tagsSlice) {
		return fmt.Errorf("uneven batch, partition %s sent %d names and %d tags", i.id, len(names), len(tagsSlice))
	}
	fs, err := i.RetainFileSet()
	if err != nil {
		return err
	}
	defer fs.Release()
	i.mu.RLock()
	if err := i.activeLogFile.AddSeriesList(i.seriesIDSet, names, tagsSlice); err != nil {
		i.mu.RUnlock()
		return err
	}
	i.mu.RUnlock()
	return i.CheckLogFile()
}
func (i *Partition) DropSeries(seriesID uint64) error {
	if err := i.activeLogFile.DeleteSeriesID(seriesID); err != nil {
		return err
	}
	i.seriesIDSet.Remove(seriesID)
	return i.CheckLogFile()
}
func (i *Partition) MeasurementsSketches() (estimator.Sketch, estimator.Sketch, error) {
	fs, err := i.RetainFileSet()
	if err != nil {
		return nil, nil, err
	}
	defer fs.Release()
	return fs.MeasurementsSketches()
}
func (i *Partition) SeriesSketches() (estimator.Sketch, estimator.Sketch, error) {
	fs, err := i.RetainFileSet()
	if err != nil {
		return nil, nil, err
	}
	defer fs.Release()
	return fs.SeriesSketches()
}
func (i *Partition) HasTagKey(name, key []byte) (bool, error) {
	fs, err := i.RetainFileSet()
	if err != nil {
		return false, err
	}
	defer fs.Release()
	return fs.HasTagKey(name, key), nil
}
func (i *Partition) HasTagValue(name, key, value []byte) (bool, error) {
	fs, err := i.RetainFileSet()
	if err != nil {
		return false, err
	}
	defer fs.Release()
	return fs.HasTagValue(name, key, value), nil
}
func (i *Partition) TagKeyIterator(name []byte) tsdb.TagKeyIterator {
	fs, err := i.RetainFileSet()
	if err != nil {
		return nil  
	}
	itr := fs.TagKeyIterator(name)
	if itr == nil {
		fs.Release()
		return nil
	}
	return newFileSetTagKeyIterator(fs, NewTSDBTagKeyIteratorAdapter(itr))
}
func (i *Partition) TagValueIterator(name, key []byte) tsdb.TagValueIterator {
	fs, err := i.RetainFileSet()
	if err != nil {
		return nil  
	}
	itr := fs.TagValueIterator(name, key)
	if itr == nil {
		fs.Release()
		return nil
	}
	return newFileSetTagValueIterator(fs, NewTSDBTagValueIteratorAdapter(itr))
}
func (i *Partition) TagKeySeriesIDIterator(name, key []byte) tsdb.SeriesIDIterator {
	fs, err := i.RetainFileSet()
	if err != nil {
		return nil  
	}
	itr := fs.TagKeySeriesIDIterator(name, key)
	if itr == nil {
		fs.Release()
		return nil
	}
	return newFileSetSeriesIDIterator(fs, itr)
}
func (i *Partition) TagValueSeriesIDIterator(name, key, value []byte) tsdb.SeriesIDIterator {
	fs, err := i.RetainFileSet()
	if err != nil {
		return nil  
	}
	itr := fs.TagValueSeriesIDIterator(name, key, value)
	if itr == nil {
		fs.Release()
		return nil
	}
	return newFileSetSeriesIDIterator(fs, itr)
}
func (i *Partition) MeasurementTagKeysByExpr(name []byte, expr influxql.Expr) (map[string]struct{}, error) {
	fs, err := i.RetainFileSet()
	if err != nil {
		return nil, err
	}
	defer fs.Release()
	return fs.MeasurementTagKeysByExpr(name, expr)
}
func (i *Partition) ForEachMeasurementTagKey(name []byte, fn func(key []byte) error) error {
	fs, err := i.RetainFileSet()
	if err != nil {
		return err
	}
	defer fs.Release()
	itr := fs.TagKeyIterator(name)
	if itr == nil {
		return nil
	}
	for e := itr.Next(); e != nil; e = itr.Next() {
		if err := fn(e.Key()); err != nil {
			return err
		}
	}
	return nil
}
func (i *Partition) TagKeyCardinality(name, key []byte) int {
	return 0
}
func (i *Partition) SetFieldName(measurement []byte, name string) {}
func (i *Partition) RemoveShard(shardID uint64)                   {}
func (i *Partition) AssignShard(k string, shardID uint64)         {}
func (i *Partition) Compact() {
	i.mu.Lock()
	defer i.mu.Unlock()
	i.compact()
}
func (i *Partition) DisableCompactions() {
	i.mu.Lock()
	defer i.mu.Unlock()
	i.compactionsDisabled++
	select {
	case <-i.closing:
		return
	default:
	}
	if i.compactionsDisabled == 0 {
		close(i.compactionInterrupt)
		i.compactionInterrupt = make(chan struct{})
	}
}
func (i *Partition) EnableCompactions() {
	i.mu.Lock()
	defer i.mu.Unlock()
	if i.compactionsEnabled() {
		return
	}
	i.compactionsDisabled--
}
func (i *Partition) compactionsEnabled() bool {
	return i.compactionsDisabled == 0
}
func (i *Partition) compact() {
	if i.isClosing() {
		return
	} else if !i.compactionsEnabled() {
		return
	}
	interrupt := i.compactionInterrupt
	fs := i.retainFileSet()
	defer fs.Release()
	minLevel, maxLevel := 1, len(i.levels)-2
	for level := minLevel; level <= maxLevel; level++ {
		if i.levelCompacting[level] {
			continue
		}
		files := fs.LastContiguousIndexFilesByLevel(level)
		if len(files) < 2 {
			continue
		} else if len(files) > MaxIndexMergeCount {
			files = files[len(files)-MaxIndexMergeCount:]
		}
		IndexFiles(files).Retain()
		i.levelCompacting[level] = true
		func(files []*IndexFile, level int) {
			i.wg.Add(1)
			go func() {
				defer i.wg.Done()
				i.compactToLevel(files, level+1, interrupt)
				i.mu.Lock()
				i.levelCompacting[level] = false
				i.mu.Unlock()
				i.Compact()
			}()
		}(files, level)
	}
}