Support ADD COLUMN on continuous aggregates

.unreleased/pr_9825

@@ -0,0 +1 @@
+Implements: #9825 Support ADD COLUMN on continuous aggregates

sql/pre_install/tables.sql

@@ -340,6 +340,7 @@ CREATE TABLE _timescaledb_catalog.continuous_agg (
   direct_view_schema name NOT NULL,
   direct_view_name name NOT NULL,
   materialized_only bool NOT NULL DEFAULT FALSE,
+  schema_change_timestamp bigint,
   -- 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),

sql/updates/latest-dev.sql

@@ -181,3 +181,98 @@ DROP FUNCTION IF EXISTS _timescaledb_functions.calculate_chunk_interval(integer,
 -- chunk_target_size is no longer used, so its check constraint is dropped.
 ALTER TABLE _timescaledb_catalog.hypertable
     DROP CONSTRAINT IF EXISTS hypertable_chunk_target_size_check;
+
+-- Rebuild the catalog table `_timescaledb_catalog.continuous_agg` to add the
+-- `schema_change_timestamp` column.
+
+-- Drop views and foreign keys that depend on the catalog table.
+DROP VIEW IF EXISTS timescaledb_experimental.policies;
+DROP VIEW IF EXISTS timescaledb_information.hypertables;
+DROP VIEW IF EXISTS timescaledb_information.continuous_aggregates;
+DROP VIEW IF EXISTS timescaledb_information.jobs;
+
+ALTER TABLE _timescaledb_catalog.continuous_aggs_watermark
+    DROP CONSTRAINT continuous_aggs_watermark_mat_hypertable_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_materialization_ranges
+    DROP CONSTRAINT continuous_aggs_materialization_ranges_materialization_id_fkey;
+ALTER TABLE _timescaledb_catalog.continuous_aggs_jobs_refresh_ranges
+    DROP CONSTRAINT continuous_aggs_jobs_refresh_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
+    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,
+  -- 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),
+  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)
+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_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;
+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_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;
+
+ANALYZE _timescaledb_catalog.continuous_agg;
+-- end rebuild _timescaledb_catalog.continuous_agg --

sql/updates/reverse-dev.sql

@@ -189,3 +189,93 @@ DROP FUNCTION IF EXISTS @extschema@.create_hypertable(relation REGCLASS, time_co
 ALTER TABLE _timescaledb_catalog.hypertable
     ADD CONSTRAINT hypertable_chunk_target_size_check CHECK (chunk_target_size >= 0);
 DROP FUNCTION IF EXISTS _timescaledb_functions.rebuild_sparse_index(REGCLASS, BOOLEAN);
+
+-- Rebuild the catalog table `_timescaledb_catalog.continuous_agg` to drop the
+-- `schema_change_timestamp` column.
+DROP VIEW IF EXISTS timescaledb_experimental.policies;
+DROP VIEW IF EXISTS timescaledb_information.hypertables;
+DROP VIEW IF EXISTS timescaledb_information.continuous_aggregates;
+DROP VIEW IF EXISTS timescaledb_information.jobs;
+
+ALTER TABLE _timescaledb_catalog.continuous_aggs_watermark
+    DROP CONSTRAINT continuous_aggs_watermark_mat_hypertable_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_materialization_ranges
+    DROP CONSTRAINT continuous_aggs_materialization_ranges_materialization_id_fkey;
+ALTER TABLE _timescaledb_catalog.continuous_aggs_jobs_refresh_ranges
+    DROP CONSTRAINT continuous_aggs_jobs_refresh_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
+    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,
+  -- 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),
+  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
+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_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;
+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_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;
+
+ANALYZE _timescaledb_catalog.continuous_agg;
+-- end rebuild _timescaledb_catalog.continuous_agg --
+

src/cross_module_fn.c

@@ -230,6 +230,13 @@ continuous_agg_update_options_default(ContinuousAgg *cagg, WithClauseResult *wit
 	pg_unreachable();
 }
 
+static void
+continuous_agg_add_column_default(ContinuousAgg *cagg, AlterTableStmt *stmt)
+{
+	error_no_default_fn_community();
+	pg_unreachable();
+}
+
 static void
 continuous_agg_invalidate_raw_ht_all_default(const Hypertable *raw_ht, int64 start, int64 end)
 {
@@ -338,6 +345,7 @@ TSDLLEXPORT CrossModuleFunctions ts_cm_functions_default = {
 	.continuous_agg_invalidate_mat_ht = continuous_agg_invalidate_mat_ht_all_default,
 	.continuous_agg_dml_invalidate = continuous_agg_dml_invalidate_default,
 	.continuous_agg_update_options = continuous_agg_update_options_default,
+	.continuous_agg_add_column = continuous_agg_add_column_default,
 	.continuous_agg_apply_rewrites_tsl = NULL,
 	.continuous_agg_validate_query = error_no_default_fn_pg_community,
 	.continuous_agg_get_bucket_function = error_no_default_fn_pg_community,

src/cross_module_fn.h

@@ -105,6 +105,7 @@ typedef struct CrossModuleFunctions
 										  bool update);
 	void (*continuous_agg_update_options)(ContinuousAgg *cagg,
 										  WithClauseResult *with_clause_options);
+	void (*continuous_agg_add_column)(ContinuousAgg *cagg, AlterTableStmt *stmt);
 	Query *(*continuous_agg_apply_rewrites_tsl)(Query *parse);
 	PGFunction continuous_agg_validate_query;
 	PGFunction continuous_agg_get_bucket_function;

src/process_utility.c

@@ -5072,6 +5072,40 @@ process_altertable_start_matview(ProcessUtilityArgs *args)
 
 	continuous_agg_with_clause_perm_check(cagg, view_relid);
 
+	/*
+	 * CAgg add column handler.
+	 * Split `stmt->cmds` into the GENERATED ALWAYS AS and everything else
+	 * (handled by the switch below). Process all ADD COLUMNs at once since
+	 * they require a rewrite of the view, and doing them one by one would
+	 * be inefficient.
+	 */
+	{
+		List *addcol_cmds = NIL;
+		List *other_cmds = NIL;
+		foreach (lc, stmt->cmds)
+		{
+			AlterTableCmd *cmd = (AlterTableCmd *) lfirst(lc);
+			if (cmd->subtype == AT_AddColumn)
+			{
+				addcol_cmds = lappend(addcol_cmds, cmd);
+			}
+			else
+			{
+				other_cmds = lappend(other_cmds, cmd);
+			}
+		}
+
+		if (addcol_cmds != NIL)
+		{
+			AlterTableStmt addcol_stmt = *stmt;
+			addcol_stmt.cmds = addcol_cmds;
+			ts_cm_functions->continuous_agg_add_column(cagg, &addcol_stmt);
+			CommandCounterIncrement();
+
+			stmt->cmds = other_cmds;
+		}
+	}
+
 	foreach (lc, stmt->cmds)
 	{
 		AlterTableCmd *cmd = (AlterTableCmd *) lfirst(lc);

src/ts_catalog/catalog.h

@@ -815,6 +815,7 @@ typedef enum Anum_continuous_agg
 	Anum_continuous_agg_direct_view_schema,
 	Anum_continuous_agg_direct_view_name,
 	Anum_continuous_agg_materialize_only,
+	Anum_continuous_agg_schema_change_timestamp,
 	_Anum_continuous_agg_max,
 } Anum_continuous_agg;
 
@@ -832,6 +833,7 @@ typedef struct FormData_continuous_agg
 	NameData direct_view_schema;
 	NameData direct_view_name;
 	bool materialized_only;
+	int64 schema_change_timestamp;
 } FormData_continuous_agg;
 
 typedef FormData_continuous_agg *Form_continuous_agg;

src/ts_catalog/continuous_agg.c

@@ -263,6 +263,16 @@ continuous_agg_formdata_make_tuple(const FormData_continuous_agg *fd, TupleDesc
 	values[AttrNumberGetAttrOffset(Anum_continuous_agg_materialize_only)] =
 		BoolGetDatum(fd->materialized_only);
 
+	if (fd->schema_change_timestamp == TS_TIME_NOBEGIN)
+	{
+		nulls[AttrNumberGetAttrOffset(Anum_continuous_agg_schema_change_timestamp)] = true;
+	}
+	else
+	{
+		values[AttrNumberGetAttrOffset(Anum_continuous_agg_schema_change_timestamp)] =
+			Int64GetDatum(fd->schema_change_timestamp);
+	}
+
 	return heap_form_tuple(desc, values, nulls);
 }
 
@@ -315,6 +325,17 @@ continuous_agg_formdata_fill(FormData_continuous_agg *fd, const TupleInfo *ti)
 
 	fd->materialized_only =
 		DatumGetBool(values[AttrNumberGetAttrOffset(Anum_continuous_agg_materialize_only)]);
+
+	if (nulls[AttrNumberGetAttrOffset(Anum_continuous_agg_schema_change_timestamp)])
+	{
+		fd->schema_change_timestamp = TS_TIME_NOBEGIN;
+	}
+	else
+	{
+		fd->schema_change_timestamp = DatumGetInt64(
+			values[AttrNumberGetAttrOffset(Anum_continuous_agg_schema_change_timestamp)]);
+	}
+
 	if (should_free)
 	{
 		heap_freetuple(tuple);
@@ -627,6 +648,42 @@ ts_continuous_agg_find_by_mat_hypertable_id(int32 mat_hypertable_id, bool missin
 	return ca;
 }
 
+/*
+ * Record the threshold below which a newly added column is not yet materialized,
+ * so the planner rewrite can avoid serving for the cagg. The threshold only ever moves
+ * forward, so we keep the highest value across multiple ADD COLUMN operations.
+ */
+void
+ts_continuous_agg_set_schema_change_timestamp(int32 mat_hypertable_id, int64 threshold)
+{
+	ScanIterator iterator =
+		ts_scan_iterator_create(CONTINUOUS_AGG, RowExclusiveLock, CurrentMemoryContext);
+	CatalogSecurityContext sec_ctx;
+
+	init_scan_by_mat_hypertable_id(&iterator, mat_hypertable_id);
+	ts_scanner_foreach(&iterator)
+	{
+		TupleInfo *ti = ts_scan_iterator_tuple_info(&iterator);
+		FormData_continuous_agg form;
+
+		continuous_agg_formdata_fill(&form, ti);
+
+		if (form.schema_change_timestamp == TS_TIME_NOBEGIN ||
+			threshold > form.schema_change_timestamp)
+		{
+			form.schema_change_timestamp = threshold;
+
+			HeapTuple new_tuple =
+				continuous_agg_formdata_make_tuple(&form, ts_scanner_get_tupledesc(ti));
+			ts_catalog_database_info_become_owner(ts_catalog_database_info_get(), &sec_ctx);
+			ts_catalog_update_tid(ti->scanrel, ts_scanner_get_tuple_tid(ti), new_tuple);
+			ts_catalog_restore_user(&sec_ctx);
+			heap_freetuple(new_tuple);
+		}
+	}
+	ts_scan_iterator_close(&iterator);
+}
+
 static bool
 continuous_agg_find_by_name(const char *schema, const char *name, ContinuousAggViewType type,
 							FormData_continuous_agg *fd)

src/ts_catalog/continuous_agg.h

@@ -151,6 +151,9 @@ extern TSDLLEXPORT ContinuousAggInfo ts_continuous_agg_get_all_caggs_info(int32
 extern TSDLLEXPORT ContinuousAgg *
 ts_continuous_agg_find_by_mat_hypertable_id(int32 mat_hypertable_id, bool missing_ok);
 
+extern TSDLLEXPORT void ts_continuous_agg_set_schema_change_timestamp(int32 mat_hypertable_id,
+																	  int64 threshold);
+
 extern TSDLLEXPORT ContinuousAggHypertableStatus
 ts_continuous_agg_hypertable_status(int32 hypertable_id);
 extern TSDLLEXPORT List *ts_continuous_aggs_find_by_raw_table_id(int32 raw_hypertable_id);

tsl/src/continuous_aggs/CMakeLists.txt

@@ -1,4 +1,5 @@
 set(SOURCES
+    ${CMAKE_CURRENT_SOURCE_DIR}/add_column.c
     ${CMAKE_CURRENT_SOURCE_DIR}/common.c
     ${CMAKE_CURRENT_SOURCE_DIR}/create.c
     ${CMAKE_CURRENT_SOURCE_DIR}/finalize.c