Ignore duplicate data points

Prometheus does not allow data points with duplicate
(timestamp, series id). To replicate this, this commit adds a UNIQUE
index to each of the timestamp data tables, and adds a fallback path so
that COPY FROMs that fail due to duplicate keys are retried as
INSERT ... ON CONFLICT DO NOTHINGs

Author: JLockerman

pkg/log/log.go

@@ -16,7 +16,7 @@ import (
 
 var (
 	// Application wide logger
-	logger log.Logger
+	logger log.Logger = log.NewNopLogger()
 
 	// logger timestamp format
 	timestampFormat = log.TimestampFormat(

pkg/pgmodel/end_to_end_tests/create_test.go

@@ -6,10 +6,12 @@ package end_to_end_tests
 import (
 	"context"
 	"fmt"
+	"reflect"
 	"testing"
 	"time"
 
 	"github.com/jackc/pgconn"
+	"github.com/jackc/pgerrcode"
 	pgx "github.com/jackc/pgx/v4"
 	"github.com/jackc/pgx/v4/pgxpool"
 	"github.com/timescale/timescale-prometheus/pkg/prompb"
@@ -316,7 +318,7 @@ func TestSQLIngest(t *testing.T) {
 					},
 				},
 			},
-			count:       2,
+			count:       1,
 			countSeries: 1,
 		},
 		{
@@ -404,8 +406,12 @@ func TestSQLIngest(t *testing.T) {
 					t.Fatalf("got an unexpected error %v", err)
 				}
 
-				if cnt != tcase.count {
-					t.Fatalf("counts not equal: got %v expected %v\n", cnt, tcase.count)
+				// our reporting of inserts is necessarily inexact due to
+				// duplicates being dropped we report the number of samples the
+				// ingestor handled, not necissarily how many were inserted
+				// into the DB
+				if cnt < tcase.count {
+					t.Fatalf("incorrect counts: got %v expected %v\n", cnt, tcase.count)
 				}
 
 				if err != nil {
@@ -430,8 +436,8 @@ func TestSQLIngest(t *testing.T) {
 					totalRows += rowsInTable
 				}
 
-				if totalRows != int(cnt) {
-					t.Fatalf("counts not equal: got %v expected %v\n", totalRows, cnt)
+				if totalRows != int(tcase.count) {
+					t.Fatalf("counts not equal: got %v expected %v\n", totalRows, tcase.count)
 				}
 
 				err = ingestor.CompleteMetricCreation()
@@ -452,6 +458,109 @@ func TestSQLIngest(t *testing.T) {
 	}
 }
 
+func TestInsertCompressedDuplicates(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	withDB(t, *testDatabase, func(db *pgxpool.Pool, t testing.TB) {
+		ts := []prompb.TimeSeries{
+			{
+				Labels: []prompb.Label{
+					{Name: MetricNameLabelName, Value: "test"},
+					{Name: "test", Value: "test"},
+				},
+				Samples: []prompb.Sample{
+					{Timestamp: 10000000, Value: 1.0},
+				},
+			},
+		}
+		ingestor, err := NewPgxIngestor(db)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer ingestor.Close()
+		_, err = ingestor.Ingest(ts, NewWriteRequest())
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = ingestor.CompleteMetricCreation()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		_, err = db.Exec(context.Background(), "SELECT compress_chunk(i) from show_chunks('prom_data.test') i;")
+
+		if err != nil {
+			if pgErr, ok := err.(*pgconn.PgError); ok && pgErr.SQLState() == pgerrcode.DuplicateObject {
+				//already compressed (could happen if policy already ran). This is fine
+			} else {
+				t.Fatal(err)
+			}
+		}
+
+		ts = []prompb.TimeSeries{
+			{
+				Labels: []prompb.Label{
+					{Name: MetricNameLabelName, Value: "test"},
+					{Name: "test", Value: "test"},
+				},
+				Samples: []prompb.Sample{
+					{Timestamp: 1, Value: 0.0},
+				},
+			},
+		}
+
+		_, err = ingestor.Ingest(ts, NewWriteRequest())
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = ingestor.CompleteMetricCreation()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		ts = []prompb.TimeSeries{
+			{
+				Labels: []prompb.Label{
+					{Name: MetricNameLabelName, Value: "test"},
+					{Name: "test", Value: "test"},
+				},
+				Samples: []prompb.Sample{
+					{Timestamp: 1, Value: 0.2},
+					{Timestamp: 10, Value: 3.0},
+					{Timestamp: 10000000, Value: 4.0},
+					{Timestamp: 10000001, Value: 5.0},
+				},
+			},
+		}
+
+		//ingest duplicate after compression
+		_, err = ingestor.Ingest(ts, NewWriteRequest())
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		rows, err := db.Query(context.Background(), "SELECT value FROM prom_data.test ORDER BY time")
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		expected := []float64{0.0, 3.0, 1.0, 5.0}
+		found := make([]float64, 0, 4)
+		for rows.Next() {
+			var value float64
+			err = rows.Scan(&value)
+			if err != nil {
+				t.Fatal(err)
+			}
+			found = append(found, value)
+		}
+		if !reflect.DeepEqual(expected, found) {
+			t.Errorf("wrong values in DB\nexpected:\n\t%v\ngot:\n\t%v", expected, found)
+		}
+	})
+}
+
 func TestInsertCompressed(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test")

pkg/pgmodel/migrations/migration_files_generated.go

@@ -266,7 +266,7 @@ DECLARE
 BEGIN
    EXECUTE format('CREATE TABLE SCHEMA_DATA.%I(time TIMESTAMPTZ NOT NULL, value DOUBLE PRECISION, series_id INT NOT NULL)',
                     NEW.table_name);
-   EXECUTE format('CREATE INDEX data_series_id_time_%s ON SCHEMA_DATA.%I (series_id, time) INCLUDE (value)',
+   EXECUTE format('CREATE UNIQUE INDEX data_series_id_time_%s ON SCHEMA_DATA.%I (series_id, time) INCLUDE (value)',
                     NEW.id, NEW.table_name);
    PERFORM create_hypertable(format('SCHEMA_DATA.%I', NEW.table_name), 'time',
                              chunk_time_interval=>SCHEMA_CATALOG.get_default_chunk_interval(),
@@ -803,10 +803,10 @@ LANGUAGE PLPGSQL VOLATILE;
 GRANT EXECUTE ON FUNCTION SCHEMA_CATALOG.create_series(int, name, SCHEMA_PROM.label_array) TO prom_writer;
 
 -- There shouldn't be a need to have a read only version of this as we'll use
--- the eq or other matcher functions to find series ids like this. However, 
+-- the eq or other matcher functions to find series ids like this. However,
 -- there are possible use cases that need the series id directly for performance
--- that we might want to see if we need to support, in which case a 
--- read only version might be useful in future. 
+-- that we might want to see if we need to support, in which case a
+-- read only version might be useful in future.
 CREATE OR REPLACE  FUNCTION SCHEMA_CATALOG.get_or_create_series_id(label jsonb)
 RETURNS BIGINT AS $$
 DECLARE
@@ -1537,4 +1537,4 @@ BEGIN
         COMMIT;
     END LOOP;
 END;
-$$ LANGUAGE PLPGSQL;
+$$ LANGUAGE PLPGSQL;

pkg/pgmodel/migrations/sql/1_base_schema.up.sql

@@ -643,37 +643,105 @@ func runCopyFrom(conn pgxConn, in chan copyRequest) {
 		if !ok {
 			return
 		}
-		_, err := conn.CopyFrom(
-			context.Background(),
-			pgx.Identifier{dataSchema, req.table},
-			copyColumns,
-			&req.data.batch,
-		)
+		err := doCopyFrom(conn, req)
 		if err != nil {
-			if pgErr, ok := err.(*pgconn.PgError); ok && strings.Contains(pgErr.Message, "insert/update/delete not permitted") {
-				/* If the error was that the table is already compressed, decompress and try again. */
-				decompressErr := decompressChunks(conn, req.data, req.table)
-				if decompressErr != nil {
-					req.data.reportResults(err)
-					pendingBuffers.Put(req.data)
-					continue
-				}
-
-				req.data.batch.ResetPosition()
-				_, err = conn.CopyFrom(
-					context.Background(),
-					pgx.Identifier{dataSchema, req.table},
-					copyColumns,
-					&req.data.batch,
-				)
-			}
+			err = copyFromErrorFallback(conn, req, err)
 		}
 
 		req.data.reportResults(err)
 		pendingBuffers.Put(req.data)
 	}
 }
 
+// certain errors are recoverable, handle those we can
+//   1. if the data contains duplicates, switch to INSERT ... ON CONFLICT DO NOTHING
+//      so the duplicates are dropped and the other data can go in
+//   2. if the table is compressed, decompress and retry the insertion
+func copyFromErrorFallback(conn pgxConn, req copyRequest, err error) error {
+	// we use either COPY FROM or INSERT depending on if the dataset has duplicates
+	// if we do need to use INSERT, we need to keep doing so, otherwise we'll
+	// fail due to duplicates.
+	insertFn := doCopyFrom
+	for i := 0; err != nil && i < 10; i++ {
+		insertFn, err = tryRecovery(conn, req, insertFn, err)
+		if insertFn == nil || err != nil {
+			return err
+		}
+		err = insertFn(conn, req)
+	}
+	if err != nil {
+		log.Warn("msg", fmt.Sprintf("time out while processing error for %s", req.table), "error", err.Error())
+	}
+	return err
+}
+
+type insertionFunction = func(conn pgxConn, req copyRequest) error
+
+// we can currently recover from two error:
+// If we inserted duplicate data we switch to using INSERT ... ON CONFLICT DO NOTHING
+// to skip redundant data points and try again.
+// If we inserted into a compressed chunk, we decompress the chunk and try again.
+// Since a single batch can have both errors, we need to remember the insert method
+// we're using, so that we deduplicate if needed.
+func tryRecovery(conn pgxConn, req copyRequest, insertFn insertionFunction, err error) (insertionFunction, error) {
+
+	// we only recover from postgres errors right now
+	pgErr, ok := err.(*pgconn.PgError)
+	if !ok {
+		errMsg := err.Error()
+		log.Warn("msg", fmt.Sprintf("unexpected error while inserting to %s", req.table), "error", errMsg)
+		return nil, err
+	}
+
+	// if we have duplicate data points try again using
+	// INSERT ... ON CONFLICT DO NOTHING to insert only the new ones
+	if pgErr.Code == pgerrcode.UniqueViolation {
+		log.Warn("msg", fmt.Sprintf("duplicate data in sample for %s", req.table), "table", req.table)
+		req.data.batch.ResetPosition()
+		return doInsert, nil
+	}
+
+	// If the error was that the table is already compressed, decompress and try again.
+	if strings.Contains(pgErr.Message, "insert/update/delete not permitted") {
+		decompressErr := decompressChunks(conn, req.data, req.table)
+		if decompressErr != nil {
+			return nil, err
+		}
+
+		req.data.batch.ResetPosition()
+		// use the old insertFn in case it was doing deduplication
+		return insertFn, nil
+	}
+
+	log.Warn("msg", fmt.Sprintf("unexpected postgres error while inserting to %s", req.table), "error", pgErr.Error())
+	return nil, err
+}
+
+func doCopyFrom(conn pgxConn, req copyRequest) error {
+	_, err := conn.CopyFrom(
+		context.Background(),
+		pgx.Identifier{dataSchema, req.table},
+		copyColumns,
+		&req.data.batch,
+	)
+	return err
+}
+
+func doInsert(conn pgxConn, req copyRequest) error {
+	queryString := fmt.Sprintf("INSERT INTO %s.%s(time, value, series_id) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING", dataSchema, req.table)
+	insertBatch := conn.NewBatch()
+	for req.data.batch.Next() {
+		values, _ := req.data.batch.Values()
+		insertBatch.Queue(queryString, values...)
+	}
+	resp, err := conn.SendBatch(context.Background(), insertBatch)
+	err2 := resp.Close()
+	if err == nil {
+		err = err2
+	}
+	return err
+}
+
 func decompressChunks(conn pgxConn, pending *pendingBuffer, table string) error {
 	minTime := model.Time(pending.batch.minSeen).Time()