Decompress automatically if trying to insert to a compressed chunk

This may happen if prometheus is replaying old data.

Author: cevian

pkg/pgmodel/end_to_end_tests/create_test.go

@@ -9,6 +9,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/jackc/pgconn"
+	pgx "github.com/jackc/pgx/v4"
 	"github.com/jackc/pgx/v4/pgxpool"
 	"github.com/timescale/timescale-prometheus/pkg/prompb"
 
@@ -448,3 +450,57 @@ func TestSQLIngest(t *testing.T) {
 		})
 	}
 }
+
+func TestInsertCompressed(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: 1, Value: 0.1},
+				},
+			},
+		}
+		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() == "42710" {
+				//already compressed (could happen if policy already ran). This is fine
+			} else {
+				t.Fatal(err)
+			}
+		}
+		//ingest after compression
+		_, err = ingestor.Ingest(ts, NewWriteRequest())
+		if err != nil {
+			t.Fatal(err)
+		}
+		var nextStartAfter time.Time
+		err = db.QueryRow(context.Background(), "SELECT next_start FROM timescaledb_information.policy_stats WHERE hypertable = $1::text::regclass", pgx.Identifier{"prom_data", "test"}.Sanitize()).Scan(&nextStartAfter)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if time.Until(nextStartAfter) < time.Hour*10 {
+			t.Error("next_start was not changed enough")
+		}
+	})
+}

pkg/pgmodel/migrations/migration_files_generated.go

@@ -1471,3 +1471,41 @@ CREATE VIEW SCHEMA_INFO.label AS
     ARRAY(SELECT value FROM SCHEMA_CATALOG.label l WHERE l.key = lk.key ORDER BY value)
       AS values
   FROM SCHEMA_CATALOG.label_key lk;
+
+
+
+
+--Decompression should take place in a procedure because we don't want locks held across
+--decompress_chunk calls since that function takes some heavier locks at the end.
+CREATE PROCEDURE SCHEMA_CATALOG.decompress_chunks_after(metric_table NAME, min_time TIMESTAMPTZ)
+AS $$
+DECLARE
+    chunk_row _timescaledb_catalog.chunk;
+    dimension_row _timescaledb_catalog.dimension;
+    hypertable_row _timescaledb_catalog.hypertable;
+    min_time_internal bigint;
+BEGIN
+    SELECT h.* INTO STRICT hypertable_row FROM _timescaledb_catalog.hypertable h
+    WHERE table_name = metric_table AND schema_name = 'SCHEMA_DATA';
+
+    SELECT d.* INTO STRICT dimension_row FROM _timescaledb_catalog.dimension d WHERE hypertable_id = hypertable_row.id ORDER BY id LIMIT 1;
+
+    SELECT _timescaledb_internal.time_to_internal(min_time) INTO STRICT min_time_internal;
+
+    FOR chunk_row IN
+        SELECT c.*
+        FROM _timescaledb_catalog.dimension_slice ds
+        INNER JOIN _timescaledb_catalog.chunk_constraint cc ON cc.dimension_slice_id = ds.id
+        INNER JOIN _timescaledb_catalog.chunk c ON cc.chunk_id = c.id
+        WHERE dimension_id = dimension_row.id
+        -- the range_starts are inclusive
+        AND min_time_internal >= ds.range_start
+        AND c.compressed_chunk_id IS NOT NULL
+        ORDER BY ds.range_start
+    LOOP
+        RAISE NOTICE 'Timescale-Prometheus is decompressing chunk: %.%', chunk_row.schema_name, chunk_row.table_name;
+        PERFORM decompress_chunk(format('%I.%I', chunk_row.schema_name, chunk_row.table_name)::regclass);
+        COMMIT;
+    END LOOP;
+END;
+$$ LANGUAGE PLPGSQL;

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

@@ -7,7 +7,9 @@ package pgmodel
 import (
 	"context"
 	"fmt"
+	"math"
 	"sort"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -116,18 +118,28 @@ type SampleInfoIterator struct {
 	sampleInfos     []samplesInfo
 	sampleInfoIndex int
 	sampleIndex     int
+	minSeen         int64
 }
 
 // NewSampleInfoIterator is the constructor
 func NewSampleInfoIterator() SampleInfoIterator {
-	return SampleInfoIterator{sampleInfos: make([]samplesInfo, 0), sampleIndex: -1, sampleInfoIndex: 0}
+	si := SampleInfoIterator{sampleInfos: make([]samplesInfo, 0)}
+	si.ResetPosition()
+	return si
 }
 
 //Append adds a sample info to the back of the iterator
 func (t *SampleInfoIterator) Append(s samplesInfo) {
 	t.sampleInfos = append(t.sampleInfos, s)
 }
 
+//ResetPosition resets the iteration position to the beginning
+func (t *SampleInfoIterator) ResetPosition() {
+	t.sampleIndex = -1
+	t.sampleInfoIndex = 0
+	t.minSeen = math.MaxInt64
+}
+
 // Next returns true if there is another row and makes the next row data
 // available to Values(). When there are no more rows available or an error
 // has occurred it returns false.
@@ -149,6 +161,9 @@ func (t *SampleInfoIterator) Values() ([]interface{}, error) {
 		sample.Value,
 		info.seriesID,
 	}
+	if t.minSeen > sample.Timestamp {
+		t.minSeen = sample.Timestamp
+	}
 	return row, nil
 }
 
@@ -390,6 +405,7 @@ type insertHandler struct {
 	pending         *pendingBuffer
 	seriesCache     map[string]SeriesID
 	metricTableName string
+	metricName      string
 }
 
 type pendingBuffer struct {
@@ -480,6 +496,7 @@ func runInserterRoutine(conn pgxConn, input chan insertDataRequest, metricName s
 		pending:         &pendingBuffer{make([]insertDataTask, 0), NewSampleInfoIterator()},
 		seriesCache:     make(map[string]SeriesID),
 		metricTableName: tableName,
+		metricName:      metricName,
 	}
 
 	for {
@@ -560,6 +577,40 @@ func (h *insertHandler) flush() {
 	h.flushPending(h.pending)
 }
 
+func (h *insertHandler) decompressChunks(pending *pendingBuffer) error {
+	log.Warn("msg", fmt.Sprintf("Table %s was compressed, decompressing", h.metricTableName), "table", h.metricTableName, "metric", h.metricName)
+	minTime := model.Time(pending.batch.minSeen).Time()
+
+	//how much faster are we at ingestion than wall-clock time?
+	ingestSpeedup := 2
+	//delay the next compression job proportional to the duration between now and the data time + a constant safety
+	delayBy := (time.Since(minTime) / time.Duration(ingestSpeedup)) + time.Duration(60*time.Minute)
+	maxDelayBy := time.Hour * 24
+	if delayBy > maxDelayBy {
+		delayBy = maxDelayBy
+	}
+
+	_, rescheduleErr := h.conn.Exec(context.Background(),
+		`SELECT alter_job_schedule(
+							(SELECT job_id
+							FROM _timescaledb_config.bgw_policy_compress_chunks p
+							INNER JOIN _timescaledb_catalog.hypertable h ON (h.id = p.hypertable_id)
+							WHERE h.schema_name = $1 and h.table_name = $2),
+							next_start=>$3)`, dataSchema, h.metricTableName, time.Now().Add(delayBy))
+	if rescheduleErr != nil {
+		log.Error("msg", rescheduleErr, "context", "Rescheduling compression")
+		return rescheduleErr
+	}
+
+	_, decompressErr := h.conn.Exec(context.Background(), "CALL "+catalogSchema+".decompress_chunks_after($1, $2);", h.metricTableName, minTime)
+	if decompressErr != nil {
+		log.Error("msg", decompressErr, "context", "Decompressing chunks")
+		return decompressErr
+	}
+
+	return nil
+}
+
 func (h *insertHandler) flushPending(pending *pendingBuffer) {
 	err := func() error {
 		_, err := h.setSeriesIds(pending.batch.sampleInfos)
@@ -573,6 +624,23 @@ func (h *insertHandler) flushPending(pending *pendingBuffer) {
 			copyColumns,
 			&pending.batch,
 		)
+		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 := h.decompressChunks(pending)
+				if decompressErr != nil {
+					return fmt.Errorf("Error while decompressing. Decompression error: %v, Original Error: %w", decompressErr, err)
+				}
+
+				pending.batch.ResetPosition()
+				_, err = h.conn.CopyFrom(
+					context.Background(),
+					pgx.Identifier{dataSchema, h.metricTableName},
+					copyColumns,
+					&pending.batch,
+				)
+			}
+		}
 		return err
 	}()
 
@@ -592,7 +660,8 @@ func (h *insertHandler) flushPending(pending *pendingBuffer) {
 		// nil all pointers to prevent memory leaks
 		pending.batch.sampleInfos[i] = samplesInfo{}
 	}
-	pending.batch = SampleInfoIterator{sampleInfos: pending.batch.sampleInfos[:0], sampleIndex: -1, sampleInfoIndex: 0}
+	pending.batch = SampleInfoIterator{sampleInfos: pending.batch.sampleInfos[:0]}
+	pending.batch.ResetPosition()
 }
 
 func (h *insertHandler) setSeriesIds(sampleInfos []samplesInfo) (string, error) {