<commit_before>package tsdb

import (
	"context"
	"encoding/binary"
	"errors"
	"fmt"
	"io/ioutil"
	"os"
	"path/filepath"
	"sync"
	"time"

	"github.com/influxdata/influxdb/kit/tracing"
	"github.com/influxdata/influxdb/logger"
	"github.com/influxdata/influxdb/models"
	"github.com/influxdata/influxdb/pkg/rhh"
	"github.com/prometheus/client_golang/prometheus"
	"go.uber.org/zap"
)

var (
	ErrSeriesPartitionClosed              = errors.New("tsdb: series partition closed")
	ErrSeriesPartitionCompactionCancelled = errors.New("tsdb: series partition compaction cancelled")
)

// DefaultSeriesPartitionCompactThreshold is the number of series IDs to hold in the in-memory
// series map before compacting and rebuilding the on-disk representation.
const DefaultSeriesPartitionCompactThreshold = 1 << 17 // 128K

// SeriesPartition represents a subset of series file data.
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 // series id sequence

	compacting          bool
	compactionsDisabled int

	CompactThreshold    int
	LargeWriteThreshold int

	tracker *seriesPartitionTracker
	Logger  *zap.Logger
}

// NewSeriesPartition returns a new instance of SeriesPartition.
func NewSeriesPartition(id int, path string) *SeriesPartition {
	p := &SeriesPartition{
		id:                  id,
		path:                path,
		closing:             make(chan struct{}),
		CompactThreshold:    DefaultSeriesPartitionCompactThreshold,
		LargeWriteThreshold: DefaultLargeSeriesWriteThreshold,
		tracker:             newSeriesPartitionTracker(newSeriesFileMetrics(nil), nil),
		Logger:              zap.NewNop(),
		seq:                 uint64(id) + 1,
	}
	p.index = NewSeriesIndex(p.IndexPath())
	return p
}

// Open memory maps the data file at the partition's path.
func (p *SeriesPartition) Open() error {
	if p.closed {
		return errors.New("tsdb: cannot reopen series partition")
	}

	// Create path if it doesn't exist.
	if err := os.MkdirAll(filepath.Join(p.path), 0777); err != nil {
		return err
	}

	// Open components.
	if err := func() (err error) {
		if err := p.openSegments(); err != nil {
			return err
		}
		// Init last segment for writes.
		if err := p.activeSegment().InitForWrite(); err != nil {
			return err
		}

		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
	}

	p.tracker.SetSeries(p.index.Count()) // Set series count metric.
	p.tracker.SetDiskSize(p.DiskSize())  // Set on-disk size metric.
	return nil
}

func (p *SeriesPartition) openSegments() error {
	fis, err := ioutil.ReadDir(p.path)
	if err != nil {
		return err
	}

	for _, fi := range fis {<commit_msg>fix(tsdb): add missing err in SeriesPartition.Open<commit_after>package tsdb

import (
	"context"
	"encoding/binary"
	"errors"
	"fmt"
	"io/ioutil"
	"os"
	"path/filepath"
	"sync"
	"time"

	"github.com/influxdata/influxdb/kit/tracing"
	"github.com/influxdata/influxdb/logger"
	"github.com/influxdata/influxdb/models"
	"github.com/influxdata/influxdb/pkg/rhh"
	"github.com/prometheus/client_golang/prometheus"
	"go.uber.org/zap"
)

var (
	ErrSeriesPartitionClosed              = errors.New("tsdb: series partition closed")
	ErrSeriesPartitionCompactionCancelled = errors.New("tsdb: series partition compaction cancelled")
)

// DefaultSeriesPartitionCompactThreshold is the number of series IDs to hold in the in-memory
// series map before compacting and rebuilding the on-disk representation.
const DefaultSeriesPartitionCompactThreshold = 1 << 17 // 128K

// SeriesPartition represents a subset of series file data.
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 // series id sequence

	compacting          bool
	compactionsDisabled int

	CompactThreshold    int
	LargeWriteThreshold int

	tracker *seriesPartitionTracker
	Logger  *zap.Logger
}

// NewSeriesPartition returns a new instance of SeriesPartition.
func NewSeriesPartition(id int, path string) *SeriesPartition {
	p := &SeriesPartition{
		id:                  id,
		path:                path,
		closing:             make(chan struct{}),
		CompactThreshold:    DefaultSeriesPartitionCompactThreshold,
		LargeWriteThreshold: DefaultLargeSeriesWriteThreshold,
		tracker:             newSeriesPartitionTracker(newSeriesFileMetrics(nil), nil),
		Logger:              zap.NewNop(),
		seq:                 uint64(id) + 1,
	}
	p.index = NewSeriesIndex(p.IndexPath())
	return p
}

// Open memory maps the data file at the partition's path.
func (p *SeriesPartition) Open() error {
	if p.closed {
		return errors.New("tsdb: cannot reopen series partition")
	}

	// Create path if it doesn't exist.
	if err := os.MkdirAll(filepath.Join(p.path), 0777); err != nil {
		return err
	}

	// Open components.
	if err := func() (err error) {
		if err := p.openSegments(); err != nil {
			return err
		}
		// Init last segment for writes.
		if err := p.activeSegment().InitForWrite(); err != nil {
			return err
		}

		if err := p.index.Open(); err != nil {
			return err
		} else if err = p.index.Recover(p.segments); err != nil {
			return err
		}
		return nil
	}(); err != nil {
		p.Close()
		return err
	}

	p.tracker.SetSeries(p.index.Count()) // Set series count metric.
	p.tracker.SetDiskSize(p.DiskSize())  // Set on-disk size metric.
	return nil
}

func (p *SeriesPartition) openSegments() error {
	fis, err := ioutil.ReadDir(p.path)
	if err != nil {
		return err
	}

	for _, fi := range fis {
{'fraction_non_alphanumeric': 0.1086543444463638, 'fraction_numerical': 0.005182242183451373, 'mean_word_length': 3.266764922623434, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4406986', 'n_tokens_mistral': 1987, 'n_tokens_neox': 1852, 'n_words': 516}