func (p *SeriesPartition) AppendSeriesIDs(a []uint64) []uint64 {
	for _, segment := range p.segments {
		a = segment.AppendSeriesIDs(a)
	}
	return a
}
AppendSeriesIDs returns a list of all series ids.
package tsdb
import (
	"encoding/binary"
	"errors"
	"fmt"
	"io/ioutil"
	"os"
	"path/filepath"
	"sync"
	"github.com/influxdata/influxdb/logger"
	"github.com/influxdata/influxdb/models"
	"github.com/influxdata/influxdb/pkg/rhh"
	"go.uber.org/zap"
)
var (
	ErrSeriesPartitionClosed              = errors.New("tsdb: series partition closed")
	ErrSeriesPartitionCompactionCancelled = errors.New("tsdb: series partition compaction cancelled")
)
const DefaultSeriesPartitionCompactThreshold = 1 << 17  
type SeriesPartition struct {
	mu   sync.RWMutex
	wg   sync.WaitGroup
	id   int
	path string
	closed  bool
	closing chan struct{}
	once    sync.Once
	segments []*SeriesSegment
	index    *SeriesIndex
	seq      uint64  
	compacting          bool
	compactionsDisabled int
	CompactThreshold int
	Logger *zap.Logger
}
func NewSeriesPartition(id int, path string) *SeriesPartition {
	return &SeriesPartition{
		id:               id,
		path:             path,
		closing:          make(chan struct{}),
		CompactThreshold: DefaultSeriesPartitionCompactThreshold,
		Logger:           zap.NewNop(),
		seq:              uint64(id) + 1,
	}
}
func (p *SeriesPartition) Open() error {
	if p.closed {
		return errors.New("tsdb: cannot reopen series partition")
	}
	if err := os.MkdirAll(filepath.Join(p.path), 0777); err != nil {
		return err
	}
	if err := func() (err error) {
		if err := p.openSegments(); err != nil {
			return err
		}
		if err := p.activeSegment().InitForWrite(); err != nil {
			return err
		}
		p.index = NewSeriesIndex(p.IndexPath())
		if err := p.index.Open(); err != nil {
			return err
		} else if p.index.Recover(p.segments); err != nil {
			return err
		}
		return nil
	}(); err != nil {
		p.Close()
		return err
	}
	return nil
}
func (p *SeriesPartition) openSegments() error {
	fis, err := ioutil.ReadDir(p.path)
	if err != nil {
		return err
	}
	for _, fi := range fis {
		segmentID, err := ParseSeriesSegmentFilename(fi.Name())
		if err != nil {
			continue
		}
		segment := NewSeriesSegment(segmentID, filepath.Join(p.path, fi.Name()))
		if err := segment.Open(); err != nil {
			return err
		}
		p.segments = append(p.segments, segment)
	}
	for i := len(p.segments) - 1; i >= 0; i-- {
		if seq := p.segments[i].MaxSeriesID(); seq >= p.seq {
			p.seq = seq + SeriesFilePartitionN
			break
		}
	}
	if len(p.segments) == 0 {
		segment, err := CreateSeriesSegment(0, filepath.Join(p.path, "0000"))
		if err != nil {
			return err
		}
		p.segments = append(p.segments, segment)
	}
	return nil
}
func (p *SeriesPartition) Close() (err error) {
	p.once.Do(func() { close(p.closing) })
	p.wg.Wait()
	p.mu.Lock()
	defer p.mu.Unlock()
	p.closed = true
	for _, s := range p.segments {
		if e := s.Close(); e != nil && err == nil {
			err = e
		}
	}
	p.segments = nil
	if p.index != nil {
		if e := p.index.Close(); e != nil && err == nil {
			err = e
		}
	}
	p.index = nil
	return err
}
func (p *SeriesPartition) ID() int { return p.id }
func (p *SeriesPartition) Path() string { return p.path }
func (p *SeriesPartition) IndexPath() string { return filepath.Join(p.path, "index") }
func (p *SeriesPartition) CreateSeriesListIfNotExists(keys [][]byte, keyPartitionIDs []int, ids []uint64) error {
	var writeRequired bool
	p.mu.RLock()
	if p.closed {
		p.mu.RUnlock()
		return ErrSeriesPartitionClosed
	}
	for i := range keys {
		if keyPartitionIDs[i] != p.id {
			continue
		}
		id := p.index.FindIDBySeriesKey(p.segments, keys[i])
		if id == 0 {
			writeRequired = true
			continue
		}
		ids[i] = id
	}
	p.mu.RUnlock()
	if !writeRequired {
		return nil
	}
	type keyRange struct {
		id     uint64
		offset int64
	}
	newKeyRanges := make([]keyRange, 0, len(keys))
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.closed {
		return ErrSeriesPartitionClosed
	}
	newIDs := make(map[string]uint64, len(ids))
	for i := range keys {
		if keyPartitionIDs[i] != p.id || ids[i] != 0 {
			continue
		}
		key := keys[i]
		if ids[i] = newIDs[string(key)]; ids[i] != 0 {
			continue
		} else if ids[i] = p.index.FindIDBySeriesKey(p.segments, key); ids[i] != 0 {
			continue
		}
		id, offset, err := p.insert(key)
		if err != nil {
			return err
		}
		ids[i] = id
		newIDs[string(key)] = id
		newKeyRanges = append(newKeyRanges, keyRange{id, offset})
	}
	if segment := p.activeSegment(); segment != nil {
		if err := segment.Flush(); err != nil {
			return err
		}
	}
	for _, keyRange := range newKeyRanges {
		p.index.Insert(p.seriesKeyByOffset(keyRange.offset), keyRange.id, keyRange.offset)
	}
	if p.compactionsEnabled() && !p.compacting && p.CompactThreshold != 0 && p.index.InMemCount() >= uint64(p.CompactThreshold) {
		p.compacting = true
		log, logEnd := logger.NewOperation(p.Logger, "Series partition compaction", "series_partition_compaction", zap.String("path", p.path))
		p.wg.Add(1)
		go func() {
			defer p.wg.Done()
			compactor := NewSeriesPartitionCompactor()
			compactor.cancel = p.closing
			if err := compactor.Compact(p); err != nil {
				log.Error("series partition compaction failed", zap.Error(err))
			}
			logEnd()
			p.mu.Lock()
			p.compacting = false
			p.mu.Unlock()
		}()
	}
	return nil
}
func (p *SeriesPartition) DeleteSeriesID(id uint64) error {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.closed {
		return ErrSeriesPartitionClosed
	}
	if p.index.IsDeleted(id) {
		return nil
	}
	_, err := p.writeLogEntry(AppendSeriesEntry(nil, SeriesEntryTombstoneFlag, id, nil))
	if err != nil {
		return err
	}
	p.index.Delete(id)
	return nil
}
func (p *SeriesPartition) IsDeleted(id uint64) bool {
	p.mu.RLock()
	if p.closed {
		p.mu.RUnlock()
		return false
	}
	v := p.index.IsDeleted(id)
	p.mu.RUnlock()
	return v
}
func (p *SeriesPartition) SeriesKey(id uint64) []byte {
	if id == 0 {
		return nil
	}
	p.mu.RLock()
	if p.closed {
		p.mu.RUnlock()
		return nil
	}
	key := p.seriesKeyByOffset(p.index.FindOffsetByID(id))
	p.mu.RUnlock()
	return key
}
func (p *SeriesPartition) Series(id uint64) ([]byte, models.Tags) {
	key := p.SeriesKey(id)
	if key == nil {
		return nil, nil
	}
	return ParseSeriesKey(key)
}
func (p *SeriesPartition) FindIDBySeriesKey(key []byte) uint64 {
	p.mu.RLock()
	if p.closed {
		p.mu.RUnlock()
		return 0
	}
	id := p.index.FindIDBySeriesKey(p.segments, key)
	p.mu.RUnlock()
	return id
}
func (p *SeriesPartition) SeriesCount() uint64 {
	p.mu.RLock()
	if p.closed {
		p.mu.RUnlock()
		return 0
	}
	n := p.index.Count()
	p.mu.RUnlock()
	return n
}
func (p *SeriesPartition) DisableCompactions() {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.compactionsDisabled++
}
func (p *SeriesPartition) EnableCompactions() {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.compactionsEnabled() {
		return
	}
	p.compactionsDisabled++
}
func (p *SeriesPartition) compactionsEnabled() bool {
	return p.compactionsDisabled == 0
}