func (s *RocksDBStore) Backup(metadata string) error {
	err := s.backupEngine.CreateNewBackupWithMetadata(s.db, metadata)
	if err != nil {
		return err
	}
	return nil
}
Backup uses the backupEngine to create backups with metadata. The backup directory has been
set up previously.
package rocks
import (
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
	"os"
	"time"
	"github.com/bbva/qed/metrics"
	"github.com/bbva/qed/rocksdb"
	"github.com/bbva/qed/storage"
	"github.com/bbva/qed/storage/pb"
	"github.com/bbva/qed/util"
)
type RocksDBStore struct {
	path string
	db   *rocksdb.DB
	stats *rocksdb.Statistics
	backupEngine *rocksdb.BackupEngine
	backupOpts   *rocksdb.Options
	restoreOpts  *rocksdb.RestoreOptions
	cfHandles rocksdb.ColumnFamilyHandles
	globalOpts *rocksdb.Options
	cfOpts []*rocksdb.Options
	blockCache *rocksdb.Cache
	ro *rocksdb.ReadOptions
	wo *rocksdb.WriteOptions
	metrics *rocksDBMetrics
}
type Options struct {
	Path             string
	EnableStatistics bool
	MaxTotalWalSize  uint64
	WALSizeLimitMB   uint64
	WALTtlSeconds    uint64
}
func DefaultOptions() *Options {
	return &Options{
		EnableStatistics: true,
		MaxTotalWalSize:  0,
		WALSizeLimitMB:   1 << 20,
		WALTtlSeconds:    0,
	}
}
func NewRocksDBStore(path string, ttl time.Duration) (*RocksDBStore, error) {
	opts := DefaultOptions()
	opts.Path = path
	opts.WALTtlSeconds = uint64(ttl.Seconds())
	return NewRocksDBStoreWithOpts(opts)
}
func NewRocksDBStoreWithOpts(opts *Options) (*RocksDBStore, error) {
	cfNames := []string{
		storage.DefaultTable.String(),
		storage.HyperTable.String(),
		storage.HyperCacheTable.String(),
		storage.HistoryTable.String(),
		storage.FSMStateTable.String(),
	}
	env := rocksdb.NewDefaultEnv()
	env.SetBackgroundThreads(5)
	env.SetHighPriorityBackgroundThreads(3)
	globalOpts := rocksdb.NewDefaultOptions()
	globalOpts.SetCreateIfMissing(true)
	globalOpts.SetCreateIfMissingColumnFamilies(true)
	globalOpts.SetMaxTotalWalSize(opts.MaxTotalWalSize)
	globalOpts.SetWalSizeLimitMb(opts.WALSizeLimitMB)
	globalOpts.SetWALTtlSeconds(opts.WALTtlSeconds)
	globalOpts.SetEnv(env)
	blockCache := rocksdb.NewLRUCache(8*1024*1024*1024, 0.4)  
	var stats *rocksdb.Statistics
	if opts.EnableStatistics {
		stats = rocksdb.NewStatistics()
		globalOpts.SetStatistics(stats)
	}
	cfOpts := []*rocksdb.Options{
		rocksdb.NewDefaultOptions(),
		getHyperTableOpts(blockCache),
		getHyperTableOpts(blockCache),  
		getHistoryTableOpts(blockCache),
		getFsmStateTableOpts(),
	}
	db, cfHandles, err := rocksdb.OpenDBColumnFamilies(opts.Path, globalOpts, cfNames, cfOpts)
	if err != nil {
		return nil, err
	}
	backupPath := opts.Path + "/backups"
	err = os.MkdirAll(backupPath, 0755)
	if err != nil {
		return nil, err
	}
	backupOpts := rocksdb.NewDefaultOptions()
	be, err := rocksdb.OpenBackupEngine(backupOpts, backupPath)
	if err != nil {
		return nil, err
	}
	restoreOpts := rocksdb.NewRestoreOptions()
	store := &RocksDBStore{
		path:         opts.Path,
		db:           db,
		stats:        stats,
		cfHandles:    cfHandles,
		blockCache:   blockCache,
		backupEngine: be,
		backupOpts:   backupOpts,
		restoreOpts:  restoreOpts,
		globalOpts:   globalOpts,
		cfOpts:       cfOpts,
		wo:           rocksdb.NewDefaultWriteOptions(),
		ro:           rocksdb.NewDefaultReadOptions(),
	}
	if stats != nil {
		store.metrics = newRocksDBMetrics(store)
	}
	return store, nil
}
func getHyperTableOpts(blockCache *rocksdb.Cache) *rocksdb.Options {
	bbto := rocksdb.NewDefaultBlockBasedTableOptions()
	bbto.SetFilterPolicy(rocksdb.NewFullBloomFilterPolicy(10))
	bbto.SetCacheIndexAndFilterBlocks(true)
	bbto.SetPinL0FilterAndIndexBlocksInCache(true)
	bbto.SetCacheIndexAndFilterBlocksWithHighPriority(true)
	bbto.SetPartitionFilters(true)
	bbto.SetPinTopLevelIndexAndFilterInCache(true)
	bbto.SetIndexType(rocksdb.KTwoLevelIndexSearchIndexType)
	bbto.SetBlockCache(blockCache)
	bbto.SetBlockSize(16 * 1024)
	opts := rocksdb.NewDefaultOptions()
	opts.SetBlockBasedTableFactory(bbto)
	opts.SetCompression(rocksdb.SnappyCompression)
	opts.SetMaxSubCompactions(2)
	opts.SetWriteBufferSize(64 * 1024 * 1024)  
	opts.SetMaxWriteBufferNumber(3)
	opts.SetMinWriteBufferNumberToMerge(2)
	opts.SetLevel0FileNumCompactionTrigger(8)
	opts.SetLevel0SlowdownWritesTrigger(24)
	opts.SetLevel0StopWritesTrigger(33)
	opts.SetTargetFileSizeBase(64 * 1024 * 1024)  
	opts.SetTargetFileSizeMultiplier(8)
	opts.SetMaxBytesForLevelBase(512 * 1024 * 1024)  
	opts.SetMaxBytesForLevelMultiplier(8)
	opts.SetNumLevels(7)
	opts.SetMaxBackgroundCompactions(8)
	opts.SetMaxBackgroundFlushes(1)
	return opts
}
func getHistoryTableOpts(blockCache *rocksdb.Cache) *rocksdb.Options {
	bbto := rocksdb.NewDefaultBlockBasedTableOptions()
	bbto.SetFilterPolicy(rocksdb.NewBloomFilterPolicy(10))  
	bbto.SetCacheIndexAndFilterBlocks(true)
	bbto.SetBlockCache(blockCache)
	bbto.SetBlockSize(16 * 1024)
	opts := rocksdb.NewDefaultOptions()
	opts.SetBlockBasedTableFactory(bbto)
	opts.SetCompression(rocksdb.SnappyCompression)
	opts.SetWriteBufferSize(64 * 1024 * 1024)  
	opts.SetMaxWriteBufferNumber(3)
	opts.SetMinWriteBufferNumberToMerge(1)
	opts.SetLevel0FileNumCompactionTrigger(8)
	opts.SetLevel0SlowdownWritesTrigger(17)
	opts.SetLevel0StopWritesTrigger(24)
	opts.SetTargetFileSizeBase(64 * 1024 * 1024)  
	opts.SetTargetFileSizeMultiplier(8)
	opts.SetMaxBytesForLevelBase(512 * 1024 * 1024)  
	opts.SetMaxBytesForLevelMultiplier(8)
	opts.SetNumLevels(5)
	opts.SetMaxBackgroundCompactions(8)
	opts.SetMaxBackgroundFlushes(1)
	return opts
}
func getFsmStateTableOpts() *rocksdb.Options {
	bbto := rocksdb.NewDefaultBlockBasedTableOptions()
	bbto.SetCacheIndexAndFilterBlocks(true)
	bbto.SetBlockSize(1024)
	opts := rocksdb.NewDefaultOptions()
	opts.SetBlockBasedTableFactory(bbto)
	opts.SetCompression(rocksdb.SnappyCompression)
	opts.SetWriteBufferSize(4 * 1024 * 1024)
	opts.SetMaxWriteBufferNumber(3)
	opts.SetMinWriteBufferNumberToMerge(2)
	opts.SetMaxBackgroundCompactions(1)
	opts.SetMaxBackgroundFlushes(1)
	return opts
}
func (s *RocksDBStore) Mutate(mutations []*storage.Mutation, metadata []byte) error {
	batch := rocksdb.NewWriteBatch()
	defer batch.Destroy()
	batch.PutLogData(metadata, len(metadata))
	for _, m := range mutations {
		batch.PutCF(s.cfHandles[m.Table], m.Key, m.Value)
	}
	return s.db.Write(s.wo, batch)
}
func (s *RocksDBStore) Get(table storage.Table, key []byte) (*storage.KVPair, error) {
	result := new(storage.KVPair)
	result.Key = key
	v, err := s.db.GetBytesCF(s.ro, s.cfHandles[table], key)
	if err != nil {
		return nil, err
	}
	if v == nil {
		return nil, storage.ErrKeyNotFound
	}
	result.Value = v
	return result, nil
}
func (s *RocksDBStore) GetRange(table storage.Table, start, end []byte) (storage.KVRange, error) {
	result := make(storage.KVRange, 0)
	it := s.db.NewIteratorCF(s.ro, s.cfHandles[table])
	defer it.Close()
	for it.Seek(start); it.Valid(); it.Next() {
		keySlice := it.Key()
		key := make([]byte, keySlice.Size())
		copy(key, keySlice.Data())
		keySlice.Free()
		if bytes.Compare(key, end) > 0 {
			break
		}
		valueSlice := it.Value()
		value := make([]byte, valueSlice.Size())
		copy(value, valueSlice.Data())
		result = append(result, storage.KVPair{Key: key, Value: value})
		valueSlice.Free()
	}
	return result, nil
}
func (s *RocksDBStore) GetLast(table storage.Table) (*storage.KVPair, error) {
	it := s.db.NewIteratorCF(s.ro, s.cfHandles[table])
	defer it.Close()
	it.SeekForPrev([]byte{0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff})
	if it.Valid() {
		result := new(storage.KVPair)
		keySlice := it.Key()
		key := make([]byte, keySlice.Size())
		copy(key, keySlice.Data())
		keySlice.Free()
		result.Key = key
		valueSlice := it.Value()
		value := make([]byte, valueSlice.Size())
		copy(value, valueSlice.Data())
		valueSlice.Free()
		result.Value = value
		return result, nil
	}
	return nil, storage.ErrKeyNotFound
}
func (s *RocksDBStore) GetAll(table storage.Table) storage.KVPairReader {
	return NewRocksDBKVPairReader(s.cfHandles[table], s.db)
}
func (s *RocksDBStore) Close() error {
	for _, cf := range s.cfHandles {
		cf.Destroy()
	}
	if s.db != nil {
		s.db.Close()
	}
	if s.backupOpts != nil {
		s.backupOpts.Destroy()
	}
	if s.restoreOpts != nil {
		s.restoreOpts.Destroy()
	}
	if s.blockCache != nil {
		s.blockCache.Destroy()
	}
	if s.stats != nil {
		s.stats.Destroy()
	}
	if s.globalOpts != nil {
		s.globalOpts.Destroy()
	}
	for _, opt := range s.cfOpts {
		opt.Destroy()
	}
	if s.ro != nil {
		s.ro.Destroy()
	}
	if s.wo != nil {
		s.wo.Destroy()
	}
	return nil
}