From c4d4539066d962190511ed6343cca25788afc50e Mon Sep 17 00:00:00 2001 From: Melih Mutlu Date: Wed, 29 Jul 2026 12:03:22 +0300 Subject: [PATCH] Updates on cagg and invalidation catalogs Following updates on two catalogs to support granular refresh: - Add seqnum column to cagg invalidation logs to both continuous_aggs_hypertable_invalidation_log and continuous_aggs_materialization_invalidation_log so invalidation entries can be matched up with granular trackings. NULL means the entry has no associated granular tracking. - Add granular_refresh_enabled to continuous_agg Add a granular_refresh_enabled flag (default FALSE) to the continuous_agg catalog table, marking caggs that opt into granular refresh. Also, this updates upgrade/downgrade tests to cover seqnum in invalidation log verifications. Removes FK on continuous_aggs_materialization_ranges to continuous_agg to isolate continuous_aggs_materialization_ranges as it's deprecated. Co-authored-by: gayyappan --- .unreleased/pr_10357 | 1 + sql/pre_install/tables.sql | 11 +- sql/updates/latest-dev.sql | 148 +++++++++++++++++++ sql/updates/reverse-dev.sql | 143 ++++++++++++++++++ src/ts_catalog/catalog.h | 6 + src/ts_catalog/continuous_agg.c | 6 + test/sql/updates/post.continuous_aggs.sql | 79 ++++++---- test/sql/updates/setup.continuous_aggs.sql | 24 ++- tsl/src/continuous_aggs/invalidation.c | 163 +++++++++++---------- tsl/src/continuous_aggs/invalidation.h | 11 +- tsl/test/expected/cagg_bgw.out | 4 +- 11 files changed, 483 insertions(+), 113 deletions(-) create mode 100644 .unreleased/pr_10357 diff --git a/.unreleased/pr_10357 b/.unreleased/pr_10357 new file mode 100644 index 00000000000..d3591f664cd --- /dev/null +++ b/.unreleased/pr_10357 @@ -0,0 +1 @@ +Implements: #10357 Updates on cagg and invalidations catalogs to support granular refresh diff --git a/sql/pre_install/tables.sql b/sql/pre_install/tables.sql index de2257cee1d..eaf8f38911e 100644 --- a/sql/pre_install/tables.sql +++ b/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', ''); diff --git a/sql/updates/latest-dev.sql b/sql/updates/latest-dev.sql index e69de29bb2d..a3c95e7b1bb 100644 --- a/sql/updates/latest-dev.sql +++ b/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 -- diff --git a/sql/updates/reverse-dev.sql b/sql/updates/reverse-dev.sql index e69de29bb2d..4bbc0ca6a62 100644 --- a/sql/updates/reverse-dev.sql +++ b/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 -- diff --git a/src/ts_catalog/catalog.h b/src/ts_catalog/catalog.h index 5eb2dcdd5e2..b118c1326c6 100644 --- a/src/ts_catalog/catalog.h +++ b/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 diff --git a/src/ts_catalog/continuous_agg.c b/src/ts_catalog/continuous_agg.c index 19fed96a6e2..a3661d8495a 100644 --- a/src/ts_catalog/continuous_agg.c +++ b/src/ts_catalog/continuous_agg.c @@ -323,6 +323,9 @@ continuous_agg_formdata_make_tuple(const FormData_continuous_agg *fd, TupleDesc Int64GetDatum(fd->schema_change_timestamp); } + values[AttrNumberGetAttrOffset(Anum_continuous_agg_granular_refresh_enabled)] = + BoolGetDatum(fd->granular_refresh_enabled); + return heap_form_tuple(desc, values, nulls); } @@ -386,6 +389,9 @@ continuous_agg_formdata_fill(FormData_continuous_agg *fd, const TupleInfo *ti) values[AttrNumberGetAttrOffset(Anum_continuous_agg_schema_change_timestamp)]); } + fd->granular_refresh_enabled = + DatumGetBool(values[AttrNumberGetAttrOffset(Anum_continuous_agg_granular_refresh_enabled)]); + if (should_free) { heap_freetuple(tuple); diff --git a/test/sql/updates/post.continuous_aggs.sql b/test/sql/updates/post.continuous_aggs.sql index 0d59379c60f..300fa73f937 100644 --- a/test/sql/updates/post.continuous_aggs.sql +++ b/test/sql/updates/post.continuous_aggs.sql @@ -105,55 +105,78 @@ END $$ LANGUAGE PLPGSQL; -- Dump the invalidation log rows of the inval_log_test fixture so they --- are part of the baseline/updated/restored comparison. -SELECT h.table_name AS hypertable, - l.lowest_modified_value, l.greatest_modified_value +-- are part of the baseline/updated/restored comparison. The seqnum +-- column only exists from 2.30; this runs on the post-update version, +-- which is the same for all three databases, so the branch taken is +-- identical and the outputs stay comparable. +SELECT EXISTS ( + SELECT FROM information_schema.columns + WHERE table_schema = '_timescaledb_catalog' + AND table_name = 'continuous_aggs_hypertable_invalidation_log' + AND column_name = 'seqnum') AS has_inval_log_seqnum \gset + +-- Collect the live rows in the same shape as inval_log_snapshot so both +-- the dumps below and the verification compare like with like. This is +-- the only place that has to know whether seqnum exists. +CREATE TEMP VIEW inval_log_live AS +SELECT 'hypertable'::text AS log, h.table_name AS name, + l.lowest_modified_value, l.greatest_modified_value, +\if :has_inval_log_seqnum + l.seqnum +\else + NULL::integer AS seqnum +\endif FROM _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log l JOIN _timescaledb_catalog.hypertable h ON h.id = l.hypertable_id WHERE h.table_name = 'inval_log_test' -ORDER BY 1, 2, 3; - -SELECT ca.user_view_name AS cagg, - l.lowest_modified_value, l.greatest_modified_value +UNION ALL +SELECT 'materialization', ca.user_view_name, + l.lowest_modified_value, l.greatest_modified_value, +\if :has_inval_log_seqnum + l.seqnum +\else + NULL::integer +\endif FROM _timescaledb_catalog.continuous_aggs_materialization_invalidation_log l JOIN _timescaledb_catalog.continuous_agg ca ON ca.mat_hypertable_id = l.materialization_id -WHERE ca.user_view_name IN ('mat_invallog_1', 'mat_invallog_2') -ORDER BY 1, 2, 3; +WHERE ca.user_view_name IN ('mat_invallog_1', 'mat_invallog_2'); + +SELECT name AS hypertable, lowest_modified_value, greatest_modified_value, seqnum +FROM inval_log_live +WHERE log = 'hypertable' +ORDER BY 1, 2, 3, 4; + +SELECT name AS cagg, lowest_modified_value, greatest_modified_value, seqnum +FROM inval_log_live +WHERE log = 'materialization' +ORDER BY 1, 2, 3, 4; -- Verify the live invalidation logs still hold exactly the rows -- snapshotted at the end of setup, i.e. an update script that rebuilds --- the log catalogs neither lost nor invented rows. +-- the log catalogs neither lost nor invented rows. seqnum is part of the +-- comparison: EXCEPT ALL compares rows with NULL treated as equal to +-- NULL, so on versions without the column both sides are NULL and only +-- the other values decide, while from 2.30 on a changed seqnum shows up +-- as a row that is both missing and unexpected. DO $$ DECLARE difference TEXT; BEGIN - WITH live (log, name, lowest_modified_value, greatest_modified_value) AS ( - SELECT 'hypertable'::text, h.table_name, - l.lowest_modified_value, l.greatest_modified_value - FROM _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log l - JOIN _timescaledb_catalog.hypertable h ON h.id = l.hypertable_id - WHERE h.table_name = 'inval_log_test' - UNION ALL - SELECT 'materialization', ca.user_view_name, - l.lowest_modified_value, l.greatest_modified_value - FROM _timescaledb_catalog.continuous_aggs_materialization_invalidation_log l - JOIN _timescaledb_catalog.continuous_agg ca ON ca.mat_hypertable_id = l.materialization_id - WHERE ca.user_view_name IN ('mat_invallog_1', 'mat_invallog_2') - ) - SELECT string_agg(format('%s [%s]: (%s, %s, %s)', src, diff.log, diff.name, - diff.lowest_modified_value, diff.greatest_modified_value), E'\n') + SELECT string_agg(format('%s [%s]: (%s, %s, %s, seqnum %s)', src, diff.log, diff.name, + diff.lowest_modified_value, diff.greatest_modified_value, + coalesce(diff.seqnum::text, 'NULL')), E'\n') INTO difference FROM ( SELECT 'missing after update' AS src, * FROM (SELECT * FROM inval_log_snapshot EXCEPT ALL - SELECT * FROM live) missing + SELECT * FROM inval_log_live) missing UNION ALL SELECT 'unexpected after update', * - FROM (SELECT * FROM live + FROM (SELECT * FROM inval_log_live EXCEPT ALL SELECT * FROM inval_log_snapshot) unexpected - ) diff (src, log, name, lowest_modified_value, greatest_modified_value); + ) diff (src, log, name, lowest_modified_value, greatest_modified_value, seqnum); IF difference IS NOT NULL THEN RAISE EXCEPTION 'invalidation log content changed across the update' diff --git a/test/sql/updates/setup.continuous_aggs.sql b/test/sql/updates/setup.continuous_aggs.sql index fc1deddccd5..9eb048c686b 100644 --- a/test/sql/updates/setup.continuous_aggs.sql +++ b/test/sql/updates/setup.continuous_aggs.sql @@ -353,16 +353,34 @@ FROM generate_series('2020-01-07 07:00:00+00'::timestamptz, -- Snapshot invalidation log rows into a plain table so the post script can -- verify the live logs still hold the same content after the update rebuilt --- the catalogs. +-- the catalogs. The seqnum column only exists from 2.30; on older versions +-- record NULL, which is also what the 2.30 migration must leave in migrated +-- rows and what the trigger writes when no granular tracking applies. +SELECT EXISTS ( + SELECT FROM information_schema.columns + WHERE table_schema = '_timescaledb_catalog' + AND table_name = 'continuous_aggs_hypertable_invalidation_log' + AND column_name = 'seqnum') AS has_inval_log_seqnum \gset + CREATE TABLE inval_log_snapshot AS SELECT 'hypertable'::text AS log, h.table_name AS name, - l.lowest_modified_value, l.greatest_modified_value + l.lowest_modified_value, l.greatest_modified_value, +\if :has_inval_log_seqnum + l.seqnum +\else + NULL::integer AS seqnum +\endif FROM _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log l JOIN _timescaledb_catalog.hypertable h ON h.id = l.hypertable_id WHERE h.table_name = 'inval_log_test' UNION ALL SELECT 'materialization', ca.user_view_name, - l.lowest_modified_value, l.greatest_modified_value + l.lowest_modified_value, l.greatest_modified_value, +\if :has_inval_log_seqnum + l.seqnum +\else + NULL::integer +\endif FROM _timescaledb_catalog.continuous_aggs_materialization_invalidation_log l JOIN _timescaledb_catalog.continuous_agg ca ON ca.mat_hypertable_id = l.materialization_id WHERE ca.user_view_name IN ('mat_invallog_1', 'mat_invallog_2'); diff --git a/tsl/src/continuous_aggs/invalidation.c b/tsl/src/continuous_aggs/invalidation.c index afa1e4b7e40..384b1cdc67b 100644 --- a/tsl/src/continuous_aggs/invalidation.c +++ b/tsl/src/continuous_aggs/invalidation.c @@ -122,9 +122,9 @@ typedef enum ContinuousAggTableType static Relation open_cagg_table(ContinuousAggTableType type, LOCKMODE lockmode); static void hypertable_invalidation_scan_init(ScanIterator *iterator, int32 hyper_id, LOCKMODE lockmode); -static bool insert_invalidation_entry(Relation cagg_log_rel, const Invalidation *invalidation); -static bool write_invalidation_entry(Relation cagg_log_rel, const Invalidation *invalidation, - ItemPointer update_tid); +static bool insert_cagg_invalidation_entry(Relation cagg_log_rel, const Invalidation *invalidation); +static bool write_cagg_invalidation_entry(Relation cagg_log_rel, const Invalidation *invalidation, + ItemPointer update_tid); static void invalidation_entry_set(Invalidation *inner_range, int32 hyper_id, int64 lowest_modified_value, int64 greatest_modified_value); static void invalidation_entry_reset(Invalidation *entry); @@ -137,8 +137,6 @@ invalidation_entry_set_from_cagg_invalidation(Invalidation *entry, const TupleIn const ContinuousAggBucketFunction *bucket_function); static bool invalidations_can_be_merged(const Invalidation *a, const Invalidation *b); static bool invalidation_entry_try_merge(Invalidation *entry, const Invalidation *newentry); -static void insert_new_cagg_invalidation(const HypertableInvalidationState *state, - const Invalidation *entry, int32 cagg_hyper_id); static void move_invalidations_from_hyper_to_cagg_log(const HypertableInvalidationState *state); static void cagg_invalidations_scan_by_hypertable_init(ScanIterator *iterator, int32 cagg_hyper_id, LOCKMODE lockmode, int64 window_end); @@ -186,7 +184,8 @@ hypertable_invalidation_scan_init(ScanIterator *iterator, int32 hyper_id, LOCKMO } HeapTuple -create_invalidation_tup(const TupleDesc tupdesc, int32 cagg_hyper_id, int64 start, int64 end) +create_cagg_invalidation_tup(const TupleDesc tupdesc, int32 cagg_hyper_id, int64 start, int64 end, + int32 seqnum) { Datum values[Natts_continuous_aggs_materialization_invalidation_log] = { 0 }; bool isnull[Natts_continuous_aggs_materialization_invalidation_log] = { false }; @@ -200,6 +199,17 @@ create_invalidation_tup(const TupleDesc tupdesc, int32 cagg_hyper_id, int64 star values[AttrNumberGetAttrOffset( Anum_continuous_aggs_materialization_invalidation_log_greatest_modified_value)] = Int64GetDatum(end); + /* seqnum 0 means "no associated granular tracking" and is stored as SQL NULL. */ + if (seqnum == 0) + { + isnull[AttrNumberGetAttrOffset( + Anum_continuous_aggs_materialization_invalidation_log_seqnum)] = true; + } + else + { + values[AttrNumberGetAttrOffset( + Anum_continuous_aggs_materialization_invalidation_log_seqnum)] = Int32GetDatum(seqnum); + } return heap_form_tuple(tupdesc, values, isnull); } @@ -215,7 +225,7 @@ invalidation_cagg_log_add_entry(int32 cagg_hyper_id, int64 start, int64 end) HeapTuple tuple; Assert(start <= end); - tuple = create_invalidation_tup(RelationGetDescr(rel), cagg_hyper_id, start, end); + tuple = create_cagg_invalidation_tup(RelationGetDescr(rel), cagg_hyper_id, start, end, 0); ts_catalog_database_info_become_owner(ts_catalog_database_info_get(), &sec_ctx); ts_catalog_insert_only(rel, tuple); ts_catalog_restore_user(&sec_ctx); @@ -240,6 +250,8 @@ invalidation_hyper_log_add_entry(int32 hyper_id, int64 start, int64 end) values[AttrNumberGetAttrOffset( Anum_continuous_aggs_hypertable_invalidation_log_greatest_modified_value)] = Int64GetDatum(end); + /* seqnum is stamped only when tenant tracking applies; leave it NULL here. */ + nulls[AttrNumberGetAttrOffset(Anum_continuous_aggs_hypertable_invalidation_log_seqnum)] = true; ts_catalog_database_info_become_owner(ts_catalog_database_info_get(), &sec_ctx); ts_catalog_insert_values(rel, RelationGetDescr(rel), values, nulls); @@ -298,8 +310,8 @@ IsValidInvalidation(const Invalidation *invalidation) } static bool -write_invalidation_entry(Relation cagg_log_rel, const Invalidation *invalidation, - ItemPointer update_tid) +write_cagg_invalidation_entry(Relation cagg_log_rel, const Invalidation *invalidation, + ItemPointer update_tid) { CatalogSecurityContext sec_ctx; HeapTuple tup; @@ -309,10 +321,11 @@ write_invalidation_entry(Relation cagg_log_rel, const Invalidation *invalidation return false; } - tup = create_invalidation_tup(RelationGetDescr(cagg_log_rel), - invalidation->hyper_id, - invalidation->lowest_modified_value, - invalidation->greatest_modified_value); + tup = create_cagg_invalidation_tup(RelationGetDescr(cagg_log_rel), + invalidation->hyper_id, + invalidation->lowest_modified_value, + invalidation->greatest_modified_value, + invalidation->seqnum); ts_catalog_database_info_become_owner(ts_catalog_database_info_get(), &sec_ctx); if (ItemPointerIsValid(update_tid)) { @@ -329,9 +342,9 @@ write_invalidation_entry(Relation cagg_log_rel, const Invalidation *invalidation } static bool -insert_invalidation_entry(Relation cagg_log_rel, const Invalidation *invalidation) +insert_cagg_invalidation_entry(Relation cagg_log_rel, const Invalidation *invalidation) { - return write_invalidation_entry(cagg_log_rel, invalidation, NULL); + return write_cagg_invalidation_entry(cagg_log_rel, invalidation, NULL); } static void @@ -453,17 +466,17 @@ cut_invalidation_along_refresh_window(const ContinuousAggInvalidationState *stat if (IsValidInvalidation(&lower_range)) { - write_invalidation_entry(state->cagg_log_rel, - &lower_range, - ItemPointerIsValid(&tid) ? &tid : NULL); + write_cagg_invalidation_entry(state->cagg_log_rel, + &lower_range, + ItemPointerIsValid(&tid) ? &tid : NULL); ItemPointerSetInvalid(&tid); /* TID consumed — upper must be a fresh insert */ } if (IsValidInvalidation(&upper_range)) { - write_invalidation_entry(state->cagg_log_rel, - &upper_range, - ItemPointerIsValid(&tid) ? &tid : NULL); + write_cagg_invalidation_entry(state->cagg_log_rel, + &upper_range, + ItemPointerIsValid(&tid) ? &tid : NULL); } } @@ -577,36 +590,56 @@ invalidation_expand_to_bucket_boundaries(Invalidation *inv, Oid time_type_oid, } /* - * Macro to set an Invalidation from a tuple. The tuple can either have the - * format of the hypertable invalidation log or the continuous aggregate - * invalidation log (as determined by the type parameter). + * Set an Invalidation from a tuple of either the hypertable invalidation log or + * the continuous aggregate invalidation log. + * + * Both logs share the same physical column layout -- (id, lowest_modified_value, + * greatest_modified_value, seqnum) -- so the same attribute offsets apply to + * both; we use the hypertable-log Anums here for both. The tuple is read with + * heap_deform_tuple() rather than GETSTRUCT() because seqnum is nullable. seqnum + * is mapped to 0 (untracked) when SQL NULL. */ -#define INVALIDATION_ENTRY_SET(entry, ti, hypertable_id, type) \ - do \ - { \ - bool should_free; \ - HeapTuple tuple = ts_scanner_fetch_heap_tuple(ti, false, &should_free); \ - type form; \ - form = (type) GETSTRUCT(tuple); \ - (entry)->hyper_id = form->hypertable_id; \ - (entry)->lowest_modified_value = form->lowest_modified_value; \ - (entry)->greatest_modified_value = form->greatest_modified_value; \ - (entry)->is_modified = false; \ - ItemPointerCopy(&tuple->t_self, &(entry)->tid); \ - \ - if (should_free) \ - heap_freetuple(tuple); \ - } while (0); +static void +invalidation_entry_set_from_tuple(Invalidation *entry, const TupleInfo *ti) +{ + bool should_free; + HeapTuple tuple = ts_scanner_fetch_heap_tuple(ti, false, &should_free); + Datum values[Natts_continuous_aggs_hypertable_invalidation_log]; + bool nulls[Natts_continuous_aggs_hypertable_invalidation_log] = { false }; + + heap_deform_tuple(tuple, ts_scanner_get_tupledesc(ti), values, nulls); + + entry->hyper_id = DatumGetInt32(values[AttrNumberGetAttrOffset( + Anum_continuous_aggs_hypertable_invalidation_log_hypertable_id)]); + entry->lowest_modified_value = DatumGetInt64(values[AttrNumberGetAttrOffset( + Anum_continuous_aggs_hypertable_invalidation_log_lowest_modified_value)]); + entry->greatest_modified_value = DatumGetInt64(values[AttrNumberGetAttrOffset( + Anum_continuous_aggs_hypertable_invalidation_log_greatest_modified_value)]); + /* seqnum is nullable: SQL NULL means untracked, mapped to 0. */ + if (nulls[AttrNumberGetAttrOffset(Anum_continuous_aggs_hypertable_invalidation_log_seqnum)]) + { + entry->seqnum = 0; + } + else + { + entry->seqnum = DatumGetInt32(values[AttrNumberGetAttrOffset( + Anum_continuous_aggs_hypertable_invalidation_log_seqnum)]); + } + entry->is_modified = false; + ItemPointerCopy(&tuple->t_self, &entry->tid); + + if (should_free) + { + heap_freetuple(tuple); + } +} static void invalidation_entry_set_from_hyper_invalidation(Invalidation *entry, const TupleInfo *ti, int32 hyper_id, Oid dimtype, const ContinuousAggBucketFunction *bucket_function) { - INVALIDATION_ENTRY_SET(entry, - ti, - hypertable_id, - Form_continuous_aggs_hypertable_invalidation_log); + invalidation_entry_set_from_tuple(entry, ti); /* Since hypertable invalidations are moved to the continuous aggregate * invalidation log, a different hypertable ID must be set (the ID of the * materialized hypertable). */ @@ -618,10 +651,7 @@ static void invalidation_entry_set_from_cagg_invalidation(Invalidation *entry, const TupleInfo *ti, Oid dimtype, const ContinuousAggBucketFunction *bucket_function) { - INVALIDATION_ENTRY_SET(entry, - ti, - materialization_id, - Form_continuous_aggs_materialization_invalidation_log); + invalidation_entry_set_from_tuple(entry, ti); /* It isn't strictly necessary to expand the invalidation to bucket * boundaries here since all invalidations were already expanded when @@ -696,23 +726,6 @@ invalidation_entry_try_merge(Invalidation *entry, const Invalidation *newentry) return true; } -static void -insert_new_cagg_invalidation(const HypertableInvalidationState *state, const Invalidation *entry, - int32 cagg_hyper_id) -{ - CatalogSecurityContext sec_ctx; - TupleDesc tupdesc = RelationGetDescr(state->cagg_log_rel); - HeapTuple tuple = create_invalidation_tup(tupdesc, - cagg_hyper_id, - entry->lowest_modified_value, - entry->greatest_modified_value); - - ts_catalog_database_info_become_owner(ts_catalog_database_info_get(), &sec_ctx); - ts_catalog_insert_only(state->cagg_log_rel, tuple); - ts_catalog_restore_user(&sec_ctx); - heap_freetuple(tuple); -} - /* * Process invalidations in the hypertable invalidation log. * @@ -782,7 +795,7 @@ move_invalidations_from_hyper_to_cagg_log(const HypertableInvalidationState *sta } else if (!invalidation_entry_try_merge(&mergedentry, &logentry)) { - insert_new_cagg_invalidation(state, &mergedentry, cagg_hyper_id); + insert_cagg_invalidation_entry(state->cagg_log_rel, &mergedentry); mergedentry = logentry; } @@ -807,7 +820,7 @@ move_invalidations_from_hyper_to_cagg_log(const HypertableInvalidationState *sta /* Handle the last merged invalidation */ if (IsValidInvalidation(&mergedentry)) { - insert_new_cagg_invalidation(state, &mergedentry, cagg_hyper_id); + insert_cagg_invalidation_entry(state->cagg_log_rel, &mergedentry); } } } @@ -864,7 +877,7 @@ cut_cagg_invalidation(const ContinuousAggInvalidationState *state, * entry, update it to reflect the expanded range. */ if (entry->is_modified) { - write_invalidation_entry(state->cagg_log_rel, entry, &tid); + write_cagg_invalidation_entry(state->cagg_log_rel, entry, &tid); } break; case INVAL_CUT: @@ -897,7 +910,7 @@ cut_cagg_invalidation_and_compute_inner_range(const ContinuousAggInvalidationSta else if (IsValidInvalidation(&new_inner_range) && !invalidation_entry_try_merge(&inner_range, &new_inner_range)) { - insert_invalidation_entry(state->cagg_log_rel, &inner_range); + insert_cagg_invalidation_entry(state->cagg_log_rel, &inner_range); inner_range = new_inner_range; } @@ -1035,7 +1048,7 @@ process_cagg_invalidations_for_refresh(const ContinuousAggInvalidationState *sta } /* Write the last (merged) inner range back to the cagg invalidation log */ - insert_invalidation_entry(state->cagg_log_rel, &inner_range); + insert_cagg_invalidation_entry(state->cagg_log_rel, &inner_range); } static void @@ -1171,10 +1184,12 @@ collect_and_delete_cagg_invalidations_in_window(const ContinuousAgg *cagg, table_close(rel, AccessShareLock); int64 inclusive_end = ts_time_saturating_sub(refresh_window->end, 1, refresh_window->type); - HeapTuple forced_tuple = create_invalidation_tup(tupdesc, - cagg->data.mat_hypertable_id, - refresh_window->start, - inclusive_end); + /* Forced refresh is not tenant-tracked: seqnum 0 -> NULL. */ + HeapTuple forced_tuple = create_cagg_invalidation_tup(tupdesc, + cagg->data.mat_hypertable_id, + refresh_window->start, + inclusive_end, + 0); tupstore = tuplestore_begin_heap(false, false, work_mem); tuplestore_puttuple(tupstore, forced_tuple); heap_freetuple(forced_tuple); diff --git a/tsl/src/continuous_aggs/invalidation.h b/tsl/src/continuous_aggs/invalidation.h index 129560bd24c..41532447cea 100644 --- a/tsl/src/continuous_aggs/invalidation.h +++ b/tsl/src/continuous_aggs/invalidation.h @@ -23,6 +23,13 @@ typedef struct Invalidation int64 greatest_modified_value; bool is_modified; ItemPointerData tid; + /* + * Per-hypertable granular-tracking sequence number, used to match up with granular + * tracking in the continuous aggregate invalidation log. + * Stored as a nullable column on disk but loaded as 0 here for null, meaning + * "no associated granular tracking". + */ + int32 seqnum; } Invalidation; #define INVAL_NEG_INFINITY PG_INT64_MIN @@ -53,8 +60,8 @@ extern void invalidation_store_free(InvalidationStore *store); extern void invalidation_expand_to_bucket_boundaries(Invalidation *inv, Oid time_type_oid, const ContinuousAggBucketFunction *bucket_function); -extern HeapTuple create_invalidation_tup(const TupleDesc tupdesc, int32 cagg_hyper_id, int64 start, - int64 end); +extern HeapTuple create_cagg_invalidation_tup(const TupleDesc tupdesc, int32 cagg_hyper_id, + int64 start, int64 end, int32 seqnum); extern bool invalidation_hypertable_has_invalidations(int32 hyper_id); extern bool invalidation_cagg_has_invalidations(ContinuousAgg *cagg); extern bool invalidation_cagg_has_pending_mat_ranges(ContinuousAgg *cagg); diff --git a/tsl/test/expected/cagg_bgw.out b/tsl/test/expected/cagg_bgw.out index c5e40abf013..3c68492f656 100644 --- a/tsl/test/expected/cagg_bgw.out +++ b/tsl/test/expected/cagg_bgw.out @@ -64,8 +64,8 @@ SELECT * FROM timescaledb_information.job_stats; -------------------+-----------------+--------+---------------------+------------------------+-----------------+------------+-------------------+------------+------------+-----------------+---------------- SELECT * FROM _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 --------------------+-------------------+--------------------------+------------------+----------------+---------------------+-------------------+--------------------+------------------+-------------------+------------------------- + 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 | granular_refresh_enabled +-------------------+-------------------+--------------------------+------------------+----------------+---------------------+-------------------+--------------------+------------------+-------------------+-------------------------+-------------------------- -- though user on access node has required GRANTS, this will propagate GRANTS to the connected data nodes GRANT CREATE ON SCHEMA public TO :ROLE_DEFAULT_PERM_USER;