Add migration to/from WAL-based invalidation

To support migration from trigger-based invalidation collection to
WAL-based invalidation collection, or the opposite, the function
`_timescaledb_functions.cagg_change_invalidation_method` is introduced.
When called with a hypertable (that has continuous aggregates
connected) and an invalidation method, it will flush the hypertable
invalidations and changes the collection method for the hypertable to
use the requested one.

It will changes the collection method for *all* continuous aggregates
connected to the given hypertable, not just for a single continuous
aggregate.

Also adds tests for the migration to check that it works properly.

Author: mkindahl

.unreleased/pr_8738

@@ -0,0 +1 @@
+Implements: #8738 Add function to migrate between WAL-based and trigger-based invalidation collection

sql/cagg_api.sql

@@ -42,7 +42,7 @@ DECLARE
 BEGIN
    SELECT ht.id AS hypertable_id,
           di.column_type::regtype,
-          EXISTS(SELECT FROM _timescaledb_catalog.continuous_agg where raw_hypertable_id = ht.id) AS has_cagg
+          EXISTS(SELECT FROM _timescaledb_catalog.continuous_agg WHERE raw_hypertable_id = ht.id) AS has_cagg
      INTO info
      FROM _timescaledb_catalog.hypertable ht
      JOIN _timescaledb_catalog.dimension di ON ht.id = di.hypertable_id

sql/cagg_migrate.sql

@@ -674,3 +674,10 @@ $BODY$;
 -- into a CAgg using the regular time_bucket function
 CREATE OR REPLACE PROCEDURE _timescaledb_functions.cagg_migrate_to_time_bucket(cagg REGCLASS)
 AS '@MODULE_PATHNAME@', 'ts_continuous_agg_migrate_to_time_bucket' LANGUAGE C;
+
+-- Migrate all continuous aggregates attached to a hypertable to use
+-- WAL-based invalidation or trigger-based invalidation.
+CREATE OR REPLACE PROCEDURE _timescaledb_functions.set_invalidation_method(
+    hypertable REGCLASS,
+    method NAME
+) AS '@MODULE_PATHNAME@', 'ts_continuous_agg_change_collection_method' LANGUAGE C;

src/cross_module_fn.c

@@ -94,6 +94,7 @@ CROSSMODULE_WRAPPER(bloom1_contains);
 CROSSMODULE_WRAPPER(continuous_agg_invalidation_trigger);
 CROSSMODULE_WRAPPER(continuous_agg_refresh);
 CROSSMODULE_WRAPPER(continuous_agg_process_hypertable_invalidations);
+CROSSMODULE_WRAPPER(continuous_agg_change_collection_method);
 CROSSMODULE_WRAPPER(continuous_agg_validate_query);
 CROSSMODULE_WRAPPER(continuous_agg_get_bucket_function);
 CROSSMODULE_WRAPPER(continuous_agg_get_bucket_function_info);
@@ -364,6 +365,7 @@ TSDLLEXPORT CrossModuleFunctions ts_cm_functions_default = {
 	.continuous_agg_process_hypertable_invalidations = error_no_default_fn_pg_community,
 	.continuous_agg_invalidate_raw_ht = continuous_agg_invalidate_raw_ht_all_default,
 	.continuous_agg_invalidate_mat_ht = continuous_agg_invalidate_mat_ht_all_default,
+	.continuous_agg_change_collection_method = error_no_default_fn_pg_community,
 	.continuous_agg_update_options = continuous_agg_update_options_default,
 	.continuous_agg_validate_query = error_no_default_fn_pg_community,
 	.continuous_agg_get_bucket_function = error_no_default_fn_pg_community,

src/cross_module_fn.h

@@ -106,6 +106,7 @@ typedef struct CrossModuleFunctions
 	PGFunction continuous_agg_invalidation_trigger;
 	PGFunction continuous_agg_refresh;
 	PGFunction continuous_agg_process_hypertable_invalidations;
+	PGFunction continuous_agg_change_collection_method;
 	void (*continuous_agg_invalidate_raw_ht)(const Hypertable *raw_ht, int64 start, int64 end);
 	void (*continuous_agg_invalidate_mat_ht)(const Hypertable *raw_ht, const Hypertable *mat_ht,
 											 int64 start, int64 end);

src/ts_catalog/catalog.h

@@ -1333,7 +1333,7 @@ extern bool ts_is_catalog_table(Oid relid);
 
 /* Functions should operate on a passed-in Catalog struct */
 static inline Oid
-catalog_get_table_id(Catalog *catalog, CatalogTable tableid)
+catalog_get_table_id(const Catalog *catalog, CatalogTable tableid)
 {
 	return catalog->tables[tableid].id;
 }

src/ts_catalog/continuous_agg.c

@@ -155,8 +155,8 @@ init_materialization_invalidation_log_scan_by_materialization_id(ScanIterator *i
 		Int32GetDatum(materialization_id));
 }
 
-static int32
-number_of_continuous_aggs_attached(int32 raw_hypertable_id)
+int32
+ts_number_of_continuous_aggs_attached(int32 raw_hypertable_id)
 {
 	ScanIterator iterator =
 		ts_scan_iterator_create(CONTINUOUS_AGG, AccessShareLock, CurrentMemoryContext);
@@ -537,8 +537,8 @@ ts_continuous_agg_get_all_caggs_info(int32 raw_hypertable_id)
  * A hypertable is using the WAL-based invalidation collection if it has a
  * attached continuous aggregate but does not have an invalidation trigger.
  */
-static bool
-hypertable_invalidation_slot_used(void)
+bool
+ts_hypertable_invalidation_slot_used(void)
 {
 	ScanIterator iterator =
 		ts_scan_iterator_create(CONTINUOUS_AGG, AccessShareLock, CurrentMemoryContext);
@@ -910,7 +910,7 @@ drop_continuous_agg(FormData_continuous_agg *cadata, bool drop_user_view)
 
 	raw_hypertable_has_other_caggs =
 		OidIsValid(raw_hypertable.objectId) &&
-		number_of_continuous_aggs_attached(cadata->raw_hypertable_id) > 1;
+		ts_number_of_continuous_aggs_attached(cadata->raw_hypertable_id) > 1;
 
 	if (!raw_hypertable_has_other_caggs)
 	{
@@ -992,7 +992,8 @@ drop_continuous_agg(FormData_continuous_agg *cadata, bool drop_user_view)
 	 */
 	char slot_name[TS_INVALIDATION_SLOT_NAME_MAX];
 	ts_get_invalidation_replication_slot_name(slot_name, sizeof(slot_name));
-	if (!hypertable_invalidation_slot_used() && SearchNamedReplicationSlot(slot_name, true) != NULL)
+	if (!ts_hypertable_invalidation_slot_used() &&
+		SearchNamedReplicationSlot(slot_name, true) != NULL)
 		ts_hypertable_drop_invalidation_replication_slot(slot_name);
 
 	if (OidIsValid(mat_hypertable.objectId))

src/ts_catalog/continuous_agg.h

@@ -177,6 +177,7 @@ extern TSDLLEXPORT ContinuousAgg *ts_continuous_agg_find_by_view_name(const char
 																	  ContinuousAggViewType type);
 extern TSDLLEXPORT ContinuousAgg *ts_continuous_agg_find_by_relid(Oid relid);
 extern TSDLLEXPORT ContinuousAgg *ts_continuous_agg_find_by_rv(const RangeVar *rv);
+extern TSDLLEXPORT bool ts_hypertable_invalidation_slot_used(void);
 
 extern bool ts_continuous_agg_drop(const char *view_schema, const char *view_name);
 extern void ts_continuous_agg_drop_hypertable_callback(int32 hypertable_id);
@@ -214,3 +215,4 @@ ts_continuous_agg_fixed_bucket_width(const ContinuousAggBucketFunction *bucket_f
 extern TSDLLEXPORT int64
 ts_continuous_agg_bucket_width(const ContinuousAggBucketFunction *bucket_function);
 extern TSDLLEXPORT void ts_get_invalidation_replication_slot_name(char *slotname, Size szslot);
+extern TSDLLEXPORT int32 ts_number_of_continuous_aggs_attached(int32 raw_hypertable_id);

tsl/src/continuous_aggs/common.c

@@ -1653,3 +1653,32 @@ cagg_find_groupingcols(ContinuousAgg *agg, Hypertable *mat_ht)
 	}
 	return retlist;
 }
+
+/*
+ * Get all hypertables that are using WAL-invalidations.
+ */
+List *
+get_all_wal_using_hypertables(void)
+{
+	ScanIterator iterator =
+		ts_scan_iterator_create(CONTINUOUS_AGG, AccessShareLock, CurrentMemoryContext);
+	List *hypertables = NIL;
+
+	/* Collect OID of all tables using continuous aggregates */
+	ts_scanner_foreach(&iterator)
+	{
+		bool isnull;
+		Datum datum = slot_getattr(ts_scan_iterator_slot(&iterator),
+								   Anum_continuous_agg_raw_hypertable_id,
+								   &isnull);
+
+		Assert(!isnull);
+		int32 hypertable_id = DatumGetInt32(datum);
+		Oid relid = ts_hypertable_id_to_relid(hypertable_id, false);
+		if (!has_invalidation_trigger(relid))
+			hypertables = list_append_unique_int(hypertables, hypertable_id);
+	}
+	ts_scan_iterator_close(&iterator);
+
+	return hypertables;
+}

tsl/src/continuous_aggs/common.h

@@ -151,4 +151,5 @@ cagg_get_time_min(const ContinuousAgg *cagg)
 	return ts_time_get_min(cagg->partition_type);
 }
 
-ContinuousAggBucketFunction *ts_cagg_get_bucket_function_info(Oid view_oid);
+extern ContinuousAggBucketFunction *ts_cagg_get_bucket_function_info(Oid view_oid);
+extern List *get_all_wal_using_hypertables(void);