New catalogs for CAgg granular refresh

.unreleased/pr_10299

@@ -0,0 +1 @@
+Implements: #10299 Add continuous_aggs_tenant_tracking and hypertable_cagg_settings catalogs

sql/pre_install/tables.sql

@@ -440,6 +440,35 @@ SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_aggs
 
 CREATE INDEX continuous_aggs_jobs_refresh_ranges_idx ON _timescaledb_catalog.continuous_aggs_jobs_refresh_ranges (materialization_id);
 
+-- Per-tenant invalidation tracking, used in granular refresh of contiuous aggregates.
+CREATE TABLE _timescaledb_catalog.continuous_aggs_tenant_tracking (
+  hypertable_id integer NOT NULL,
+  tenant_id text,
+  min_timestamp bigint NOT NULL,
+  max_timestamp bigint NOT NULL,
+  seqnum integer NOT NULL,
+  -- table constraints
+  CONSTRAINT continuous_aggs_tenant_tracking_hypertable_id_fkey FOREIGN KEY (hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE
+);
+
+SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_aggs_tenant_tracking', '');
+
+CREATE INDEX continuous_aggs_tenant_tracking_idx ON _timescaledb_catalog.continuous_aggs_tenant_tracking (hypertable_id, seqnum);
+
+-- Per-hypertable settings for granular refresh of continuous aggregates.
+-- Row existence means granular refresh is configured for the hypertable.
+CREATE TABLE _timescaledb_catalog.hypertable_cagg_settings (
+  hypertable_id integer NOT NULL,
+  granular_refresh_column name NOT NULL,
+  granular_refresh_start_offset text,
+  granular_refresh_end_offset text,
+  -- table constraints
+  CONSTRAINT hypertable_cagg_settings_pkey PRIMARY KEY (hypertable_id),
+  CONSTRAINT hypertable_cagg_settings_hypertable_id_fkey FOREIGN KEY (hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE
+);
+
+SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.hypertable_cagg_settings', '');
+
 /* the source of this data is the enum from the source code that lists
  *  the algorithms. This table is NOT dumped.
  */

sql/updates/latest-dev.sql

@@ -147,3 +147,28 @@ GRANT SELECT ON _timescaledb_catalog.chunk TO PUBLIC;
 -- END add chunk.relid
 --
 
+-- add continuous_aggs_tenant_tracking
+CREATE TABLE _timescaledb_catalog.continuous_aggs_tenant_tracking (
+  hypertable_id integer NOT NULL,
+  tenant_id text,
+  min_timestamp bigint NOT NULL,
+  max_timestamp bigint NOT NULL,
+  seqnum integer NOT NULL,
+  CONSTRAINT continuous_aggs_tenant_tracking_hypertable_id_fkey FOREIGN KEY (hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE
+);
+
+SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_aggs_tenant_tracking', '');
+
+CREATE INDEX continuous_aggs_tenant_tracking_idx ON _timescaledb_catalog.continuous_aggs_tenant_tracking (hypertable_id, seqnum);
+
+-- add hypertable_cagg_settings
+CREATE TABLE _timescaledb_catalog.hypertable_cagg_settings (
+  hypertable_id integer NOT NULL,
+  granular_refresh_column name NOT NULL,
+  granular_refresh_start_offset text,
+  granular_refresh_end_offset text,
+  CONSTRAINT hypertable_cagg_settings_pkey PRIMARY KEY (hypertable_id),
+  CONSTRAINT hypertable_cagg_settings_hypertable_id_fkey FOREIGN KEY (hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE
+);
+
+SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.hypertable_cagg_settings', '');

sql/updates/reverse-dev.sql

@@ -260,3 +260,10 @@ DROP FUNCTION IF EXISTS _timescaledb_functions.policy_compaction_check(JSONB);
 
 DROP FUNCTION IF EXISTS @extschema@.alter_job(job_id INTEGER, schedule_interval INTERVAL, max_runtime INTERVAL, max_retries INTEGER, retry_period INTERVAL, scheduled BOOL, config JSONB, next_start TIMESTAMPTZ, if_exists BOOL, check_config REGPROC, fixed_schedule BOOL, initial_start TIMESTAMPTZ, timezone TEXT, job_name TEXT, config_merge JSONB);
 
+-- drop continuous_aggs_tenant_tracking
+ALTER EXTENSION timescaledb DROP TABLE _timescaledb_catalog.continuous_aggs_tenant_tracking;
+DROP TABLE _timescaledb_catalog.continuous_aggs_tenant_tracking;
+
+-- drop hypertable_cagg_settings
+ALTER EXTENSION timescaledb DROP TABLE _timescaledb_catalog.hypertable_cagg_settings;
+DROP TABLE _timescaledb_catalog.hypertable_cagg_settings;

src/ts_catalog/CMakeLists.txt

@@ -7,7 +7,9 @@ set(SOURCES
     ${CMAKE_CURRENT_SOURCE_DIR}/compression_settings.c
     ${CMAKE_CURRENT_SOURCE_DIR}/continuous_agg.c
     ${CMAKE_CURRENT_SOURCE_DIR}/continuous_aggs_jobs_refresh_ranges.c
+    ${CMAKE_CURRENT_SOURCE_DIR}/continuous_aggs_tenant_tracking.c
     ${CMAKE_CURRENT_SOURCE_DIR}/continuous_aggs_watermark.c
+    ${CMAKE_CURRENT_SOURCE_DIR}/hypertable_cagg_settings.c
     ${CMAKE_CURRENT_SOURCE_DIR}/metadata.c
     ${CMAKE_CURRENT_SOURCE_DIR}/tablespace.c)
 target_sources(${PROJECT_NAME} PRIVATE ${SOURCES})

src/ts_catalog/catalog.c

@@ -93,6 +93,14 @@ static const TableInfoDef catalog_table_names[_MAX_CATALOG_TABLES + 1] = {
 		.schema_name = CATALOG_SCHEMA_NAME,
 		.table_name = CONTINUOUS_AGGS_JOBS_REFRESH_RANGES_TABLE_NAME,
 	},
+	[CONTINUOUS_AGGS_TENANT_TRACKING] = {
+		.schema_name = CATALOG_SCHEMA_NAME,
+		.table_name = CONTINUOUS_AGGS_TENANT_TRACKING_TABLE_NAME,
+	},
+	[HYPERTABLE_CAGG_SETTINGS] = {
+		.schema_name = CATALOG_SCHEMA_NAME,
+		.table_name = HYPERTABLE_CAGG_SETTINGS_TABLE_NAME,
+	},
 	[COMPRESSION_SETTINGS] = {
 		.schema_name = CATALOG_SCHEMA_NAME,
 		.table_name = COMPRESSION_SETTINGS_TABLE_NAME,
@@ -242,6 +250,18 @@ static const TableIndexDef catalog_table_index_definitions[_MAX_CATALOG_TABLES]
 			[CONTINUOUS_AGGS_JOBS_REFRESH_RANGES_IDX] = "continuous_aggs_jobs_refresh_ranges_idx",
 		},
 	},
+	[CONTINUOUS_AGGS_TENANT_TRACKING] = {
+		.length = _MAX_CONTINUOUS_AGGS_TENANT_TRACKING_INDEX,
+		.names = (char *[]) {
+			[CONTINUOUS_AGGS_TENANT_TRACKING_IDX] = "continuous_aggs_tenant_tracking_idx",
+		},
+	},
+	[HYPERTABLE_CAGG_SETTINGS] = {
+		.length = _MAX_HYPERTABLE_CAGG_SETTINGS_INDEX,
+		.names = (char *[]) {
+			[HYPERTABLE_CAGG_SETTINGS_PKEY] = "hypertable_cagg_settings_pkey",
+		},
+	},
 	[CONTINUOUS_AGGS_WATERMARK] = {
 		.length = _MAX_CONTINUOUS_AGGS_WATERMARK_INDEX,
 		.names = (char *[]) {

src/ts_catalog/catalog.h

@@ -49,6 +49,8 @@ typedef enum CatalogTable
 	CONTINUOUS_AGGS_MATERIALIZATION_INVALIDATION_LOG,
 	CONTINUOUS_AGGS_MATERIALIZATION_RANGES,
 	CONTINUOUS_AGGS_JOBS_REFRESH_RANGES,
+	CONTINUOUS_AGGS_TENANT_TRACKING,
+	HYPERTABLE_CAGG_SETTINGS,
 	COMPRESSION_SETTINGS,
 	COMPRESSION_CHUNK_SIZE,
 	CONTINUOUS_AGGS_BUCKET_FUNCTION,
@@ -1080,6 +1082,84 @@ enum
 	_MAX_CONTINUOUS_AGGS_JOBS_REFRESH_RANGES_INDEX,
 };
 
+/****** CONTINUOUS_AGGS_TENANT_TRACKING definitions */
+#define CONTINUOUS_AGGS_TENANT_TRACKING_TABLE_NAME "continuous_aggs_tenant_tracking"
+
+typedef enum Anum_continuous_aggs_tenant_tracking
+{
+	Anum_continuous_aggs_tenant_tracking_hypertable_id = 1,
+	Anum_continuous_aggs_tenant_tracking_tenant_id,
+	Anum_continuous_aggs_tenant_tracking_min_timestamp,
+	Anum_continuous_aggs_tenant_tracking_max_timestamp,
+	Anum_continuous_aggs_tenant_tracking_seqnum,
+	_Anum_continuous_aggs_tenant_tracking_max,
+} Anum_continuous_aggs_tenant_tracking;
+
+#define Natts_continuous_aggs_tenant_tracking (_Anum_continuous_aggs_tenant_tracking_max - 1)
+
+typedef struct FormData_continuous_aggs_tenant_tracking
+{
+	int32 hypertable_id;
+	text *tenant_id;
+	int64 min_timestamp;
+	int64 max_timestamp;
+	int32 seqnum;
+} FormData_continuous_aggs_tenant_tracking;
+
+typedef FormData_continuous_aggs_tenant_tracking *Form_continuous_aggs_tenant_tracking;
+
+enum
+{
+	CONTINUOUS_AGGS_TENANT_TRACKING_IDX = 0,
+	_MAX_CONTINUOUS_AGGS_TENANT_TRACKING_INDEX,
+};
+typedef enum Anum_continuous_aggs_tenant_tracking_idx
+{
+	Anum_continuous_aggs_tenant_tracking_idx_hypertable_id = 1,
+	Anum_continuous_aggs_tenant_tracking_idx_seqnum,
+	_Anum_continuous_aggs_tenant_tracking_idx_max,
+} Anum_continuous_aggs_tenant_tracking_idx;
+
+#define Natts_continuous_aggs_tenant_tracking_idx                                                  \
+	(_Anum_continuous_aggs_tenant_tracking_idx_max - 1)
+
+/****** HYPERTABLE_CAGG_SETTINGS definitions */
+#define HYPERTABLE_CAGG_SETTINGS_TABLE_NAME "hypertable_cagg_settings"
+
+typedef enum Anum_hypertable_cagg_settings
+{
+	Anum_hypertable_cagg_settings_hypertable_id = 1,
+	Anum_hypertable_cagg_settings_granular_refresh_column,
+	Anum_hypertable_cagg_settings_granular_refresh_start_offset,
+	Anum_hypertable_cagg_settings_granular_refresh_end_offset,
+	_Anum_hypertable_cagg_settings_max,
+} Anum_hypertable_cagg_settings;
+
+#define Natts_hypertable_cagg_settings (_Anum_hypertable_cagg_settings_max - 1)
+
+typedef struct FormData_hypertable_cagg_settings
+{
+	int32 hypertable_id;
+	NameData granular_refresh_column;
+	text *granular_refresh_start_offset;
+	text *granular_refresh_end_offset;
+} FormData_hypertable_cagg_settings;
+
+typedef FormData_hypertable_cagg_settings *Form_hypertable_cagg_settings;
+
+enum
+{
+	HYPERTABLE_CAGG_SETTINGS_PKEY = 0,
+	_MAX_HYPERTABLE_CAGG_SETTINGS_INDEX,
+};
+typedef enum Anum_hypertable_cagg_settings_pkey
+{
+	Anum_hypertable_cagg_settings_pkey_hypertable_id = 1,
+	_Anum_hypertable_cagg_settings_pkey_max,
+} Anum_hypertable_cagg_settings_pkey;
+
+#define Natts_hypertable_cagg_settings_pkey (_Anum_hypertable_cagg_settings_pkey_max - 1)
+
 typedef enum Anum_continuous_aggs_jobs_refresh_ranges_idx
 {
 	Anum_continuous_aggs_jobs_refresh_ranges_idx_materialization_id = 1,