Hypertable invalidations logical decoding plugin

Implementation of a logical decoding plugin for retrieving hypertable
invalidations from the WAL.

This plugin will allow you to create a logical replication slot for
hypertables that will process changes to the hypertables and return
hypertable invalidation ranges.

It also adds two functions to read and write hypertable invalidation
ranges, allowing tests to be written that read the format.

Author: mkindahl

.unreleased/pr_8207

@@ -0,0 +1 @@
+Implements: #8207 Logical decoding plugin for continuous aggregate invalidations

sql/cagg_utils.sql

@@ -27,3 +27,12 @@ CREATE OR REPLACE FUNCTION _timescaledb_functions.cagg_get_bucket_function_info(
     -- fixed or variable sized bucket
     OUT bucket_fixed_width BOOLEAN
 ) RETURNS RECORD AS '@MODULE_PATHNAME@', 'ts_continuous_agg_get_bucket_function_info' LANGUAGE C STRICT VOLATILE;
+
+CREATE OR REPLACE FUNCTION _timescaledb_functions.cagg_parse_invalidation_record(
+    invalidation_record BYTEA,
+    OUT hypertable_relid REGCLASS,
+    OUT lowest_modified_value BIGINT,
+    OUT greatest_modified_value BIGINT)
+RETURNS RECORD
+AS '@MODULE_PATHNAME@', 'ts_continuous_agg_read_invalidation_record'
+LANGUAGE C STRICT IMMUTABLE;

sql/updates/reverse-dev.sql

@@ -3,6 +3,7 @@ DROP PROCEDURE @extschema@.add_process_hypertable_invalidations_policy(REGCLASS,
 DROP PROCEDURE @extschema@.remove_process_hypertable_invalidations_policy(REGCLASS, BOOL);
 DROP PROCEDURE _timescaledb_functions.policy_process_hypertable_invalidations(INTEGER, JSONB);
 DROP FUNCTION _timescaledb_functions.policy_process_hypertable_invalidations_check(JSONB);
+DROP FUNCTION _timescaledb_functions.cagg_parse_invalidation_record(BYTEA);
 
 DROP PROCEDURE IF EXISTS _timescaledb_functions.policy_compression(job_id INTEGER, config JSONB);
 DROP PROCEDURE IF EXISTS _timescaledb_functions.policy_compression_execute(

src/CMakeLists.txt

@@ -108,6 +108,10 @@ if(USE_OPENSSL)
   endif(MSVC)
 endif(USE_OPENSSL)
 
+set(TSL_CAGG_INVALIDATION_PLUGIN_NAME
+    "${PROJECT_NAME}-invalidations"
+    CACHE STRING "Continuous aggregates invalidation plugin name")
+
 configure_file(config.h.in config.h)
 add_dependencies(${PROJECT_NAME} gitcheck)
 include_directories(${CMAKE_CURRENT_SOURCE_DIR})

src/compat/compat.h

@@ -699,3 +699,15 @@ pg_cmp_u32(uint32 a, uint32 b)
 #define make_range_compat(typcache, lower, upper, empty, escontext)                                \
 	make_range(typcache, lower, upper, empty, escontext)
 #endif
+
+/* Copied from PG17. We can remove it once we deprecate older versions. */
+#if PG17_LT
+static inline void
+initReadOnlyStringInfo(StringInfo str, char *data, int len)
+{
+	str->data = data;
+	str->len = len;
+	str->maxlen = 0; /* read-only */
+	str->cursor = 0;
+}
+#endif

src/config.h.in

@@ -11,6 +11,7 @@
 #define BUILD_OS_VERSION "@CMAKE_SYSTEM_VERSION@"
 #define BUILD_PROCESSOR "@CMAKE_SYSTEM_PROCESSOR@"
 #define BUILD_POINTER_BYTES @CMAKE_SIZEOF_VOID_P@
+#define CONTINUOUS_AGGS_HYPERTABLE_INVALIDATION_PLUGIN_NAME "@TSL_CAGG_INVALIDATION_PLUGIN_NAME@"
 
 /*
  * Value should be set in package release scripts. Otherwise

src/cross_module_fn.c

@@ -96,6 +96,7 @@ CROSSMODULE_WRAPPER(continuous_agg_validate_query);
 CROSSMODULE_WRAPPER(continuous_agg_get_bucket_function);
 CROSSMODULE_WRAPPER(continuous_agg_get_bucket_function_info);
 CROSSMODULE_WRAPPER(continuous_agg_migrate_to_time_bucket);
+CROSSMODULE_WRAPPER(continuous_agg_read_invalidation_record);
 CROSSMODULE_WRAPPER(cagg_try_repair);
 
 CROSSMODULE_WRAPPER(chunk_freeze_chunk);
@@ -426,6 +427,7 @@ TSDLLEXPORT CrossModuleFunctions ts_cm_functions_default = {
 	.continuous_agg_get_bucket_function = error_no_default_fn_pg_community,
 	.continuous_agg_get_bucket_function_info = error_no_default_fn_pg_community,
 	.continuous_agg_migrate_to_time_bucket = error_no_default_fn_pg_community,
+	.continuous_agg_read_invalidation_record = error_no_default_fn_pg_community,
 	.cagg_try_repair = process_cagg_try_repair,
 
 	/* compression */

src/cross_module_fn.h

@@ -123,6 +123,7 @@ typedef struct CrossModuleFunctions
 	PGFunction continuous_agg_get_bucket_function_info;
 	PGFunction continuous_agg_migrate_to_time_bucket;
 	PGFunction cagg_try_repair;
+	PGFunction continuous_agg_read_invalidation_record;
 
 	PGFunction compressed_data_send;
 	PGFunction compressed_data_recv;

src/ts_catalog/catalog.h

@@ -1004,9 +1004,16 @@ typedef enum Anum_continuous_aggs_bucket_function_pkey
 #define Natts_continuous_aggs_bucket_function_pkey                                                 \
 	(_Anum_continuous_aggs_bucket_function_pkey_max - 1)
 
-/****** CONTINUOUS_AGGS_HYPERTABLE_INVALIDATION_LOG_TABLE definitions*/
+/*
+ * CONTINUOUS_AGGS_HYPERTABLE_INVALIDATION_LOG_TABLE definitions
+ *
+ * The definition of CONTINUOUS_AGGS_HYPERTABLE_INVALIDATION_PLUGIN_NAME is
+ * generated from config.h.in and can be found in the generated file.
+ */
 #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
 {
 	Anum_continuous_aggs_hypertable_invalidation_log_hypertable_id = 1,

tsl/src/continuous_aggs/CMakeLists.txt

@@ -3,6 +3,7 @@ set(SOURCES
     ${CMAKE_CURRENT_SOURCE_DIR}/create.c
     ${CMAKE_CURRENT_SOURCE_DIR}/finalize.c
     ${CMAKE_CURRENT_SOURCE_DIR}/insert.c
+    ${CMAKE_CURRENT_SOURCE_DIR}/invalidation_record.c
     ${CMAKE_CURRENT_SOURCE_DIR}/invalidation_threshold.c
     ${CMAKE_CURRENT_SOURCE_DIR}/invalidation.c
     ${CMAKE_CURRENT_SOURCE_DIR}/materialize.c
@@ -12,3 +13,21 @@ set(SOURCES
     ${CMAKE_CURRENT_SOURCE_DIR}/repair.c
     ${CMAKE_CURRENT_SOURCE_DIR}/utils.c)
 target_sources(${TSL_LIBRARY_NAME} PRIVATE ${SOURCES})
+
+add_library(
+  ${TSL_CAGG_INVALIDATION_PLUGIN_NAME} MODULE
+  invalidation_plugin.c invalidation_plugin_cache.c invalidation_record.c)
+
+# VERSION and SOVERSION are not used since they do not make sense for modules,
+# and actually breaks the build on MacOS.
+#
+# See https://gitlab.kitware.com/cmake/cmake/-/issues/20782.
+set_target_properties(
+  ${TSL_CAGG_INVALIDATION_PLUGIN_NAME}
+  PROPERTIES OUTPUT_NAME ${TSL_CAGG_INVALIDATION_PLUGIN_NAME}
+             PREFIX ""
+             POSITION_INDEPENDENT_CODE ON)
+target_include_directories(${TSL_CAGG_INVALIDATION_PLUGIN_NAME}
+                           PRIVATE ${PG_INCLUDEDIR})
+install(TARGETS ${TSL_CAGG_INVALIDATION_PLUGIN_NAME}
+        LIBRARY DESTINATION ${PG_PKGLIBDIR})