func (w *NoopTSMRewriter) Close() error {
	return nil
}
Close implements Rewriter interface
package storage
import (
	"log"
	"os"
	"github.com/influxdata/influxdb/pkg/limiter"
	"github.com/influxdata/influxdb/tsdb/engine/tsm1"
)
type TSMRewriter interface {
	Write(key []byte, values []tsm1.Value) error
	WriteSnapshot() error
	CompactFull() ([]string, error)
	Close() error
}
type CachedTSMRewriter struct {
	cache              *tsm1.Cache
	flushSizeThreshold uint64
	compactor *tsm1.Compactor
	fileStore *tsm1.FileStore
	path     string
	tsmFiles []string
}
func NewCachedTSMRewriter(maxSize uint64, flushSizeThrehsold uint64, path string) *CachedTSMRewriter {
	cache := tsm1.NewCache(maxSize)
	fs := tsm1.NewFileStore(path)
	compactor := tsm1.NewCompactor()
	compactor.Dir = path
	compactor.FileStore = fs
	compactor.RateLimit = limiter.NewRate(DefaultCompactThroughput, DefaultCompactThroughputBurst)
	compactor.Open()
	return &CachedTSMRewriter{
		cache:              cache,
		flushSizeThreshold: flushSizeThrehsold,
		path:               path,
		compactor:          compactor,
		fileStore:          fs,
	}
}
func (w *CachedTSMRewriter) Write(key []byte, values []tsm1.Value) error {
	if err := w.cache.Write(key, values); err != nil {
		return err
	}
	sz := w.cache.Size()
	if sz > w.flushSizeThreshold {
		return w.WriteSnapshot()
	}
	return nil
}
func (w *CachedTSMRewriter) WriteSnapshot() error {
	log.Printf("snapshoting cache")
	snapshot, err := w.cache.Snapshot()
	if err != nil {
		return err
	}
	if snapshot.Size() == 0 {
		w.cache.ClearSnapshot(true)
		return nil
	}
	snapshot.Deduplicate()
	defer func() {
		if err != nil {
			w.cache.ClearSnapshot(false)
		}
	}()
	newFiles, err := w.compactor.WriteSnapshot(snapshot)
	if err != nil {
		return err
	}
	err = w.fileStore.ReplaceWithCallback(nil, newFiles, func(r []tsm1.TSMFile) {
		for _, f := range r {
			log.Printf("wrote new TSM file '%s'\n", f.Path())
			w.tsmFiles = append(w.tsmFiles, f.Path())
		}
	})
	if err != nil {
		log.Println("error adding new TSM files from snapshot. Removing temporary files.")
		for _, file := range newFiles {
			if err := os.Remove(file); err != nil {
				return err
			}
		}
		return err
	}
	w.cache.ClearSnapshot(true)
	return nil
}
func (w *CachedTSMRewriter) CompactFull() ([]string, error) {
	if len(w.tsmFiles) == 0 {
		log.Println("skipping full compaction. No TSM files have been written")
		return nil, nil
	}
	files, err := w.compactor.CompactFull(w.tsmFiles)
	if err != nil {
		return nil, err
	}
	var newFiles []string
	err = w.fileStore.ReplaceWithCallback(w.tsmFiles, files, func(r []tsm1.TSMFile) {
		for _, f := range r {
			newFiles = append(newFiles, f.Path())
		}
	})
	if err != nil {
		log.Println("error adding new TSM files from full compaction. Removing temporary files.")
		for _, file := range files {
			if err := os.Remove(file); err != nil {
				return nil, err
			}
		}
		return nil, err
	}
	return newFiles, nil
}
func (w *CachedTSMRewriter) Close() error {
	return os.RemoveAll(w.path)
}
type NoopTSMRewriter struct {
}
func (w *NoopTSMRewriter) Write(key []byte, values []tsm1.Value) error {
	return nil
}
func (w *NoopTSMRewriter) WriteSnapshot() error {
	return nil
}
func (w *NoopTSMRewriter) CompactFull() ([]string, error) {
	return nil, nil
}