One step policy for continuous aggregates

sql/policy_api.sql

@@ -51,7 +51,61 @@ RETURNS INTEGER
 AS '@MODULE_PATHNAME@', 'ts_policy_refresh_cagg_add'
 LANGUAGE C VOLATILE;
 
-CREATE OR REPLACE FUNCTION @[email protected]_continuous_aggregate_policy(continuous_aggregate REGCLASS, if_not_exists BOOL = false)
+CREATE OR REPLACE FUNCTION @[email protected]_continuous_aggregate_policy(
+    continuous_aggregate REGCLASS,
+    if_not_exists BOOL = false, -- deprecating this argument, if_exists overrides it
+    if_exists BOOL = NULL) -- when NULL get the value from if_not_exists
+
 RETURNS VOID
 AS '@MODULE_PATHNAME@', 'ts_policy_refresh_cagg_remove'
-LANGUAGE C VOLATILE STRICT;
+LANGUAGE C VOLATILE;
+
+/* 1 step policies */
+
+/* Add policies */
+CREATE OR REPLACE FUNCTION timescaledb_experimental.add_policies(
+    relation REGCLASS,
+    if_not_exists BOOL = false,
+    refresh_start_offset "any" = NULL,
+    refresh_end_offset "any" = NULL,
+    compress_after "any" = NULL,
+    drop_after "any" = NULL)
+RETURNS BOOL
+AS '@MODULE_PATHNAME@', 'ts_policies_add'
+LANGUAGE C VOLATILE;
+
+/* Remove policies */
+CREATE OR REPLACE FUNCTION timescaledb_experimental.remove_policies(
+    relation REGCLASS,
+    if_exists BOOL = false,
+    VARIADIC policy_names TEXT[] = NULL)
+RETURNS BOOL
+AS '@MODULE_PATHNAME@', 'ts_policies_remove'
+LANGUAGE C VOLATILE;
+
+/* Remove all policies */
+CREATE OR REPLACE FUNCTION timescaledb_experimental.remove_all_policies(
+    relation REGCLASS,
+    if_exists BOOL = false)
+RETURNS BOOL
+AS '@MODULE_PATHNAME@', 'ts_policies_remove_all'
+LANGUAGE C VOLATILE;
+
+/* Alter policies */
+CREATE OR REPLACE FUNCTION timescaledb_experimental.alter_policies(
+    relation REGCLASS,
+    if_exists BOOL = false,
+    refresh_start_offset "any" = NULL,
+    refresh_end_offset "any" = NULL,
+    compress_after "any" = NULL,
+    drop_after "any" = NULL)
+RETURNS BOOL
+AS '@MODULE_PATHNAME@', 'ts_policies_alter'
+LANGUAGE C VOLATILE;
+
+/* Show policies info */
+CREATE OR REPLACE FUNCTION timescaledb_experimental.show_policies(
+    relation REGCLASS)
+RETURNS SETOF JSONB
+AS '@MODULE_PATHNAME@', 'ts_policies_show'
+LANGUAGE C  VOLATILE;

sql/updates/latest-dev.sql

@@ -2,3 +2,4 @@ DROP FUNCTION IF EXISTS @[email protected]_retention_policy(REGCLASS, "any", BOOL);
 
 DROP FUNCTION IF EXISTS @[email protected]_compression_policy(REGCLASS, "any", BOOL);
 DROP FUNCTION IF EXISTS @[email protected]_data_node;
+DROP FUNCTION IF EXISTS @[email protected]_continuous_aggregate_policy(REGCLASS, BOOL);

sql/updates/reverse-dev.sql

@@ -15,3 +15,17 @@ CREATE FUNCTION @[email protected]_data_node(
     repartition            BOOLEAN = TRUE
 ) RETURNS INTEGER
 AS '@MODULE_PATHNAME@', 'ts_data_node_detach' LANGUAGE C VOLATILE;
+
+DROP FUNCTION IF EXISTS timescaledb_experimental.add_policies;
+DROP FUNCTION IF EXISTS timescaledb_experimental.remove_policies;
+DROP FUNCTION IF EXISTS timescaledb_experimental.remove_all_policies;
+DROP FUNCTION IF EXISTS timescaledb_experimental.alter_policies;
+DROP FUNCTION IF EXISTS timescaledb_experimental.show_policies;
+DROP FUNCTION IF EXISTS @[email protected]_continuous_aggregate_policy(REGCLASS, BOOL, BOOL);
+CREATE FUNCTION @[email protected]_continuous_aggregate_policy(continuous_aggregate REGCLASS, if_not_exists BOOL = false)
+RETURNS VOID
+AS '@MODULE_PATHNAME@', 'ts_policy_refresh_cagg_remove'
+LANGUAGE C VOLATILE STRICT;
+
+DROP VIEW IF EXISTS timescaledb_experimental.policies;
+

sql/views_experimental.sql

@@ -25,4 +25,17 @@ INNER JOIN _timescaledb_catalog.hypertable h ON (h.id = c.hypertable_id)
 GROUP BY h.id, c.id, hypertable_schema, hypertable_name, chunk_schema, chunk_name
 ORDER BY h.id, c.id, hypertable_schema, hypertable_name, chunk_schema, chunk_name;
 
+CREATE OR REPLACE VIEW timescaledb_experimental.policies AS
+SELECT ca.view_name AS relation_name,
+  ca.view_schema AS relation_schema,
+  j.schedule_interval,
+  j.proc_schema,
+  j.proc_name,
+  j.config,
+  ht.schema_name AS hypertable_schema,
+  ht.table_name AS hypertable_name
+FROM _timescaledb_config.bgw_job j, timescaledb_information.continuous_aggregates ca,
+ _timescaledb_catalog.hypertable ht
+WHERE ht.id = j.hypertable_id AND ca.view_schema = hypertable_schema;
+
 GRANT SELECT ON ALL TABLES IN SCHEMA timescaledb_experimental TO PUBLIC;

src/cross_module_fn.c

@@ -51,6 +51,12 @@ CROSSMODULE_WRAPPER(copy_chunk_proc);
 CROSSMODULE_WRAPPER(copy_chunk_cleanup_proc);
 CROSSMODULE_WRAPPER(subscription_exec);
 
+CROSSMODULE_WRAPPER(policies_add);
+CROSSMODULE_WRAPPER(policies_remove);
+CROSSMODULE_WRAPPER(policies_remove_all);
+CROSSMODULE_WRAPPER(policies_alter);
+CROSSMODULE_WRAPPER(policies_show);
+
 /* partialize/finalize aggregate */
 CROSSMODULE_WRAPPER(partialize_agg);
 CROSSMODULE_WRAPPER(finalize_agg_sfunc);
@@ -404,6 +410,12 @@ TSDLLEXPORT CrossModuleFunctions ts_cm_functions_default = {
 	.subscription_exec = error_no_default_fn_pg_community,
 	.reorder_chunk = error_no_default_fn_pg_community,
 
+	.policies_add = error_no_default_fn_pg_community,
+	.policies_remove = error_no_default_fn_pg_community,
+	.policies_remove_all = error_no_default_fn_pg_community,
+	.policies_alter = error_no_default_fn_pg_community,
+	.policies_show = error_no_default_fn_pg_community,
+
 	.partialize_agg = error_no_default_fn_pg_community,
 	.finalize_agg_sfunc = error_no_default_fn_pg_community,
 	.finalize_agg_ffunc = error_no_default_fn_pg_community,

src/cross_module_fn.h

@@ -53,6 +53,12 @@ typedef struct CrossModuleFunctions
 	PGFunction policy_retention_proc;
 	PGFunction policy_retention_remove;
 
+	PGFunction policies_add;
+	PGFunction policies_remove;
+	PGFunction policies_remove_all;
+	PGFunction policies_alter;
+	PGFunction policies_show;
+
 	PGFunction job_add;
 	PGFunction job_alter;
 	PGFunction job_alter_set_hypertable_id;

src/ts_catalog/continuous_agg.h

@@ -125,6 +125,14 @@ typedef struct CaggsInfoData
 	List *bucket_functions;
 } CaggsInfo;
 
+typedef struct CaggPolicyOffset
+{
+	Datum value;
+	Oid type;
+	bool isnull;
+	const char *name;
+} CaggPolicyOffset;
+
 extern TSDLLEXPORT const CaggsInfo ts_continuous_agg_get_all_caggs_info(int32 raw_hypertable_id);
 extern TSDLLEXPORT void ts_populate_caggs_info_from_arrays(ArrayType *mat_hypertable_ids,
 														   ArrayType *bucket_widths,

test/expected/pg_dump.out

@@ -558,6 +558,7 @@ WHERE   refclassid = 'pg_catalog.pg_extension'::pg_catalog.regclass AND
         ORDER BY objid::text DESC;
                        objid                       
 ---------------------------------------------------
+ timescaledb_experimental.policies
  timescaledb_experimental.chunk_replication_status
  timescaledb_information.compression_settings
  timescaledb_information.dimensions
@@ -575,7 +576,7 @@ WHERE   refclassid = 'pg_catalog.pg_extension'::pg_catalog.regclass AND
  _timescaledb_internal.bgw_policy_chunk_stats
  _timescaledb_internal.bgw_job_stat
  _timescaledb_catalog.tablespace_id_seq
-(17 rows)
+(18 rows)
 
 -- Make sure we can't run our restoring functions as a normal perm user as that would disable functionality for the whole db
 \c :TEST_DBNAME :ROLE_DEFAULT_PERM_USER

tsl/src/bgw_policy/CMakeLists.txt

@@ -5,6 +5,7 @@ set(SOURCES
     ${CMAKE_CURRENT_SOURCE_DIR}/job_api.c
     ${CMAKE_CURRENT_SOURCE_DIR}/reorder_api.c
     ${CMAKE_CURRENT_SOURCE_DIR}/retention_api.c
-    ${CMAKE_CURRENT_SOURCE_DIR}/policy_utils.c)
+    ${CMAKE_CURRENT_SOURCE_DIR}/policy_utils.c
+    ${CMAKE_CURRENT_SOURCE_DIR}/policies_v2.c)
 target_sources(${TSL_LIBRARY_NAME} PRIVATE ${SOURCES})
 target_include_directories(${TSL_LIBRARY_NAME} PRIVATE ${CMAKE_SOURCE_DIR})