Cache label IDs

Add inverted cache ( (metric + label pair) -> (id, pos)) to avoid DB calls for
fetching label IDs in cases when series ID is not cached.
This cache is only used when ingesting data.
Benchmarks are showing around 5-10% gains in ingest performance and
about 25% less DB calls for fetching label IDs (note that these numbers
depend a lot on a shape of the dataset).

Author: niksajakovljevic

pkg/pgclient/client.go

@@ -137,9 +137,10 @@ func NewClientWithPool(cfg *Config, numCopiers int, connPool *pgxpool.Pool, mt t
 	labelsCache := cache.NewLabelsCache(cfg.CacheConfig)
 	seriesCache := cache.NewSeriesCache(cfg.CacheConfig, sigClose)
 	c := ingestor.Cfg{
-		NumCopiers:             numCopiers,
-		IgnoreCompressedChunks: cfg.IgnoreCompressedChunks,
-		AsyncAcks:              cfg.AsyncAcks,
+		NumCopiers:              numCopiers,
+		IgnoreCompressedChunks:  cfg.IgnoreCompressedChunks,
+		AsyncAcks:               cfg.AsyncAcks,
+		InvertedLabelsCacheSize: cfg.CacheConfig.InvertedLabelsCacheSize,
 	}
 
 	labelsReader := lreader.NewLabelsReader(dbConn, labelsCache, mt.ReadAuthorizer())

pkg/pgmodel/cache/cache.go

@@ -14,6 +14,7 @@ import (
 
 const (
 	DefaultMetricCacheSize = 10000
+	DefaultLabelsCacheSize = 100000
 )
 
 type LabelsCache interface {

pkg/pgmodel/cache/flags.go

@@ -30,15 +30,17 @@ type Config struct {
 	MetricsCacheSize        uint64
 	LabelsCacheSize         uint64
 	ExemplarKeyPosCacheSize uint64
+	InvertedLabelsCacheSize uint64
 }
 
 var DefaultConfig = Config{
 	SeriesCacheInitialSize:    DefaultSeriesCacheSize,
 	SeriesCacheMemoryMaxBytes: 1000000,
 
 	MetricsCacheSize:        DefaultMetricCacheSize,
-	LabelsCacheSize:         1000,
+	LabelsCacheSize:         DefaultLabelsCacheSize,
 	ExemplarKeyPosCacheSize: DefaultExemplarKeyPosCacheSize,
+	InvertedLabelsCacheSize: DefaultInvertedLabelsCacheSize,
 }
 
 func ParseFlags(fs *flag.FlagSet, cfg *Config) *Config {
@@ -47,11 +49,12 @@ func ParseFlags(fs *flag.FlagSet, cfg *Config) *Config {
 
 	fs.Uint64Var(&cfg.MetricsCacheSize, "metrics.cache.metrics.size", DefaultMetricCacheSize, "Maximum number of metric names to cache.")
 	fs.Uint64Var(&cfg.SeriesCacheInitialSize, "metrics.cache.series.initial-size", DefaultSeriesCacheSize, "Maximum number of series to cache.")
-	fs.Uint64Var(&cfg.LabelsCacheSize, "metrics.cache.labels.size", 10000, "Maximum number of labels to cache.")
+	fs.Uint64Var(&cfg.LabelsCacheSize, "metrics.cache.labels.size", DefaultLabelsCacheSize, "Maximum number of labels to cache.")
 	fs.Uint64Var(&cfg.ExemplarKeyPosCacheSize, "metrics.cache.exemplar.size", DefaultExemplarKeyPosCacheSize, "Maximum number of exemplar metrics key-position to cache. "+
 		"It has one-to-one mapping with number of metrics that have exemplar, as key positions are saved per metric basis.")
 	fs.Var(&cfg.seriesCacheMemoryMaxFlag, "metrics.cache.series.max-bytes", "Initial number of elements in the series cache. "+
 		"Specified in bytes or as a percentage of the memory-target (e.g. 50%).")
+	fs.Uint64Var(&cfg.InvertedLabelsCacheSize, "metrics.cache.inverted-labels.size", DefaultInvertedLabelsCacheSize, "Maximum number of label-ids to cache. This helps increase ingest performance.")
 	return cfg
 }
 

pkg/pgmodel/cache/inverted_labels_cache.go

@@ -0,0 +1,64 @@
+package cache
+
+import (
+	"fmt"
+
+	"github.com/timescale/promscale/pkg/clockcache"
+)
+
+const DefaultInvertedLabelsCacheSize = 500000
+
+type LabelInfo struct {
+	LabelID int32 // id of label
+	Pos     int32 // position of specific label within a specific metric.
+}
+
+type LabelKey struct {
+	MetricName, Name, Value string
+}
+
+func NewLabelKey(metricName, name, value string) LabelKey {
+	return LabelKey{MetricName: metricName, Name: name, Value: value}
+}
+
+func (lk LabelKey) len() int {
+	return len(lk.MetricName) + len(lk.Name) + len(lk.Value)
+}
+
+func NewLabelInfo(lableID, pos int32) LabelInfo {
+	return LabelInfo{LabelID: lableID, Pos: pos}
+}
+
+func (li LabelInfo) len() int {
+	return 8
+}
+
+// (metric, label key-pair) -> (label id,label position) cache
+// Used when creating series to avoid DB calls for labels
+// Each label position is unique for a specific metric, meaning that
+// one label can have different position for different metrics
+type InvertedLabelsCache struct {
+	cache *clockcache.Cache
+}
+
+// Cache is thread-safe
+func NewInvertedLablesCache(size uint64) (*InvertedLabelsCache, error) {
+	if size <= 0 {
+		return nil, fmt.Errorf("labels cache size must be > 0")
+	}
+	cache := clockcache.WithMetrics("inverted_labels", "metric", size)
+	return &InvertedLabelsCache{cache}, nil
+}
+
+func (c *InvertedLabelsCache) GetLabelsId(key LabelKey) (LabelInfo, bool) {
+	id, found := c.cache.Get(key)
+	if found {
+		return id.(LabelInfo), found
+	}
+	return LabelInfo{}, false
+}
+
+func (c *InvertedLabelsCache) Put(key LabelKey, val LabelInfo) bool {
+	_, added := c.cache.Insert(key, val, uint64(key.len())+uint64(val.len())+17)
+	return added
+}

pkg/pgmodel/ingestor/dispatcher.go

@@ -51,7 +51,7 @@ type pgxDispatcher struct {
 
 var _ model.Dispatcher = &pgxDispatcher{}
 
-func newPgxDispatcher(conn pgxconn.PgxConn, cache cache.MetricCache, scache cache.SeriesCache, eCache cache.PositionCache, cfg *Cfg) (*pgxDispatcher, error) {
+func newPgxDispatcher(conn pgxconn.PgxConn, mCache cache.MetricCache, scache cache.SeriesCache, eCache cache.PositionCache, cfg *Cfg) (*pgxDispatcher, error) {
 	numCopiers := cfg.NumCopiers
 	if numCopiers < 1 {
 		log.Warn("msg", "num copiers less than 1, setting to 1")
@@ -76,7 +76,11 @@ func newPgxDispatcher(conn pgxconn.PgxConn, cache cache.MetricCache, scache cach
 	}
 
 	labelArrayOID := model.GetCustomTypeOID(model.LabelArray)
-	sw := NewSeriesWriter(conn, labelArrayOID)
+	labelsCache, err := cache.NewInvertedLablesCache(cfg.InvertedLabelsCacheSize)
+	if err != nil {
+		return nil, err
+	}
+	sw := NewSeriesWriter(conn, labelArrayOID, labelsCache)
 	elf := NewExamplarLabelFormatter(conn, eCache)
 
 	for i := 0; i < numCopiers; i++ {
@@ -85,7 +89,7 @@ func newPgxDispatcher(conn pgxconn.PgxConn, cache cache.MetricCache, scache cach
 
 	inserter := &pgxDispatcher{
 		conn:                   conn,
-		metricTableNames:       cache,
+		metricTableNames:       mCache,
 		scache:                 scache,
 		exemplarKeyPosCache:    eCache,
 		completeMetricCreation: make(chan struct{}, 1),

pkg/pgmodel/ingestor/handler_test.go

@@ -19,8 +19,8 @@ func getSeries(t *testing.T, scache *cache.SeriesCacheImpl, labels labels.Labels
 	return series
 }
 
-func makeLabelKey(l labels.Label) labelKey {
-	return labelKey{MetricName: "metric", Name: l.Name, Value: l.Value}
+func makeLabelKey(l labels.Label) cache.LabelKey {
+	return cache.LabelKey{MetricName: "metric", Name: l.Name, Value: l.Value}
 }
 
 func TestLabelArrayCreator(t *testing.T) {
@@ -31,9 +31,9 @@ func TestLabelArrayCreator(t *testing.T) {
 	seriesSet := []*model.Series{
 		getSeries(t, scache, labels.Labels{metricNameLabel, valOne}),
 	}
-	labelMap := map[labelKey]labelInfo{
-		makeLabelKey(metricNameLabel): {2, 1},
-		makeLabelKey(valOne):          {3, 2},
+	labelMap := map[cache.LabelKey]cache.LabelInfo{
+		makeLabelKey(metricNameLabel): cache.NewLabelInfo(2, 1),
+		makeLabelKey(valOne):          cache.NewLabelInfo(3, 2),
 	}
 
 	res, _, err := createLabelArrays(seriesSet, labelMap, 2)
@@ -46,7 +46,7 @@ func TestLabelArrayCreator(t *testing.T) {
 	expected = [][]int32{{2, 3}}
 	require.Equal(t, res, expected)
 
-	labelMap[makeLabelKey(valOne)] = labelInfo{3, 3}
+	labelMap[makeLabelKey(valOne)] = cache.NewLabelInfo(3, 3)
 	res, _, err = createLabelArrays(seriesSet, labelMap, 3)
 	require.NoError(t, err)
 	expected = [][]int32{{2, 0, 3}}
@@ -57,10 +57,10 @@ func TestLabelArrayCreator(t *testing.T) {
 		getSeries(t, scache, labels.Labels{metricNameLabel, valOne}),
 		getSeries(t, scache, labels.Labels{metricNameLabel, valTwo}),
 	}
-	labelMap = map[labelKey]labelInfo{
-		makeLabelKey(metricNameLabel): {100, 1},
-		makeLabelKey(valOne):          {1, 5},
-		makeLabelKey(valTwo):          {2, 5},
+	labelMap = map[cache.LabelKey]cache.LabelInfo{
+		makeLabelKey(metricNameLabel): cache.NewLabelInfo(100, 1),
+		makeLabelKey(valOne):          cache.NewLabelInfo(1, 5),
+		makeLabelKey(valTwo):          cache.NewLabelInfo(2, 5),
 	}
 
 	res, ser, err := createLabelArrays(seriesSet, labelMap, 5)
@@ -79,10 +79,10 @@ func TestLabelArrayCreator(t *testing.T) {
 		getSeries(t, scache, labels.Labels{metricNameLabel, valOne}),
 		setSeries,
 	}
-	labelMap = map[labelKey]labelInfo{
-		makeLabelKey(metricNameLabel): {100, 1},
-		makeLabelKey(valOne):          {1, 5},
-		makeLabelKey(valTwo):          {2, 5},
+	labelMap = map[cache.LabelKey]cache.LabelInfo{
+		makeLabelKey(metricNameLabel): cache.NewLabelInfo(100, 1),
+		makeLabelKey(valOne):          cache.NewLabelInfo(1, 5),
+		makeLabelKey(valTwo):          cache.NewLabelInfo(2, 5),
 	}
 	res, ser, err = createLabelArrays(seriesSet, labelMap, 5)
 	require.NoError(t, err)

pkg/pgmodel/ingestor/ingestor.go

@@ -24,10 +24,11 @@ import (
 )
 
 type Cfg struct {
-	AsyncAcks              bool
-	NumCopiers             int
-	DisableEpochSync       bool
-	IgnoreCompressedChunks bool
+	AsyncAcks               bool
+	NumCopiers              int
+	DisableEpochSync        bool
+	IgnoreCompressedChunks  bool
+	InvertedLabelsCacheSize uint64
 }
 
 // DBIngestor ingest the TimeSeries data into Timescale database.
@@ -57,7 +58,7 @@ func NewPgxIngestor(conn pgxconn.PgxConn, cache cache.MetricCache, sCache cache.
 // with an empty config, a new default size metrics cache and a non-ha-aware data parser
 func NewPgxIngestorForTests(conn pgxconn.PgxConn, cfg *Cfg) (*DBIngestor, error) {
 	if cfg == nil {
-		cfg = &Cfg{}
+		cfg = &Cfg{InvertedLabelsCacheSize: cache.DefaultConfig.InvertedLabelsCacheSize}
 	}
 	cacheConfig := cache.DefaultConfig
 	c := cache.NewMetricCache(cacheConfig)

pkg/pgmodel/ingestor/ingestor_sql_test.go

@@ -274,8 +274,8 @@ func TestPGXInserterInsertSeries(t *testing.T) {
 			mock := model.NewSqlRecorder(c.sqlQueries, t)
 			scache := cache.NewSeriesCache(cache.DefaultConfig, nil)
 			scache.Reset()
-
-			sw := NewSeriesWriter(mock, 0)
+			lCache, _ := cache.NewInvertedLablesCache(10)
+			sw := NewSeriesWriter(mock, 0, lCache)
 
 			lsi := make([]model.Insertable, 0)
 			for _, ser := range c.series {
@@ -391,21 +391,6 @@ func TestPGXInserterCacheReset(t *testing.T) {
 		},
 		{Sql: "COMMIT;"},
 		{Sql: "BEGIN;"},
-		{
-			Sql: "SELECT * FROM _prom_catalog.get_or_create_label_ids($1, $2, $3, $4)",
-			Args: []interface{}{
-				"metric_1",
-				tableName,
-				[]string{"__name__", "name_1", "name_1"},
-				[]string{"metric_1", "value_1", "value_2"},
-			},
-			Results: model.RowResults{
-				{[]int32{1, 2, 2}, []int32{1, 2, 3}, []string{"__name__", "name_1", "name_1"}, []string{"metric_1", "value_1", "value_2"}},
-			},
-			Err: error(nil),
-		},
-		{Sql: "COMMIT;"},
-		{Sql: "BEGIN;"},
 		{
 			Sql: seriesInsertSQL,
 			Args: []interface{}{
@@ -432,8 +417,8 @@ func TestPGXInserterCacheReset(t *testing.T) {
 
 	mock := model.NewSqlRecorder(sqlQueries, t)
 	scache := cache.NewSeriesCache(cache.DefaultConfig, nil)
-
-	sw := NewSeriesWriter(mock, 0)
+	lcache, _ := cache.NewInvertedLablesCache(10)
+	sw := NewSeriesWriter(mock, 0, lcache)
 	inserter := pgxDispatcher{
 		conn:   mock,
 		scache: scache,
@@ -848,7 +833,7 @@ func TestPGXInserterInsertData(t *testing.T) {
 			if err != nil {
 				t.Fatalf("error setting up mock cache: %s", err.Error())
 			}
-			inserter, err := newPgxDispatcher(mock, mockMetrics, scache, nil, &Cfg{DisableEpochSync: true})
+			inserter, err := newPgxDispatcher(mock, mockMetrics, scache, nil, &Cfg{DisableEpochSync: true, InvertedLabelsCacheSize: 10})
 			if err != nil {
 				t.Fatal(err)
 			}