Aggregates for downsampling caggs

CHANGELOG.md

@@ -32,8 +32,9 @@ We use the following categories for changes:
 - Telemetry for active series and last updated [#534]
 
 ### Fixed
-- Column conflict when creating a metric view with a label called `series`
-  [#559]
+- Column conflict when creating a metric view with a label called `series` [#559]
+- Support to create and delete metric rollups based on metric types [#538]
+- Support for maintenance of Metric-rollups and Caggs for downsampling [#554]
 
 ## [0.7.0 - 2022-10-03]
 

docs/sql-api.md

@@ -47,6 +47,11 @@ Executes a specified maintenance job type like dropping data according to retent
 ```
 procedure void **prom_api.execute_maintenance**(IN signal _ps_catalog.signal_type, IN job_type _ps_catalog.job_type, IN log_verbose boolean DEFAULT false)
 ```
+### prom_api.get_automatic_downsample
+Get automatic downsample state for metrics (a.k.a. metric rollups)
+```
+function boolean **prom_api.get_automatic_downsample**()
+```
 ### prom_api.get_default_chunk_interval
 Get the default chunk interval for all metrics
 ```
@@ -135,7 +140,7 @@ function void **prom_api.promscale_post_restore**()
 ### prom_api.register_metric_view
 
 ```
-function boolean **prom_api.register_metric_view**(schema_name text, view_name text, if_not_exists boolean DEFAULT false)
+function boolean **prom_api.register_metric_view**(schema_name text, view_name text, refresh_interval interval, if_not_exists boolean DEFAULT false, rollup_id bigint DEFAULT NULL::bigint)
 ```
 ### prom_api.reset_metric_chunk_interval
 resets the chunk interval for a specific metric to using the default
@@ -157,6 +162,11 @@ resets the retention period for a specific metric to using the default
 ```
 function boolean **prom_api.reset_metric_retention_period**(schema_name text, metric_name text)
 ```
+### prom_api.set_automatic_downsample
+Set automatic downsample state for metrics (a.k.a. metric rollups). Metric rollups will be created only if this returns true
+```
+function boolean **prom_api.set_automatic_downsample**(_state boolean)
+```
 ### prom_api.set_compression_on_metric_table
 set a compression for a specific metric table
 ```
@@ -492,6 +502,11 @@ the specified span_id.
 ```
 function TABLE(trace_id trace_id, parent_span_id bigint, span_id bigint, dist integer, path bigint[]) **ps_trace.upstream_spans**(_trace_id trace_id, _span_id bigint, _max_dist integer DEFAULT NULL::integer)
 ```
+### _prom_catalog.add_compression_clause_to_rollups
+
+```
+procedure void **_prom_catalog.add_compression_clause_to_rollups**(IN _rollup_schema text, IN _table_name text)
+```
 ### _prom_catalog.attach_series_partition
 
 ```
@@ -517,6 +532,17 @@ procedure void **_prom_catalog.compress_old_chunks**(IN _hypertable_schema_name
 ```
 function integer **_prom_catalog.count_jsonb_keys**(j jsonb)
 ```
+### _prom_catalog.counter_reset_sum
+
+```
+function double precision **_prom_catalog.counter_reset_sum**(v double precision[])
+```
+### _prom_catalog.create_cagg_refresh_job_if_not_exists
+Creates a Cagg refresh job that refreshes all Caggs registered by register_metric_view().
+This function creates a refresh job only if no execute_caggs_refresh_policy() exists currently with the given refresh_interval.
+```
+function void **_prom_catalog.create_cagg_refresh_job_if_not_exists**(_refresh_interval interval)
+```
 ### _prom_catalog.create_exemplar_table_if_not_exists
 
 ```
@@ -537,6 +563,11 @@ function text **_prom_catalog.create_ingest_temp_table**(table_name text, schema
 ```
 function record **_prom_catalog.create_label_key**(new_key text, OUT id integer, OUT value_column_name name, OUT id_column_name name)
 ```
+### _prom_catalog.create_metric_rollup_view
+
+```
+function boolean **_prom_catalog.create_metric_rollup_view**(_rollup_schema text, _metric_name text, _table_name text, _resolution interval)
+```
 ### _prom_catalog.create_metric_table
 
 ```
@@ -547,6 +578,26 @@ function record **_prom_catalog.create_metric_table**(metric_name_arg text, OUT
 ```
 function boolean **_prom_catalog.create_metric_view**(metric_name text)
 ```
+### _prom_catalog.create_rollup
+
+```
+procedure void **_prom_catalog.create_rollup**(IN _name text, IN _resolution interval, IN _retention interval)
+```
+### _prom_catalog.create_rollup_for_counter
+
+```
+procedure void **_prom_catalog.create_rollup_for_counter**(IN rollup_schema text, IN table_name text, IN resolution interval)
+```
+### _prom_catalog.create_rollup_for_gauge
+
+```
+procedure void **_prom_catalog.create_rollup_for_gauge**(IN rollup_schema text, IN table_name text, IN resolution interval)
+```
+### _prom_catalog.create_rollup_for_summary
+
+```
+procedure void **_prom_catalog.create_rollup_for_summary**(IN rollup_schema text, IN table_name text, IN resolution interval)
+```
 ### _prom_catalog.create_series
 
 ```
@@ -577,6 +628,11 @@ function void **_prom_catalog.delay_compression_job**(ht_table text, new_start t
 ```
 function void **_prom_catalog.delete_expired_series**(metric_schema text, metric_table text, metric_series_table text, ran_at timestamp with time zone, present_epoch bigint, last_updated_epoch timestamp with time zone)
 ```
+### _prom_catalog.delete_rollup
+
+```
+procedure void **_prom_catalog.delete_rollup**(IN _rollup_name text)
+```
 ### _prom_catalog.delete_series_catalog_row
 
 ```
@@ -607,6 +663,32 @@ ABORT an INSERT transaction due to the ID epoch being out of date
 ```
 function void **_prom_catalog.epoch_abort**(user_epoch bigint)
 ```
+### _prom_catalog.execute_caggs_compression_policy
+execute_caggs_compression_policy is responsible to compress Caggs registered via
+register_metric_view() in _prom_catalog.metric. It goes through all the entries in the _prom_catalog.metric and tries to compress any Cagg that supports compression.
+These include metric-rollups and custom Caggs based downsampling.
+Note: execute_caggs_compression_policy runs every X interval and compresses only the inactive chunks of those Caggs which have timescaledb.compress = true.
+By default, these include metric-rollups.
+```
+procedure void **_prom_catalog.execute_caggs_compression_policy**(IN job_id integer, IN config jsonb)
+```
+### _prom_catalog.execute_caggs_refresh_policy
+execute_caggs_refresh_policy runs every refresh_interval passed in config. Its
+main aim is to refresh those Caggs that have been registered under _prom_catalog.metric and whose view_refresh_interval
+matches the given refresh_interval. It refreshes 2 kinds of Caggs:
+1. Caggs created by metric rollups
+2. Custom Caggs created by the user
+```
+procedure void **_prom_catalog.execute_caggs_refresh_policy**(IN job_id integer, IN config jsonb)
+```
+### _prom_catalog.execute_caggs_retention_policy
+execute_caggs_retention_policy is responsible to perform retention behaviour on compress continuous aggregates registered via
+register_metric_view(). It loops through all entries in the _prom_catalog.metric that are Caggs and tries to delete the stale chunks of those Caggs.
+The staleness is determined by rollup_retention (for metric rollups) and default_retention_period of parent hypertable (for custom Caggs).
+These include metric-rollups and custom Caggs based downsampling.
+```
+procedure void **_prom_catalog.execute_caggs_retention_policy**(IN job_id integer, IN config jsonb)
+```
 ### _prom_catalog.execute_compression_policy
 compress data according to the policy. This procedure should be run regularly in a cron job
 ```
@@ -842,6 +924,11 @@ function bigint **_prom_catalog.insert_metric_metadatas**(t timestamp with time
 ```
 function bigint **_prom_catalog.insert_metric_row**(metric_table text, time_array timestamp with time zone[], value_array double precision[], series_id_array bigint[])
 ```
+### _prom_catalog.irate
+
+```
+function double precision **_prom_catalog.irate**(v double precision[])
+```
 ### _prom_catalog.is_multinode
 
 ```
@@ -990,6 +1077,11 @@ function void **_prom_catalog.resurrect_series_ids**(metric_table text, series_i
 ```
 function bigint **_prom_catalog.safe_approximate_row_count**(table_name_input regclass)
 ```
+### _prom_catalog.scan_for_new_rollups
+
+```
+procedure void **_prom_catalog.scan_for_new_rollups**(IN job_id integer, IN config jsonb)
+```
 ### _prom_catalog.set_app_name
 
 ```

migration/idempotent/001-base.sql

@@ -12,7 +12,8 @@ FROM
     ('trace_retention_period'   , (30 * INTERVAL '1 days')::text),
     ('ha_lease_timeout'         , '1m'),
     ('ha_lease_refresh'         , '10s'),
-    ('epoch_duration'           , (INTERVAL '12 hours')::text)
+    ('epoch_duration'           , (INTERVAL '12 hours')::text),
+    ('automatic_downsample'     , 'true')
 ) d(key, value)
 ;
 GRANT SELECT ON _prom_catalog.initial_default TO prom_reader;
@@ -2562,7 +2563,7 @@ LANGUAGE PLPGSQL;
 REVOKE ALL ON FUNCTION _prom_catalog.create_metric_view(text) FROM PUBLIC;
 GRANT EXECUTE ON FUNCTION _prom_catalog.create_metric_view(text) TO prom_writer;
 
-CREATE OR REPLACE FUNCTION prom_api.register_metric_view(schema_name text, view_name text, if_not_exists BOOLEAN = false)
+CREATE OR REPLACE FUNCTION prom_api.register_metric_view(schema_name text, view_name text, refresh_interval INTERVAL, if_not_exists BOOLEAN = false, rollup_id BIGINT = NULL)
     RETURNS BOOLEAN
     SECURITY DEFINER
     VOLATILE
@@ -2580,29 +2581,49 @@ BEGIN
     AND    table_name   = register_metric_view.view_name;
 
     IF NOT FOUND THEN
-        RAISE EXCEPTION 'cannot register non-existent metric view in specified schema';
+        RAISE EXCEPTION 'cannot register non-existent metric view with name % in specified schema %', view_name, schema_name;
     END IF;
 
     -- cannot register view in data schema
     IF schema_name = 'prom_data' THEN
         RAISE EXCEPTION 'cannot register metric view in prom_data schema';
     END IF;
 
-    -- check if view is based on a metric from prom_data
-    -- we check for two levels so we can support 2-step continuous aggregates
-    SELECT v.view_schema, v.view_name, v.metric_table_name
-    INTO agg_schema, agg_name, metric_table_name
-    FROM _prom_catalog.get_first_level_view_on_metric(schema_name, view_name) v;
+    IF rollup_id IS NOT NULL THEN
+        -- Check if the materialized view is at least created.
+        PERFORM 1 FROM pg_views WHERE viewname = view_name AND schemaname = schema_name;
+        IF NOT FOUND THEN
+            RAISE EXCEPTION 'No materialized view like %.% found', schema_name, view_name;
+        END IF;
 
-    IF NOT FOUND THEN
-        RAISE EXCEPTION 'view not based on a metric table from prom_data schema';
+        IF refresh_interval IS NULL THEN
+            RAISE EXCEPTION 'refresh_interval must not be null for rollup views';
+        END IF;
+
+        -- We do not do the checks offered by get_first_level_view_on_metric() for metric-rollups
+        -- because those checks are not meant for Caggs with timescaledb.materialized_only = true.
+        --
+        -- Since metric-rollups are an internal creation of Caggs, we should be fine with not doing
+        -- "strict" safety checks.
+        metric_table_name := view_name;
+        agg_name := view_name;
+        agg_schema := schema_name;
+    ELSE
+        -- check if view is based on a metric from prom_data
+        -- we check for two levels so we can support 2-step continuous aggregates
+        SELECT v.view_schema, v.view_name, v.metric_table_name
+        INTO agg_schema, agg_name, metric_table_name
+        FROM _prom_catalog.get_first_level_view_on_metric(schema_name, view_name) v;
+
+        IF NOT FOUND THEN
+            RAISE EXCEPTION 'view with name % not based on a metric table from prom_data schema', view_name;
+        END IF;
     END IF;
 
     -- check if the view contains necessary columns with the correct types
-    SELECT count(*) FROM information_schema.columns
-    INTO column_count
+    SELECT count(*) INTO column_count FROM information_schema.columns
     WHERE table_schema = register_metric_view.schema_name
-    AND table_name   = register_metric_view.view_name
+    AND table_name = register_metric_view.view_name
     AND ((column_name = 'time' AND data_type = 'timestamp with time zone')
     OR (column_name = 'series_id' AND data_type = 'bigint')
     OR data_type = 'double precision');
@@ -2611,9 +2632,16 @@ BEGIN
         RAISE EXCEPTION 'view must contain time (data type: timestamp with time zone), series_id (data type: bigint), and at least one column with double precision data type';
     END IF;
 
+    IF refresh_interval IS NULL THEN
+        -- When a non-metric-rollup Cagg is created, we should inform the user that he needs to create a refresh policy himself.
+        RAISE NOTICE 'Automatic refresh is disabled since refresh_interval is NULL. Please create refresh policy for this Cagg';
+    ELSE
+        PERFORM _prom_catalog.create_cagg_refresh_job_if_not_exists(refresh_interval);
+    END IF;
+
     -- insert into metric table
-    INSERT INTO _prom_catalog.metric (metric_name, table_name, table_schema, series_table, is_view, creation_completed)
-    VALUES (register_metric_view.view_name, register_metric_view.view_name, register_metric_view.schema_name, metric_table_name, true, true)
+    INSERT INTO _prom_catalog.metric (metric_name, table_name, table_schema, series_table, is_view, creation_completed, view_refresh_interval, rollup_id)
+    VALUES (register_metric_view.view_name, register_metric_view.view_name, register_metric_view.schema_name, metric_table_name, true, true, refresh_interval, register_metric_view.rollup_id)
     ON CONFLICT DO NOTHING;
 
     IF NOT FOUND THEN
@@ -2638,8 +2666,8 @@ END
 $func$
 LANGUAGE PLPGSQL;
 --redundant given schema settings but extra caution for security definers
-REVOKE ALL ON FUNCTION prom_api.register_metric_view(text, text, boolean) FROM PUBLIC;
-GRANT EXECUTE ON FUNCTION prom_api.register_metric_view(text, text, boolean) TO prom_admin;
+REVOKE ALL ON FUNCTION prom_api.register_metric_view(text, text, interval, boolean, bigint) FROM PUBLIC;
+GRANT EXECUTE ON FUNCTION prom_api.register_metric_view(text, text, interval, boolean, bigint) TO prom_admin;
 
 CREATE OR REPLACE FUNCTION prom_api.unregister_metric_view(schema_name text, view_name text, if_exists BOOLEAN = false)
     RETURNS BOOLEAN