Catalog updates to support granular refresh

.unreleased/pr_10357

@@ -0,0 +1 @@
+Implements: #10357 Updates on cagg and invalidations catalogs to support granular refresh

sql/pre_install/tables.sql

@@ -327,6 +327,7 @@ CREATE TABLE _timescaledb_catalog.continuous_agg (
   direct_view_name name NOT NULL,
   materialized_only bool NOT NULL DEFAULT FALSE,
   schema_change_timestamp bigint,
+  granular_refresh_enabled bool NOT NULL DEFAULT FALSE,
   -- table constraints
   CONSTRAINT continuous_agg_pkey PRIMARY KEY (mat_hypertable_id),
   CONSTRAINT continuous_agg_partial_view_schema_partial_view_name_key UNIQUE (partial_view_schema, partial_view_name),
@@ -391,7 +392,9 @@ SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_aggs
 CREATE TABLE _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log (
   hypertable_id integer NOT NULL,
   lowest_modified_value bigint NOT NULL,
-  greatest_modified_value bigint NOT NULL
+  greatest_modified_value bigint NOT NULL,
+  -- Granular sequence number. NULL when tenant tracking does not apply
+  seqnum integer
 );
 
 SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_aggs_hypertable_invalidation_log', '');
@@ -403,6 +406,8 @@ CREATE TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_l
   materialization_id integer,
   lowest_modified_value bigint NOT NULL,
   greatest_modified_value bigint NOT NULL,
+  -- Granular sequence number. NULL when tenant tracking does not apply
+  seqnum integer,
   -- table constraints
   CONSTRAINT continuous_aggs_materialization_invalid_materialization_id_fkey FOREIGN KEY (materialization_id) REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE
 );
@@ -415,9 +420,7 @@ CREATE INDEX continuous_aggs_materialization_invalidation_log_idx ON _timescaled
 CREATE TABLE _timescaledb_catalog.continuous_aggs_materialization_ranges (
   materialization_id integer,
   lowest_modified_value bigint NOT NULL,
-  greatest_modified_value bigint NOT NULL,
-  -- table constraints
-  CONSTRAINT continuous_aggs_materialization_ranges_materialization_id_fkey FOREIGN KEY (materialization_id) REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE
+  greatest_modified_value bigint NOT NULL
 );
 
 SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_aggs_materialization_ranges', '');

sql/updates/latest-dev.sql

@@ -0,0 +1,148 @@
+-- Rebuild the catalog table `_timescaledb_catalog.continuous_aggs_hypertable_invalidation_log`
+-- to add the `seqnum` column.
+CREATE TABLE _timescaledb_catalog._tmp_continuous_aggs_hypertable_invalidation_log AS
+    SELECT hypertable_id, lowest_modified_value, greatest_modified_value
+    FROM _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
+
+ALTER EXTENSION timescaledb
+    DROP TABLE _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
+DROP TABLE _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
+
+CREATE TABLE _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log (
+  hypertable_id integer NOT NULL,
+  lowest_modified_value bigint NOT NULL,
+  greatest_modified_value bigint NOT NULL,
+  seqnum integer
+);
+
+INSERT INTO _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log
+    (hypertable_id, lowest_modified_value, greatest_modified_value, seqnum)
+SELECT *, NULL::integer FROM _timescaledb_catalog._tmp_continuous_aggs_hypertable_invalidation_log;
+DROP TABLE _timescaledb_catalog._tmp_continuous_aggs_hypertable_invalidation_log;
+
+CREATE INDEX continuous_aggs_hypertable_invalidation_log_idx ON _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log (hypertable_id, lowest_modified_value ASC);
+
+SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_aggs_hypertable_invalidation_log', '');
+-- end rebuild _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log --
+
+-- Rebuild the catalog table `_timescaledb_catalog.continuous_aggs_materialization_invalidation_log`
+-- to add the `seqnum` column.
+CREATE TABLE _timescaledb_catalog._tmp_continuous_aggs_materialization_invalidation_log AS
+    SELECT materialization_id, lowest_modified_value, greatest_modified_value
+    FROM _timescaledb_catalog.continuous_aggs_materialization_invalidation_log;
+
+ALTER EXTENSION timescaledb
+    DROP TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log;
+DROP TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log;
+
+CREATE TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log (
+  materialization_id integer,
+  lowest_modified_value bigint NOT NULL,
+  greatest_modified_value bigint NOT NULL,
+  seqnum integer,
+  CONSTRAINT continuous_aggs_materialization_invalid_materialization_id_fkey FOREIGN KEY (materialization_id) REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE
+);
+
+INSERT INTO _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
+    (materialization_id, lowest_modified_value, greatest_modified_value, seqnum)
+SELECT *, NULL::integer FROM _timescaledb_catalog._tmp_continuous_aggs_materialization_invalidation_log;
+DROP TABLE _timescaledb_catalog._tmp_continuous_aggs_materialization_invalidation_log;
+
+CREATE INDEX continuous_aggs_materialization_invalidation_log_idx ON _timescaledb_catalog.continuous_aggs_materialization_invalidation_log (materialization_id, lowest_modified_value ASC);
+
+SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_aggs_materialization_invalidation_log', '');
+-- end rebuild _timescaledb_catalog.continuous_aggs_materialization_invalidation_log --
+
+-- Rebuild _timescaledb_catalog.continuous_agg to add granular_refresh_enabled.
+DROP VIEW IF EXISTS timescaledb_experimental.policies;
+
+ALTER TABLE _timescaledb_catalog.continuous_aggs_jobs_refresh_ranges
+    DROP CONSTRAINT continuous_aggs_jobs_refresh_ranges_materialization_id_fkey;
+ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
+    DROP CONSTRAINT continuous_aggs_materialization_invalid_materialization_id_fkey;
+ALTER TABLE _timescaledb_catalog.continuous_aggs_watermark
+    DROP CONSTRAINT continuous_aggs_watermark_mat_hypertable_id_fkey;
+-- We're dropping this FK and not recreating it since continuous_aggs_materialization_ranges
+-- is being deprecated and will be removed in a future release.
+ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_ranges
+    DROP CONSTRAINT continuous_aggs_materialization_ranges_materialization_id_fkey;
+
+ALTER EXTENSION timescaledb DROP TABLE _timescaledb_catalog.continuous_agg;
+
+CREATE TABLE _timescaledb_catalog._tmp_continuous_agg AS
+    SELECT
+        mat_hypertable_id,
+        raw_hypertable_id,
+        parent_mat_hypertable_id,
+        user_view_schema,
+        user_view_name,
+        partial_view_schema,
+        partial_view_name,
+        direct_view_schema,
+        direct_view_name,
+        materialized_only,
+        schema_change_timestamp
+    FROM
+        _timescaledb_catalog.continuous_agg
+    ORDER BY
+        mat_hypertable_id;
+
+DROP TABLE _timescaledb_catalog.continuous_agg;
+
+CREATE TABLE _timescaledb_catalog.continuous_agg (
+    mat_hypertable_id integer NOT NULL,
+    raw_hypertable_id integer NOT NULL,
+    parent_mat_hypertable_id integer,
+    user_view_schema name NOT NULL,
+    user_view_name name NOT NULL,
+    partial_view_schema name NOT NULL,
+    partial_view_name name NOT NULL,
+    direct_view_schema name NOT NULL,
+    direct_view_name name NOT NULL,
+    materialized_only bool NOT NULL DEFAULT FALSE,
+    schema_change_timestamp bigint,
+    granular_refresh_enabled bool NOT NULL DEFAULT FALSE,
+    CONSTRAINT continuous_agg_pkey PRIMARY KEY (mat_hypertable_id),
+    CONSTRAINT continuous_agg_partial_view_schema_partial_view_name_key UNIQUE (partial_view_schema, partial_view_name),
+    CONSTRAINT continuous_agg_user_view_schema_user_view_name_key UNIQUE (user_view_schema, user_view_name),
+    CONSTRAINT continuous_agg_mat_hypertable_id_fkey
+        FOREIGN KEY (mat_hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE,
+    CONSTRAINT continuous_agg_raw_hypertable_id_fkey
+        FOREIGN KEY (raw_hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE,
+    CONSTRAINT continuous_agg_parent_mat_hypertable_id_fkey
+        FOREIGN KEY (parent_mat_hypertable_id)
+        REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE
+);
+
+INSERT INTO _timescaledb_catalog.continuous_agg (
+    mat_hypertable_id,
+    raw_hypertable_id,
+    parent_mat_hypertable_id,
+    user_view_schema,
+    user_view_name,
+    partial_view_schema,
+    partial_view_name,
+    direct_view_schema,
+    direct_view_name,
+    materialized_only,
+    schema_change_timestamp
+)
+SELECT * FROM _timescaledb_catalog._tmp_continuous_agg;
+DROP TABLE _timescaledb_catalog._tmp_continuous_agg;
+
+CREATE INDEX continuous_agg_raw_hypertable_id_idx ON _timescaledb_catalog.continuous_agg (raw_hypertable_id);
+
+SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_agg', '');
+
+GRANT SELECT ON TABLE _timescaledb_catalog.continuous_agg TO PUBLIC;
+
+ALTER TABLE _timescaledb_catalog.continuous_aggs_jobs_refresh_ranges
+    ADD CONSTRAINT continuous_aggs_jobs_refresh_ranges_materialization_id_fkey
+        FOREIGN KEY (materialization_id) REFERENCES _timescaledb_catalog.continuous_agg(mat_hypertable_id) ON DELETE CASCADE;
+ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
+    ADD CONSTRAINT continuous_aggs_materialization_invalid_materialization_id_fkey
+        FOREIGN KEY (materialization_id) REFERENCES _timescaledb_catalog.continuous_agg(mat_hypertable_id) ON DELETE CASCADE;
+ALTER TABLE _timescaledb_catalog.continuous_aggs_watermark
+    ADD CONSTRAINT continuous_aggs_watermark_mat_hypertable_id_fkey
+        FOREIGN KEY (mat_hypertable_id) REFERENCES _timescaledb_catalog.continuous_agg(mat_hypertable_id) ON DELETE CASCADE;
+-- end rebuild _timescaledb_catalog.continuous_agg --

sql/updates/reverse-dev.sql

@@ -0,0 +1,143 @@
+-- Rebuild the catalog table `_timescaledb_catalog.continuous_aggs_hypertable_invalidation_log`
+-- to drop the `seqnum` column.
+CREATE TABLE _timescaledb_catalog._tmp_continuous_aggs_hypertable_invalidation_log AS
+    SELECT hypertable_id, lowest_modified_value, greatest_modified_value
+    FROM _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
+
+ALTER EXTENSION timescaledb
+    DROP TABLE _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
+DROP TABLE _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
+
+CREATE TABLE _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log (
+  hypertable_id integer NOT NULL,
+  lowest_modified_value bigint NOT NULL,
+  greatest_modified_value bigint NOT NULL
+);
+
+INSERT INTO _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log
+SELECT * FROM _timescaledb_catalog._tmp_continuous_aggs_hypertable_invalidation_log;
+DROP TABLE _timescaledb_catalog._tmp_continuous_aggs_hypertable_invalidation_log;
+
+CREATE INDEX continuous_aggs_hypertable_invalidation_log_idx ON _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log (hypertable_id, lowest_modified_value ASC);
+
+SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_aggs_hypertable_invalidation_log', '');
+-- end rebuild _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log --
+
+-- Rebuild the catalog table `_timescaledb_catalog.continuous_aggs_materialization_invalidation_log`
+-- to drop the `seqnum` column.
+CREATE TABLE _timescaledb_catalog._tmp_continuous_aggs_materialization_invalidation_log AS
+    SELECT materialization_id, lowest_modified_value, greatest_modified_value
+    FROM _timescaledb_catalog.continuous_aggs_materialization_invalidation_log;
+
+ALTER EXTENSION timescaledb
+    DROP TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log;
+DROP TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log;
+
+CREATE TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log (
+  materialization_id integer,
+  lowest_modified_value bigint NOT NULL,
+  greatest_modified_value bigint NOT NULL,
+  CONSTRAINT continuous_aggs_materialization_invalid_materialization_id_fkey FOREIGN KEY (materialization_id) REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE
+);
+
+INSERT INTO _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
+    (materialization_id, lowest_modified_value, greatest_modified_value)
+SELECT * FROM _timescaledb_catalog._tmp_continuous_aggs_materialization_invalidation_log;
+DROP TABLE _timescaledb_catalog._tmp_continuous_aggs_materialization_invalidation_log;
+
+CREATE INDEX continuous_aggs_materialization_invalidation_log_idx ON _timescaledb_catalog.continuous_aggs_materialization_invalidation_log (materialization_id, lowest_modified_value ASC);
+
+SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_aggs_materialization_invalidation_log', '');
+-- end rebuild _timescaledb_catalog.continuous_aggs_materialization_invalidation_log --
+
+-- Rebuild _timescaledb_catalog.continuous_agg to drop granular_refresh_enabled.
+DROP VIEW IF EXISTS timescaledb_experimental.policies;
+
+ALTER TABLE _timescaledb_catalog.continuous_aggs_jobs_refresh_ranges
+    DROP CONSTRAINT continuous_aggs_jobs_refresh_ranges_materialization_id_fkey;
+ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
+    DROP CONSTRAINT continuous_aggs_materialization_invalid_materialization_id_fkey;
+ALTER TABLE _timescaledb_catalog.continuous_aggs_watermark
+    DROP CONSTRAINT continuous_aggs_watermark_mat_hypertable_id_fkey;
+
+ALTER EXTENSION timescaledb DROP TABLE _timescaledb_catalog.continuous_agg;
+
+CREATE TABLE _timescaledb_catalog._tmp_continuous_agg AS
+    SELECT
+        mat_hypertable_id,
+        raw_hypertable_id,
+        parent_mat_hypertable_id,
+        user_view_schema,
+        user_view_name,
+        partial_view_schema,
+        partial_view_name,
+        direct_view_schema,
+        direct_view_name,
+        materialized_only,
+        schema_change_timestamp
+    FROM
+        _timescaledb_catalog.continuous_agg
+    ORDER BY
+        mat_hypertable_id;
+
+DROP TABLE _timescaledb_catalog.continuous_agg;
+
+CREATE TABLE _timescaledb_catalog.continuous_agg (
+    mat_hypertable_id integer NOT NULL,
+    raw_hypertable_id integer NOT NULL,
+    parent_mat_hypertable_id integer,
+    user_view_schema name NOT NULL,
+    user_view_name name NOT NULL,
+    partial_view_schema name NOT NULL,
+    partial_view_name name NOT NULL,
+    direct_view_schema name NOT NULL,
+    direct_view_name name NOT NULL,
+    materialized_only bool NOT NULL DEFAULT FALSE,
+    schema_change_timestamp bigint,
+    CONSTRAINT continuous_agg_pkey PRIMARY KEY (mat_hypertable_id),
+    CONSTRAINT continuous_agg_partial_view_schema_partial_view_name_key UNIQUE (partial_view_schema, partial_view_name),
+    CONSTRAINT continuous_agg_user_view_schema_user_view_name_key UNIQUE (user_view_schema, user_view_name),
+    CONSTRAINT continuous_agg_mat_hypertable_id_fkey
+        FOREIGN KEY (mat_hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE,
+    CONSTRAINT continuous_agg_raw_hypertable_id_fkey
+        FOREIGN KEY (raw_hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE,
+    CONSTRAINT continuous_agg_parent_mat_hypertable_id_fkey
+        FOREIGN KEY (parent_mat_hypertable_id)
+        REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE
+);
+
+INSERT INTO _timescaledb_catalog.continuous_agg (
+    mat_hypertable_id,
+    raw_hypertable_id,
+    parent_mat_hypertable_id,
+    user_view_schema,
+    user_view_name,
+    partial_view_schema,
+    partial_view_name,
+    direct_view_schema,
+    direct_view_name,
+    materialized_only,
+    schema_change_timestamp
+)
+SELECT * FROM _timescaledb_catalog._tmp_continuous_agg;
+DROP TABLE _timescaledb_catalog._tmp_continuous_agg;
+
+CREATE INDEX continuous_agg_raw_hypertable_id_idx ON _timescaledb_catalog.continuous_agg (raw_hypertable_id);
+
+SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_agg', '');
+
+GRANT SELECT ON TABLE _timescaledb_catalog.continuous_agg TO PUBLIC;
+
+ALTER TABLE _timescaledb_catalog.continuous_aggs_jobs_refresh_ranges
+    ADD CONSTRAINT continuous_aggs_jobs_refresh_ranges_materialization_id_fkey
+        FOREIGN KEY (materialization_id) REFERENCES _timescaledb_catalog.continuous_agg(mat_hypertable_id) ON DELETE CASCADE;
+ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
+    ADD CONSTRAINT continuous_aggs_materialization_invalid_materialization_id_fkey
+        FOREIGN KEY (materialization_id) REFERENCES _timescaledb_catalog.continuous_agg(mat_hypertable_id) ON DELETE CASCADE;
+ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_ranges
+    ADD CONSTRAINT continuous_aggs_materialization_ranges_materialization_id_fkey
+        FOREIGN KEY (materialization_id) REFERENCES _timescaledb_catalog.continuous_agg(mat_hypertable_id) ON DELETE CASCADE;
+ALTER TABLE _timescaledb_catalog.continuous_aggs_watermark
+    ADD CONSTRAINT continuous_aggs_watermark_mat_hypertable_id_fkey
+        FOREIGN KEY (mat_hypertable_id) REFERENCES _timescaledb_catalog.continuous_agg(mat_hypertable_id) ON DELETE CASCADE;
+-- end rebuild _timescaledb_catalog.continuous_agg --

src/ts_catalog/catalog.h

@@ -790,6 +790,7 @@ typedef enum Anum_continuous_agg
 	Anum_continuous_agg_direct_view_name,
 	Anum_continuous_agg_materialize_only,
 	Anum_continuous_agg_schema_change_timestamp,
+	Anum_continuous_agg_granular_refresh_enabled,
 	_Anum_continuous_agg_max,
 } Anum_continuous_agg;
 
@@ -808,6 +809,7 @@ typedef struct FormData_continuous_agg
 	NameData direct_view_name;
 	bool materialized_only;
 	int64 schema_change_timestamp;
+	bool granular_refresh_enabled;
 } FormData_continuous_agg;
 
 typedef FormData_continuous_agg *Form_continuous_agg;
@@ -904,6 +906,7 @@ typedef enum Anum_continuous_aggs_hypertable_invalidation_log
 	Anum_continuous_aggs_hypertable_invalidation_log_hypertable_id = 1,
 	Anum_continuous_aggs_hypertable_invalidation_log_lowest_modified_value,
 	Anum_continuous_aggs_hypertable_invalidation_log_greatest_modified_value,
+	Anum_continuous_aggs_hypertable_invalidation_log_seqnum,
 	_Anum_continuous_aggs_hypertable_invalidation_log_max,
 } Anum_continuous_aggs_hypertable_invalidation_log;
 
@@ -915,6 +918,7 @@ typedef struct FormData_continuous_aggs_hypertable_invalidation_log
 	int32 hypertable_id;
 	int64 lowest_modified_value;
 	int64 greatest_modified_value;
+	int32 seqnum;
 } FormData_continuous_aggs_hypertable_invalidation_log;
 
 typedef FormData_continuous_aggs_hypertable_invalidation_log
@@ -978,6 +982,7 @@ typedef enum Anum_continuous_aggs_materialization_invalidation_log
 	Anum_continuous_aggs_materialization_invalidation_log_materialization_id = 1,
 	Anum_continuous_aggs_materialization_invalidation_log_lowest_modified_value,
 	Anum_continuous_aggs_materialization_invalidation_log_greatest_modified_value,
+	Anum_continuous_aggs_materialization_invalidation_log_seqnum,
 	_Anum_continuous_aggs_materialization_invalidation_log_max,
 } Anum_continuous_aggs_materialization_invalidation_log;
 
@@ -989,6 +994,7 @@ typedef struct FormData_continuous_aggs_materialization_invalidation_log
 	int32 materialization_id;
 	int64 lowest_modified_value;
 	int64 greatest_modified_value;
+	int32 seqnum;
 } FormData_continuous_aggs_materialization_invalidation_log;
 
 typedef FormData_continuous_aggs_materialization_invalidation_log