Add invalidate_using to continuous aggregates

Add new storage option `timescaledb.invalidate_using` to continuous
aggregates for source of invalidations: either use `trigger` to read
invalidations from hypertable using a trigger, or use `wal` to read
invalidations for hypertables from WAL using logical decoding.

It is not possible to attach continuous aggregates to hypertables using
mixed collection method and an error will be thrown if you try to
attach a continuous aggregate to a hypertable that is already using a
different collection method.

If no `invalidate_using` option is given to the continuous aggregate, it
will use whatever collection method is already used for the hypertable.

Author: mkindahl

.unreleased/pr_8306

@@ -0,0 +1 @@
+Implements: #8306 Add option for invalidation collection using WAL for continuous aggregates

sql/pre_install/types.functions.sql

@@ -58,3 +58,8 @@ CREATE OR REPLACE FUNCTION _timescaledb_functions.dimension_info_out(_timescaled
 -- Type for bloom filters used by the sparse indexes on compressed hypertables.
 CREATE OR REPLACE FUNCTION _timescaledb_functions.bloom1in(cstring) RETURNS _timescaledb_internal.bloom1 AS 'byteain' LANGUAGE INTERNAL STRICT IMMUTABLE PARALLEL SAFE;
 CREATE OR REPLACE FUNCTION _timescaledb_functions.bloom1out(_timescaledb_internal.bloom1) RETURNS cstring AS 'byteaout' LANGUAGE INTERNAL STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION _timescaledb_functions.has_invalidation_trigger(regclass)
+   RETURNS bool
+   LANGUAGE C STRICT PARALLEL SAFE
+   AS '@MODULE_PATHNAME@', 'ts_has_invalidation_trigger';

sql/updates/reverse-dev.sql

@@ -1,4 +1,9 @@
+ALTER EXTENSION timescaledb DROP VIEW timescaledb_information.continuous_aggregates;
+
+DROP VIEW timescaledb_information.continuous_aggregates;
+
 DROP FUNCTION _timescaledb_functions.cagg_parse_invalidation_record(BYTEA);
+DROP FUNCTION _timescaledb_functions.has_invalidation_trigger(regclass);
 
 CREATE FUNCTION ts_hypercore_handler(internal) RETURNS table_am_handler
 AS '@MODULE_PATHNAME@', 'ts_hypercore_handler' LANGUAGE C;

sql/views.sql

@@ -122,7 +122,11 @@ SELECT ht.schema_name AS hypertable_schema,
   mat_ht.schema_name AS materialization_hypertable_schema,
   mat_ht.table_name AS materialization_hypertable_name,
   directview.viewdefinition AS view_definition,
-  cagg.finalized
+  cagg.finalized,
+  CASE WHEN _timescaledb_functions.has_invalidation_trigger(format('%I.%I', ht.schema_name, ht.table_name)::regclass)
+       THEN 'trigger'
+       ELSE 'wal'
+  END AS invalidate_using
 FROM _timescaledb_catalog.continuous_agg cagg,
   _timescaledb_catalog.hypertable ht,
   LATERAL (

src/hypertable.c

@@ -607,6 +607,17 @@ ts_hypertable_create_trigger(const Hypertable *ht, CreateTrigStmt *stmt, const c
 	return root_trigger_addr;
 }
 
+TSDLLEXPORT void
+ts_hypertable_drop_invalidation_replication_slot(const char *slot_name)
+{
+	CatalogSecurityContext sec_ctx;
+	NameData slot;
+	namestrcpy(&slot, slot_name);
+	ts_catalog_database_info_become_owner(ts_catalog_database_info_get(), &sec_ctx);
+	DirectFunctionCall1(pg_drop_replication_slot, NameGetDatum(&slot));
+	ts_catalog_restore_user(&sec_ctx);
+}
+
 /* based on RemoveObjects */
 TSDLLEXPORT void
 ts_hypertable_drop_trigger(Oid relid, const char *trigger_name)

src/hypertable.h

@@ -116,6 +116,7 @@ extern int ts_hypertable_delete_by_id(int32 hypertable_id);
 extern TSDLLEXPORT ObjectAddress ts_hypertable_create_trigger(const Hypertable *ht,
 															  CreateTrigStmt *stmt,
 															  const char *query);
+extern TSDLLEXPORT void ts_hypertable_drop_invalidation_replication_slot(const char *slot_name);
 extern TSDLLEXPORT void ts_hypertable_drop_trigger(Oid relid, const char *trigger_name);
 extern TSDLLEXPORT void ts_hypertable_drop(Hypertable *hypertable, DropBehavior behavior);
 

src/ts_catalog/catalog.h

@@ -1012,7 +1012,6 @@ typedef enum Anum_continuous_aggs_bucket_function_pkey
  */
 #define CONTINUOUS_AGGS_HYPERTABLE_INVALIDATION_LOG_TABLE_NAME                                     \
 	"continuous_aggs_hypertable_invalidation_log"
-#define CONTINUOUS_AGGS_HYPERTABLE_INVALIDATION_SLOT_NAME "continuous_aggs_hypertable_invalidations"
 
 typedef enum Anum_continuous_aggs_hypertable_invalidation_log
 {

src/ts_catalog/continuous_agg.c

@@ -10,13 +10,17 @@
  */
 
 #include <postgres.h>
+
 #include <access/htup_details.h>
 #include <catalog/dependency.h>
 #include <catalog/namespace.h>
 #include <catalog/pg_trigger.h>
 #include <commands/trigger.h>
+#include <executor/spi.h>
 #include <fmgr.h>
+#include <lib/stringinfo.h>
 #include <nodes/makefuncs.h>
+#include <replication/slot.h>
 #include <storage/lmgr.h>
 #include <utils/acl.h>
 #include <utils/builtins.h>
@@ -46,6 +50,14 @@
 #define BUCKET_FUNCTION_SERIALIZE_VERSION 1
 #define CHECK_NAME_MATCH(name1, name2) (namestrcmp(name1, name2) == 0)
 
+TS_FUNCTION_INFO_V1(ts_has_invalidation_trigger);
+
+Datum
+ts_has_invalidation_trigger(PG_FUNCTION_ARGS)
+{
+	PG_RETURN_BOOL(has_invalidation_trigger(PG_GETARG_OID(0)));
+}
+
 static void
 init_scan_by_mat_hypertable_id(ScanIterator *iterator, const int32 mat_hypertable_id)
 {
@@ -193,6 +205,12 @@ hypertable_invalidation_log_delete(int32 raw_hypertable_id)
 	}
 }
 
+void
+ts_get_invalidation_replication_slot_name(char *slotname, Size szslot)
+{
+	snprintf(slotname, szslot, "ts_%u_cagg", MyDatabaseId);
+}
+
 void
 ts_materialization_invalidation_log_delete_inner(int32 mat_hypertable_id)
 {
@@ -502,6 +520,37 @@ ts_continuous_agg_get_all_caggs_info(int32 raw_hypertable_id)
 	return all_caggs_info;
 }
 
+/*
+ * Return true if there is any continuous aggregate that is using WAL-based
+ * invalidation collection.
+ *
+ * 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)
+{
+	ScanIterator iterator =
+		ts_scan_iterator_create(CONTINUOUS_AGG, AccessShareLock, CurrentMemoryContext);
+	ts_scanner_foreach(&iterator)
+	{
+		bool isnull;
+		Datum datum = slot_getattr(ts_scan_iterator_slot(&iterator),
+								   Anum_continuous_agg_raw_hypertable_id,
+								   &isnull);
+
+		Assert(!isnull);
+		Oid relid = ts_hypertable_id_to_relid(DatumGetInt32(datum), true);
+		if (!has_invalidation_trigger(relid))
+		{
+			ts_scan_iterator_close(&iterator);
+			return true;
+		}
+	}
+	ts_scan_iterator_close(&iterator);
+	return false;
+}
+
 TSDLLEXPORT ContinuousAggHypertableStatus
 ts_continuous_agg_hypertable_status(int32 hypertable_id)
 {
@@ -829,6 +878,10 @@ drop_continuous_agg(FormData_continuous_agg *cadata, bool drop_user_view)
 	 *
 	 * AccessExclusiveLock is needed to drop triggers and also prevent
 	 * concurrent DML commands.
+	 *
+	 * It is needed also in the case that we are using WAL-based invalidation
+	 * collection since we want to serialize create and drop of continuous
+	 * aggregates.
 	 */
 	if (drop_user_view)
 		user_view = get_and_lock_rel_by_name(&cadata->user_view_schema,
@@ -856,17 +909,13 @@ drop_continuous_agg(FormData_continuous_agg *cadata, bool drop_user_view)
 		LockRelationOid(catalog_get_table_id(catalog, CONTINUOUS_AGGS_INVALIDATION_THRESHOLD),
 						RowExclusiveLock);
 
-		/* The trigger will be dropped if the hypertable still exists and no other
+		/* The trigger will be dropped if it exists (it does not for WAL-based
+		 * invalidation collection), the hypertable still exists and no other
 		 * caggs attached. */
-		if (OidIsValid(raw_hypertable.objectId))
+		Oid tgoid = get_trigger_oid(raw_hypertable.objectId, CAGGINVAL_TRIGGER_NAME, true);
+		if (OidIsValid(raw_hypertable.objectId) && OidIsValid(tgoid))
 		{
-			ObjectAddressSet(raw_hypertable_trig,
-							 TriggerRelationId,
-							 get_trigger_oid(raw_hypertable.objectId,
-											 CAGGINVAL_TRIGGER_NAME,
-											 false));
-
-			/* Raw hypertable is locked above */
+			ObjectAddressSet(raw_hypertable_trig, TriggerRelationId, tgoid);
 			LockRelationOid(raw_hypertable_trig.objectId, AccessExclusiveLock);
 		}
 	}
@@ -924,6 +973,18 @@ drop_continuous_agg(FormData_continuous_agg *cadata, bool drop_user_view)
 		ts_hypertable_drop_trigger(raw_hypertable.objectId, CAGGINVAL_TRIGGER_NAME);
 	}
 
+	/*
+	 * Drop invalidation slot if there are no hypertables using WAL-based
+	 * invalidation collection.
+	 *
+	 * This is important since there is no actor that reads the slot, which
+	 * means that the WAL cannot be pruned.
+	 */
+	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)
+		ts_hypertable_drop_invalidation_replication_slot(slot_name);
+
 	if (OidIsValid(mat_hypertable.objectId))
 	{
 		performDeletion(&mat_hypertable, DROP_CASCADE, 0);

src/ts_catalog/continuous_agg.h

@@ -16,6 +16,7 @@
 #include "with_clause/with_clause_parser.h"
 
 #define CAGGINVAL_TRIGGER_NAME "ts_cagg_invalidation_trigger"
+#define TS_INVALIDATION_SLOT_NAME_MAX (32)
 
 /*switch to ts user for _timescaledb_internal access */
 #define SWITCH_TO_TS_USER(schemaname, newuid, saved_uid, saved_secctx)                             \
@@ -54,6 +55,13 @@ typedef enum ContinuousAggViewType
 	ContinuousAggAnyView
 } ContinuousAggViewType;
 
+typedef enum ContinuousAggInvalidateUsing
+{
+	ContinuousAggInvalidateUsingDefault = 0,
+	ContinuousAggInvalidateUsingTrigger,
+	ContinuousAggInvalidateUsingWal,
+} ContinuousAggInvalidateUsing;
+
 /*
  * Information about the bucketing function.
  */
@@ -146,6 +154,12 @@ typedef struct CaggPolicyOffset
 	const char *name;
 } CaggPolicyOffset;
 
+static inline bool
+has_invalidation_trigger(Oid relid)
+{
+	return OidIsValid(get_trigger_oid(relid, CAGGINVAL_TRIGGER_NAME, true));
+}
+
 extern TSDLLEXPORT Oid ts_cagg_permissions_check(Oid cagg_oid, Oid userid);
 
 extern TSDLLEXPORT CaggsInfo ts_continuous_agg_get_all_caggs_info(int32 raw_hypertable_id);
@@ -199,3 +213,4 @@ extern TSDLLEXPORT int64
 ts_continuous_agg_fixed_bucket_width(const ContinuousAggsBucketFunction *bucket_function);
 extern TSDLLEXPORT int64
 ts_continuous_agg_bucket_width(const ContinuousAggsBucketFunction *bucket_function);
+extern TSDLLEXPORT void ts_get_invalidation_replication_slot_name(char *slotname, Size szslot);

src/with_clause/create_materialized_view_with_clause.c

@@ -59,6 +59,11 @@ static const WithClauseDefinition continuous_aggregate_with_clause_def[] = {
         .arg_names = {"compress_chunk_interval", "compress_chunk_time_interval", NULL},
          .type_id = INTERVALOID,
     },
+	[CreateMaterializedViewFlagInvalidateUsing] = {
+        .arg_names = {"invalidate_using", NULL},
+		.type_id = TEXTOID,
+		.default_val = (Datum) 0,
+	},
 };
 
 WithClauseResult *