Support ADD COLUMN on continuous aggregates

Allow ALTER MATERIALIZED VIEW <cagg> ADD COLUMN <name> <type>
GENERATED ALWAYS AS (<aggregate>) STORED.

The new column is added to the materialization hypertable and surfaced
through the partial, direct, and user views. AccessExclusiveLock's are
taken on each of them at the beginning. New columns will have NULL
values on already existing rows from the CAgg, new rows inserted after
the ADD COLUMN statement will have appropriate values. To backfill
previous rows, forced refresh is required.

Author: melihmutlu

.unreleased/pr_9825

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

sql/pre_install/tables.sql

@@ -341,6 +341,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

@@ -158,3 +158,7 @@ ALTER EXTENSION timescaledb DROP SEQUENCE _timescaledb_catalog.chunk_constraint_
 DROP TABLE _timescaledb_catalog.chunk_constraint;
 DROP SEQUENCE _timescaledb_catalog.chunk_constraint_name;
 
+-- Track the watermark at which a column was added to a continuous aggregate via
+-- ALTER ... ADD COLUMN, so the planner does not rewrite
+ALTER TABLE _timescaledb_catalog.continuous_agg ADD COLUMN schema_change_timestamp bigint;
+

sql/updates/reverse-dev.sql

@@ -183,4 +183,5 @@ SELECT pg_catalog.pg_extension_config_dump(pg_get_serial_sequence('_timescaledb_
 GRANT SELECT ON _timescaledb_catalog.dimension_slice TO PUBLIC;
 GRANT SELECT ON _timescaledb_catalog.dimension_slice_id_seq TO PUBLIC;
 -- end rebuild _timescaledb_catalog.dimension_slice table --
+ALTER TABLE _timescaledb_catalog.continuous_agg DROP COLUMN schema_change_timestamp;
 

src/cross_module_fn.c

@@ -229,6 +229,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)
 {
@@ -337,6 +344,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

@@ -5080,6 +5080,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 == PG_INT64_MIN)
+	{
+		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 = PG_INT64_MIN;
+	}
+	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 == PG_INT64_MIN ||
+			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);