Fix cache population when initializing metric batcher

Series table name was not set correctly when initializing the metric batcher
creating issues when trying to query metrics. The issue would resolve itself
once the cache was dropped and recreated.

pkg/pgmodel/ingestor/metric_batcher.go

@@ -95,7 +95,7 @@ func initializeMetricBatcher(conn pgxconn.PgxConn, metricName string, completeMe
 		metricName,
 		model.MetricInfo{
 			TableSchema: schema.Data, TableName: tableName,
-			SeriesTable: "",
+			SeriesTable: tableName, // Series table name is always the same for raw metrics.
 		},
 		false,
 	)

pkg/pgmodel/ingestor/metric_batcher_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 	"github.com/timescale/promscale/pkg/pgmodel/cache"
+	"github.com/timescale/promscale/pkg/pgmodel/common/schema"
 	"github.com/timescale/promscale/pkg/pgmodel/model"
 	pgmodel "github.com/timescale/promscale/pkg/pgmodel/model"
 	"github.com/timescale/promscale/pkg/prompb"
@@ -84,6 +85,35 @@ func TestMetricTableName(t *testing.T) {
 	}
 }
 
+func TestInitializeMetricBatcher(t *testing.T) {
+	metricName := "mock_metric"
+	metricTableName := "mock_metric_table_name"
+	sqlQueries := []model.SqlQuery{
+		{
+			Sql:     "SELECT table_name, possibly_new FROM _prom_catalog.get_or_create_metric_table_name($1)",
+			Args:    []interface{}{metricName},
+			Results: model.RowResults{{metricTableName, true}},
+		},
+	}
+	mock := model.NewSqlRecorder(sqlQueries, t)
+	mockMetrics := &model.MockMetricCache{
+		MetricCache: make(map[string]model.MetricInfo),
+	}
+	completeMetricCreation := make(chan struct{}, 1)
+
+	tableName, err := initializeMetricBatcher(mock, metricName, completeMetricCreation, mockMetrics)
+	require.Nil(t, err)
+	require.Equal(t, metricTableName, tableName)
+
+	// Double-check the cache was set properly.
+	mInfo, err := mockMetrics.Get(schema.Data, metricName, false)
+	require.Nil(t, err)
+	require.Equal(t, schema.Data, mInfo.TableSchema)
+	require.Equal(t, metricTableName, mInfo.TableName)
+	require.Equal(t, metricTableName, mInfo.SeriesTable)
+
+}
+
 type insertableVisitor []model.Insertable
 
 func (insertables insertableVisitor) VisitExemplar(callBack func(s *pgmodel.PromExemplars) error) error {

pkg/pgmodel/model/sql_test_utils.go

@@ -475,7 +475,7 @@ func (m *MockMetricCache) Get(schema, metric string, isExemplar bool) (MetricInf
 }
 
 func (m *MockMetricCache) Set(schema, metric string, mInfo MetricInfo, isExemplar bool) error {
-	m.MetricCache[schema+"*"+metric] = mInfo
+	m.MetricCache[fmt.Sprintf("%s_%s_%t", schema, metric, isExemplar)] = mInfo
 	return m.SetMetricErr
 }