func (e *Engine) disableLevelCompactions(wait bool) {
	e.mu.Lock()
	old := e.levelWorkers
	if wait {
		e.levelWorkers += 1
	}
	waitCh := e.done
	wg := e.wg
	if old == 0 && e.done != nil {
		select {
		case <-e.done:
			e.mu.Unlock()
			return
		default:
		}
		e.Compactor.DisableCompactions()
		close(e.done)
		e.mu.Unlock()
		wg.Wait()
		e.mu.Lock()
		e.done = nil
		e.mu.Unlock()
		return
	}
	e.mu.Unlock()
	if waitCh == nil {
		return
	}
	<-waitCh
	wg.Wait()
}
disableLevelCompactions will stop level compactions before returning.

If 'wait' is set to true, then a corresponding call to enableLevelCompactions(true) will be
required before level compactions will start back up again.
package tsm1  
import (
	"archive/tar"
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"math"
	"os"
	"path/filepath"
	"regexp"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"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/pkg/limiter"
	"github.com/influxdata/influxdb/pkg/metrics"
	intar "github.com/influxdata/influxdb/pkg/tar"
	"github.com/influxdata/influxdb/pkg/tracing"
	"github.com/influxdata/influxdb/query"
	"github.com/influxdata/influxdb/tsdb"
	_ "github.com/influxdata/influxdb/tsdb/index"
	"github.com/influxdata/influxdb/tsdb/index/inmem"
	"github.com/influxdata/influxdb/tsdb/index/tsi1"
	"github.com/influxdata/influxql"
	"go.uber.org/zap"
)
func init() {
	tsdb.RegisterEngine("tsm1", NewEngine)
}
var (
	_ tsdb.Engine = &Engine{}
	timeBytes              = []byte("time")
	keyFieldSeparatorBytes = []byte(keyFieldSeparator)
	emptyBytes             = []byte{}
)
var (
	tsmGroup                   = metrics.MustRegisterGroup("tsm1")
	numberOfRefCursorsCounter  = metrics.MustRegisterCounter("cursors_ref", metrics.WithGroup(tsmGroup))
	numberOfAuxCursorsCounter  = metrics.MustRegisterCounter("cursors_aux", metrics.WithGroup(tsmGroup))
	numberOfCondCursorsCounter = metrics.MustRegisterCounter("cursors_cond", metrics.WithGroup(tsmGroup))
	planningTimer              = metrics.MustRegisterTimer("planning_time", metrics.WithGroup(tsmGroup))
)
func NewContextWithMetricsGroup(ctx context.Context) context.Context {
	group := metrics.NewGroup(tsmGroup)
	return metrics.NewContextWithGroup(ctx, group)
}
func MetricsGroupFromContext(ctx context.Context) *metrics.Group {
	return metrics.GroupFromContext(ctx)
}
const (
	keyFieldSeparator = "#!~#"
	deleteFlushThreshold = 50 * 1024 * 1024
)
const (
	statCacheCompactions        = "cacheCompactions"
	statCacheCompactionsActive  = "cacheCompactionsActive"
	statCacheCompactionError    = "cacheCompactionErr"
	statCacheCompactionDuration = "cacheCompactionDuration"
	statTSMLevel1Compactions        = "tsmLevel1Compactions"
	statTSMLevel1CompactionsActive  = "tsmLevel1CompactionsActive"
	statTSMLevel1CompactionError    = "tsmLevel1CompactionErr"
	statTSMLevel1CompactionDuration = "tsmLevel1CompactionDuration"
	statTSMLevel1CompactionQueue    = "tsmLevel1CompactionQueue"
	statTSMLevel2Compactions        = "tsmLevel2Compactions"
	statTSMLevel2CompactionsActive  = "tsmLevel2CompactionsActive"
	statTSMLevel2CompactionError    = "tsmLevel2CompactionErr"
	statTSMLevel2CompactionDuration = "tsmLevel2CompactionDuration"
	statTSMLevel2CompactionQueue    = "tsmLevel2CompactionQueue"
	statTSMLevel3Compactions        = "tsmLevel3Compactions"
	statTSMLevel3CompactionsActive  = "tsmLevel3CompactionsActive"
	statTSMLevel3CompactionError    = "tsmLevel3CompactionErr"
	statTSMLevel3CompactionDuration = "tsmLevel3CompactionDuration"
	statTSMLevel3CompactionQueue    = "tsmLevel3CompactionQueue"
	statTSMOptimizeCompactions        = "tsmOptimizeCompactions"
	statTSMOptimizeCompactionsActive  = "tsmOptimizeCompactionsActive"
	statTSMOptimizeCompactionError    = "tsmOptimizeCompactionErr"
	statTSMOptimizeCompactionDuration = "tsmOptimizeCompactionDuration"
	statTSMOptimizeCompactionQueue    = "tsmOptimizeCompactionQueue"
	statTSMFullCompactions        = "tsmFullCompactions"
	statTSMFullCompactionsActive  = "tsmFullCompactionsActive"
	statTSMFullCompactionError    = "tsmFullCompactionErr"
	statTSMFullCompactionDuration = "tsmFullCompactionDuration"
	statTSMFullCompactionQueue    = "tsmFullCompactionQueue"
)
type Engine struct {
	mu sync.RWMutex
	index tsdb.Index
	wg           *sync.WaitGroup  
	done         chan struct{}    
	levelWorkers int              
	snapDone chan struct{}    
	snapWG   *sync.WaitGroup  
	id           uint64
	database     string
	path         string
	sfile        *tsdb.SeriesFile
	logger       *zap.Logger  
	traceLogger  *zap.Logger  
	traceLogging bool
	fieldset *tsdb.MeasurementFieldSet
	WAL            *WAL
	Cache          *Cache
	Compactor      *Compactor
	CompactionPlan CompactionPlanner
	FileStore      *FileStore
	MaxPointsPerBlock int
	CacheFlushMemorySizeThreshold uint64
	CacheFlushWriteColdDuration time.Duration
	enableCompactionsOnOpen bool
	stats *EngineStatistics
	compactionLimiter limiter.Fixed
	scheduler *scheduler
	seriesIDSets tsdb.SeriesIDSets
}
func NewEngine(id uint64, idx tsdb.Index, database, path string, walPath string, sfile *tsdb.SeriesFile, opt tsdb.EngineOptions) tsdb.Engine {
	w := NewWAL(walPath)
	w.syncDelay = time.Duration(opt.Config.WALFsyncDelay)
	fs := NewFileStore(path)
	cache := NewCache(uint64(opt.Config.CacheMaxMemorySize), path)
	c := &Compactor{
		Dir:       path,
		FileStore: fs,
		RateLimit: opt.CompactionThroughputLimiter,
	}
	logger := zap.NewNop()
	stats := &EngineStatistics{}
	e := &Engine{
		id:           id,
		database:     database,
		path:         path,
		index:        idx,
		sfile:        sfile,
		logger:       logger,
		traceLogger:  logger,
		traceLogging: opt.Config.TraceLoggingEnabled,
		WAL:   w,
		Cache: cache,
		FileStore:      fs,
		Compactor:      c,
		CompactionPlan: NewDefaultPlanner(fs, time.Duration(opt.Config.CompactFullWriteColdDuration)),
		CacheFlushMemorySizeThreshold: uint64(opt.Config.CacheSnapshotMemorySize),
		CacheFlushWriteColdDuration:   time.Duration(opt.Config.CacheSnapshotWriteColdDuration),
		enableCompactionsOnOpen:       true,
		stats:             stats,
		compactionLimiter: opt.CompactionLimiter,
		scheduler:         newScheduler(stats, opt.CompactionLimiter.Capacity()),
		seriesIDSets:      opt.SeriesIDSets,
	}
	if e.traceLogging {
		fs.enableTraceLogging(true)
		w.enableTraceLogging(true)
	}
	return e
}
func (e *Engine) Digest() (io.ReadCloser, int64, error) {
	digestPath := filepath.Join(e.path, "digest.tsd")
	f, err := os.Open(digestPath)
	if err == nil {
		fi, err := f.Stat()
		if err != nil {
			f.Close()
			return nil, 0, err
		}
		if !e.LastModified().After(fi.ModTime()) {
			fi, err := f.Stat()
			if err != nil {
				f.Close()
				return nil, 0, err
			}
			return f, fi.Size(), nil
		}
		if err := f.Close(); err != nil {
			return nil, 0, err
		}
	}
	tf, err := os.Create(digestPath + ".tmp")
	if err != nil {
		return nil, 0, err
	}
	if err := Digest(e.path, tf); err != nil {
		tf.Close()
		os.Remove(tf.Name())
		return nil, 0, err
	}
	if err := renameFile(tf.Name(), digestPath); err != nil {
		return nil, 0, err
	}
	f, err = os.Open(digestPath)
	if err != nil {
		return nil, 0, err
	}
	fi, err := f.Stat()
	if err != nil {
		f.Close()
		return nil, 0, err
	}
	return f, fi.Size(), nil
}
func (e *Engine) SetEnabled(enabled bool) {
	e.enableCompactionsOnOpen = enabled
	e.SetCompactionsEnabled(enabled)
}
func (e *Engine) SetCompactionsEnabled(enabled bool) {
	if enabled {
		e.enableSnapshotCompactions()
		e.enableLevelCompactions(false)
	} else {
		e.disableSnapshotCompactions()
		e.disableLevelCompactions(false)
	}
}
func (e *Engine) enableLevelCompactions(wait bool) {
	if !wait {
		e.mu.RLock()
		if e.done != nil {
			e.mu.RUnlock()
			return
		}
		e.mu.RUnlock()
	}
	e.mu.Lock()
	if wait {
		e.levelWorkers -= 1
	}
	if e.levelWorkers != 0 || e.done != nil {
		e.mu.Unlock()
		return
	}
	e.Compactor.EnableCompactions()
	e.done = make(chan struct{})
	wg := new(sync.WaitGroup)
	wg.Add(1)
	e.wg = wg
	e.mu.Unlock()
	go func() { defer wg.Done(); e.compact(wg) }()
}