intermediate

tsdb/engine.go

@@ -53,7 +53,6 @@ type Engine interface {
 	CreateSeriesListIfNotExists(keys, names [][]byte, tags []models.Tags) error
 	DeleteSeriesRange(keys [][]byte, min, max int64) error
 
-	SeriesSketches() (estimator.Sketch, estimator.Sketch, error)
 	MeasurementsSketches() (estimator.Sketch, estimator.Sketch, error)
 	SeriesN() int64
 

tsdb/engine/tsm1/engine.go

@@ -393,10 +393,6 @@ func (e *Engine) SeriesN() int64 {
 	return e.index.SeriesN()
 }
 
-func (e *Engine) SeriesSketches() (estimator.Sketch, estimator.Sketch, error) {
-	return e.index.SeriesSketches()
-}
-
 func (e *Engine) MeasurementsSketches() (estimator.Sketch, estimator.Sketch, error) {
 	return e.index.MeasurementsSketches()
 }

tsdb/index/tsi1/file_set_test.go

@@ -141,10 +141,10 @@ func TestFileSet_MeasurementSeriesIDIterator(t *testing.T) {
 		if name, tags := fs.SeriesFile().Series(itr.Next().SeriesID); string(name) != `cpu` || tags.String() != `[{region east}]` {
 			t.Fatalf("unexpected series: %s/%s", name, tags.String())
 		}
-		if name, tags := fs.SeriesFile().Series(itr.Next().SeriesID); string(name) != `cpu` || tags.String() != `[{region north}]` {
+		if name, tags := fs.SeriesFile().Series(itr.Next().SeriesID); string(name) != `cpu` || tags.String() != `[{region west}]` {
 			t.Fatalf("unexpected series: %s/%s", name, tags.String())
 		}
-		if name, tags := fs.SeriesFile().Series(itr.Next().SeriesID); string(name) != `cpu` || tags.String() != `[{region west}]` {
+		if name, tags := fs.SeriesFile().Series(itr.Next().SeriesID); string(name) != `cpu` || tags.String() != `[{region north}]` {
 			t.Fatalf("unexpected series: %s/%s", name, tags.String())
 		}
 		if e := itr.Next(); e.SeriesID != 0 {

tsdb/index/tsi1/index.go

@@ -521,12 +521,6 @@ func (i *Index) CreateSeriesListIfNotExists(_, names [][]byte, tagsSlice []model
 	fs := i.RetainFileSet()
 	defer fs.Release()
 
-	// Filter out existing series. Exit if no new series exist.
-	names, tagsSlice = i.sfile.FilterSeriesList(names, tagsSlice)
-	if len(names) == 0 {
-		return nil
-	}
-
 	// Ensure fileset cannot change during insert.
 	i.mu.RLock()
 	// Insert series into log file.
@@ -546,10 +540,6 @@ func (i *Index) InitializeSeries(key, name []byte, tags models.Tags) error {
 
 // CreateSeriesIfNotExists creates a series if it doesn't exist or is deleted.
 func (i *Index) CreateSeriesIfNotExists(key, name []byte, tags models.Tags) error {
-	if i.sfile.HasSeries(name, tags, nil) {
-		return nil
-	}
-
 	if err := func() error {
 		i.mu.RLock()
 		defer i.mu.RUnlock()

tsdb/index/tsi1/log_file.go

@@ -648,7 +648,7 @@ func (f *LogFile) execSeriesEntry(e *LogEntry) {
 
 	// Read measurement name.
 	name, remainder := ReadSeriesKeyMeasurement(remainder)
-	mm := f.createMeasurementIfNotExists(e.Name)
+	mm := f.createMeasurementIfNotExists(name)
 
 	// Undelete measurement if it's been tombstoned previously.
 	if !deleted && mm.deleted {

tsdb/index/tsi1/log_file_test.go

@@ -100,27 +100,15 @@ func TestLogFile_SeriesStoredInOrder(t *testing.T) {
 		t.Fatal("nil iterator")
 	}
 
-	mname := []string{"cpu", "mem"}
-	var j int
+	var prevSeriesID uint32
 	for i := 0; i < len(tvs); i++ {
 		elem := itr.Next()
 		if elem.SeriesID == 0 {
 			t.Fatal("got nil series")
+		} else if elem.SeriesID < prevSeriesID {
+			t.Fatal("series out of order: %d !< %d ", elem.SeriesID, prevSeriesID)
 		}
-
-		name, tags := sfile.Series(elem.SeriesID)
-		if got, exp := string(name), mname[j]; got != exp {
-			t.Fatalf("[series %d] got %s, expected %s", i, got, exp)
-		}
-
-		if got, exp := string(tags[0].Value), tvs[i]; got != exp {
-			t.Fatalf("[series %d] got %s, expected %s", i, got, exp)
-		}
-
-		if i == (len(tvs)/2)-1 {
-			// Next measurement
-			j++
-		}
+		prevSeriesID = elem.SeriesID
 	}
 }
 

tsdb/index/tsi1/series_file.go

@@ -122,17 +122,14 @@ func (f *SeriesFile) reindex() error {
 
 	// Series data begins with an offset of 1.
 	data := f.data[1:f.offset]
+	offset := uint32(1)
 
-	var offset uint32
 	for len(data) > 0 {
-		sz, n := binary.Uvarint(data)
-		data = data[n:]
-
-		key := data[:sz]
-		data = data[sz:]
+		var key []byte
+		key, data = ReadSeriesKey(data)
 
 		m.Put(key, offset)
-		offset += uint32(sz) + uint32(n)
+		offset += uint32(len(key))
 	}
 
 	f.hashMap = m
@@ -140,11 +137,6 @@ func (f *SeriesFile) reindex() error {
 	return nil
 }
 
-// FilterSeriesList returns a list of series that don't exist.
-func (f *SeriesFile) FilterSeriesList(names [][]byte, tagsSlice []models.Tags) ([][]byte, []models.Tags) {
-	panic("TODO")
-}
-
 // CreateSeriesIfNotExists creates series if it doesn't exist. Returns the offset of the series.
 func (f *SeriesFile) CreateSeriesIfNotExists(name []byte, tags models.Tags, buf []byte) (offset uint32, err error) {
 	// Return offset if series exists.
@@ -415,9 +407,9 @@ func AppendSeriesKey(dst []byte, name []byte, tags models.Tags) []byte {
 }
 
 // ReadSeriesKey returns the series key from the beginning of the buffer.
-func ReadSeriesKey(data []byte) []byte {
+func ReadSeriesKey(data []byte) (key, remainder []byte) {
 	sz, n := binary.Uvarint(data)
-	return data[:int(sz)+n]
+	return data[:int(sz)+n], data[int(sz)+n:]
 }
 
 func ReadSeriesKeyLen(data []byte) (sz int, remainder []byte) {

tsdb/index/tsi1/tsi1_test.go

@@ -187,7 +187,7 @@ func TestMergeSeriesIDIterators(t *testing.T) {
 		t.Fatalf("unexpected elem(1): %#v", e)
 	} else if e := itr.Next(); !reflect.DeepEqual(e, tsi1.SeriesIDElem{SeriesID: 3}) {
 		t.Fatalf("unexpected elem(2): %#v", e)
-	} else if e := itr.Next(); !reflect.DeepEqual(e, tsi1.SeriesIDElem{SeriesID: 3}) {
+	} else if e := itr.Next(); !reflect.DeepEqual(e, tsi1.SeriesIDElem{SeriesID: 4}) {
 		t.Fatalf("unexpected elem(3): %#v", e)
 	} else if e := itr.Next(); e.SeriesID != 0 {
 		t.Fatalf("expected nil elem: %#v", e)

tsdb/shard.go

@@ -509,6 +509,70 @@ func (s *Shard) WritePoints(points []models.Point) error {
 	return writeError
 }
 
+// DeleteSeries deletes a list of series.
+func (s *Shard) DeleteSeries(seriesKeys [][]byte) error {
+	return s.DeleteSeriesRange(seriesKeys, math.MinInt64, math.MaxInt64)
+}
+
+// DeleteSeriesRange deletes all values from for seriesKeys between min and max (inclusive)
+func (s *Shard) DeleteSeriesRange(seriesKeys [][]byte, min, max int64) error {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	if err := s.ready(); err != nil {
+		return err
+	}
+
+	if err := s.engine.DeleteSeriesRange(seriesKeys, min, max); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// DeleteMeasurement deletes a measurement and all underlying series.
+func (s *Shard) DeleteMeasurement(name []byte) error {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	if err := s.ready(); err != nil {
+		return err
+	}
+
+	return s.engine.DeleteMeasurement(name)
+}
+
+// SeriesN returns the unique number of series in the shard.
+func (s *Shard) SeriesN() int64 {
+	return s.engine.SeriesN()
+}
+
+// MeasurementsSketches returns the measurement sketches for the shard.
+func (s *Shard) MeasurementsSketches() (estimator.Sketch, estimator.Sketch, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	if s.engine == nil {
+		return nil, nil, nil
+	}
+	return s.engine.MeasurementsSketches()
+}
+
+func (s *Shard) createFieldsAndMeasurements(fieldsToCreate []*FieldCreate) error {
+	if len(fieldsToCreate) == 0 {
+		return nil
+	}
+
+	// add fields
+	for _, f := range fieldsToCreate {
+		mf := s.engine.MeasurementFields(f.Measurement)
+		if err := mf.CreateFieldIfNotExists([]byte(f.Field.Name), f.Field.Type, false); err != nil {
+			return err
+		}
+
+		s.index.SetFieldName(f.Measurement, f.Field.Name)
+	}
+
+	return nil
+}
+
 // validateSeriesAndFields checks which series and fields are new and whose metadata should be saved and indexed.
 func (s *Shard) validateSeriesAndFields(points []models.Point) ([]models.Point, []*FieldCreate, error) {
 	var (

tsdb/store.go

@@ -755,12 +755,18 @@ func (s *Store) estimateCardinality(dbName string, getSketches func(*Shard) (est
 
 // SeriesCardinality returns the series cardinality for the provided database.
 func (s *Store) SeriesCardinality(database string) (int64, error) {
-	return s.estimateCardinality(database, func(sh *Shard) (estimator.Sketch, estimator.Sketch, error) {
-		if sh == nil {
-			return nil, nil, errors.New("shard nil, can't get cardinality")
+	s.mu.RLock()
+	shards := s.filterShards(byDatabase(database))
+	s.mu.RUnlock()
+
+	// TODO(benbjohnson): Series file will be shared by the DB.
+	var max int64
+	for _, shard := range shards {
+		if n := shard.Index().SeriesN(); n > max {
+			max = n
 		}
-		return sh.SeriesSketches()
-	})
+	}
+	return max, nil
 }
 
 // MeasurementsCardinality returns the measurement cardinality for the provided

tsdb/store_test.go

@@ -657,6 +657,8 @@ func testStoreCardinalityUnique(t *testing.T, store *Store) {
 }
 
 func TestStore_Cardinality_Unique(t *testing.T) {
+	t.Skip("TODO(benbjohnson): Merge series file to DB level")
+
 	t.Parallel()
 
 	if testing.Short() || os.Getenv("GORACE") != "" || os.Getenv("APPVEYOR") != "" {