Skip to content

Commit ee80bfa

Browse files
melihmutlumelihmutlu
authored
Support ADD COLUMN on continuous aggregates (#9825)
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.
1 parent fd3c049 commit ee80bfa

33 files changed

Lines changed: 3715 additions & 27 deletions

‎.unreleased/pr_9825‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Implements: #9825 Support ADD COLUMN on continuous aggregates

‎sql/pre_install/tables.sql‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -340,6 +340,7 @@ CREATE TABLE _timescaledb_catalog.continuous_agg (
340340
direct_view_schema name NOT NULL,
341341
direct_view_name name NOT NULL,
342342
materialized_only bool NOT NULL DEFAULT FALSE,
343+
schema_change_timestamp bigint,
343344
-- table constraints
344345
CONSTRAINT continuous_agg_pkey PRIMARY KEY (mat_hypertable_id),
345346
CONSTRAINT continuous_agg_partial_view_schema_partial_view_name_key UNIQUE (partial_view_schema, partial_view_name),

‎sql/updates/latest-dev.sql‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -181,3 +181,98 @@ DROP FUNCTION IF EXISTS _timescaledb_functions.calculate_chunk_interval(integer,
181181
-- chunk_target_size is no longer used, so its check constraint is dropped.
182182
ALTER TABLE _timescaledb_catalog.hypertable
183183
DROP CONSTRAINT IF EXISTS hypertable_chunk_target_size_check;
184+
185+
-- Rebuild the catalog table `_timescaledb_catalog.continuous_agg` to add the
186+
-- `schema_change_timestamp` column.
187+
188+
-- Drop views and foreign keys that depend on the catalog table.
189+
DROP VIEW IF EXISTS timescaledb_experimental.policies;
190+
DROP VIEW IF EXISTS timescaledb_information.hypertables;
191+
DROP VIEW IF EXISTS timescaledb_information.continuous_aggregates;
192+
DROP VIEW IF EXISTS timescaledb_information.jobs;
193+
194+
ALTER TABLE _timescaledb_catalog.continuous_aggs_watermark
195+
DROP CONSTRAINT continuous_aggs_watermark_mat_hypertable_id_fkey;
196+
ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
197+
DROP CONSTRAINT continuous_aggs_materialization_invalid_materialization_id_fkey;
198+
ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_ranges
199+
DROP CONSTRAINT continuous_aggs_materialization_ranges_materialization_id_fkey;
200+
ALTER TABLE _timescaledb_catalog.continuous_aggs_jobs_refresh_ranges
201+
DROP CONSTRAINT continuous_aggs_jobs_refresh_ranges_materialization_id_fkey;
202+
203+
ALTER EXTENSION timescaledb
204+
DROP TABLE _timescaledb_catalog.continuous_agg;
205+
206+
CREATE TABLE _timescaledb_catalog._tmp_continuous_agg AS
207+
SELECT
208+
mat_hypertable_id,
209+
raw_hypertable_id,
210+
parent_mat_hypertable_id,
211+
user_view_schema,
212+
user_view_name,
213+
partial_view_schema,
214+
partial_view_name,
215+
direct_view_schema,
216+
direct_view_name,
217+
materialized_only
218+
FROM
219+
_timescaledb_catalog.continuous_agg
220+
ORDER BY
221+
mat_hypertable_id;
222+
223+
DROP TABLE _timescaledb_catalog.continuous_agg;
224+
225+
CREATE TABLE _timescaledb_catalog.continuous_agg (
226+
mat_hypertable_id integer NOT NULL,
227+
raw_hypertable_id integer NOT NULL,
228+
parent_mat_hypertable_id integer,
229+
user_view_schema name NOT NULL,
230+
user_view_name name NOT NULL,
231+
partial_view_schema name NOT NULL,
232+
partial_view_name name NOT NULL,
233+
direct_view_schema name NOT NULL,
234+
direct_view_name name NOT NULL,
235+
materialized_only bool NOT NULL DEFAULT FALSE,
236+
schema_change_timestamp bigint,
237+
-- table constraints
238+
CONSTRAINT continuous_agg_pkey PRIMARY KEY (mat_hypertable_id),
239+
CONSTRAINT continuous_agg_partial_view_schema_partial_view_name_key UNIQUE (partial_view_schema, partial_view_name),
240+
CONSTRAINT continuous_agg_user_view_schema_user_view_name_key UNIQUE (user_view_schema, user_view_name),
241+
CONSTRAINT continuous_agg_mat_hypertable_id_fkey FOREIGN KEY (mat_hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE,
242+
CONSTRAINT continuous_agg_raw_hypertable_id_fkey FOREIGN KEY (raw_hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE,
243+
CONSTRAINT continuous_agg_parent_mat_hypertable_id_fkey FOREIGN KEY (parent_mat_hypertable_id)
244+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE
245+
);
246+
247+
INSERT INTO _timescaledb_catalog.continuous_agg
248+
(mat_hypertable_id, raw_hypertable_id, parent_mat_hypertable_id, user_view_schema,
249+
user_view_name, partial_view_schema, partial_view_name, direct_view_schema,
250+
direct_view_name, materialized_only)
251+
SELECT * FROM _timescaledb_catalog._tmp_continuous_agg;
252+
DROP TABLE _timescaledb_catalog._tmp_continuous_agg;
253+
254+
CREATE INDEX continuous_agg_raw_hypertable_id_idx ON _timescaledb_catalog.continuous_agg (raw_hypertable_id);
255+
256+
SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_agg', '');
257+
258+
GRANT SELECT ON TABLE _timescaledb_catalog.continuous_agg TO PUBLIC;
259+
260+
ALTER TABLE _timescaledb_catalog.continuous_aggs_watermark
261+
ADD CONSTRAINT continuous_aggs_watermark_mat_hypertable_id_fkey
262+
FOREIGN KEY (mat_hypertable_id)
263+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE;
264+
ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
265+
ADD CONSTRAINT continuous_aggs_materialization_invalid_materialization_id_fkey
266+
FOREIGN KEY (materialization_id)
267+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE;
268+
ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_ranges
269+
ADD CONSTRAINT continuous_aggs_materialization_ranges_materialization_id_fkey
270+
FOREIGN KEY (materialization_id)
271+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE;
272+
ALTER TABLE _timescaledb_catalog.continuous_aggs_jobs_refresh_ranges
273+
ADD CONSTRAINT continuous_aggs_jobs_refresh_ranges_materialization_id_fkey
274+
FOREIGN KEY (materialization_id)
275+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE;
276+
277+
ANALYZE _timescaledb_catalog.continuous_agg;
278+
-- end rebuild _timescaledb_catalog.continuous_agg --

‎sql/updates/reverse-dev.sql‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -189,3 +189,93 @@ DROP FUNCTION IF EXISTS @extschema@.create_hypertable(relation REGCLASS, time_co
189189
ALTER TABLE _timescaledb_catalog.hypertable
190190
ADD CONSTRAINT hypertable_chunk_target_size_check CHECK (chunk_target_size >= 0);
191191
DROP FUNCTION IF EXISTS _timescaledb_functions.rebuild_sparse_index(REGCLASS, BOOLEAN);
192+
193+
-- Rebuild the catalog table `_timescaledb_catalog.continuous_agg` to drop the
194+
-- `schema_change_timestamp` column.
195+
DROP VIEW IF EXISTS timescaledb_experimental.policies;
196+
DROP VIEW IF EXISTS timescaledb_information.hypertables;
197+
DROP VIEW IF EXISTS timescaledb_information.continuous_aggregates;
198+
DROP VIEW IF EXISTS timescaledb_information.jobs;
199+
200+
ALTER TABLE _timescaledb_catalog.continuous_aggs_watermark
201+
DROP CONSTRAINT continuous_aggs_watermark_mat_hypertable_id_fkey;
202+
ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
203+
DROP CONSTRAINT continuous_aggs_materialization_invalid_materialization_id_fkey;
204+
ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_ranges
205+
DROP CONSTRAINT continuous_aggs_materialization_ranges_materialization_id_fkey;
206+
ALTER TABLE _timescaledb_catalog.continuous_aggs_jobs_refresh_ranges
207+
DROP CONSTRAINT continuous_aggs_jobs_refresh_ranges_materialization_id_fkey;
208+
209+
ALTER EXTENSION timescaledb
210+
DROP TABLE _timescaledb_catalog.continuous_agg;
211+
212+
CREATE TABLE _timescaledb_catalog._tmp_continuous_agg AS
213+
SELECT
214+
mat_hypertable_id,
215+
raw_hypertable_id,
216+
parent_mat_hypertable_id,
217+
user_view_schema,
218+
user_view_name,
219+
partial_view_schema,
220+
partial_view_name,
221+
direct_view_schema,
222+
direct_view_name,
223+
materialized_only
224+
FROM
225+
_timescaledb_catalog.continuous_agg
226+
ORDER BY
227+
mat_hypertable_id;
228+
229+
DROP TABLE _timescaledb_catalog.continuous_agg;
230+
231+
CREATE TABLE _timescaledb_catalog.continuous_agg (
232+
mat_hypertable_id integer NOT NULL,
233+
raw_hypertable_id integer NOT NULL,
234+
parent_mat_hypertable_id integer,
235+
user_view_schema name NOT NULL,
236+
user_view_name name NOT NULL,
237+
partial_view_schema name NOT NULL,
238+
partial_view_name name NOT NULL,
239+
direct_view_schema name NOT NULL,
240+
direct_view_name name NOT NULL,
241+
materialized_only bool NOT NULL DEFAULT FALSE,
242+
-- table constraints
243+
CONSTRAINT continuous_agg_pkey PRIMARY KEY (mat_hypertable_id),
244+
CONSTRAINT continuous_agg_partial_view_schema_partial_view_name_key UNIQUE (partial_view_schema, partial_view_name),
245+
CONSTRAINT continuous_agg_user_view_schema_user_view_name_key UNIQUE (user_view_schema, user_view_name),
246+
CONSTRAINT continuous_agg_mat_hypertable_id_fkey FOREIGN KEY (mat_hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE,
247+
CONSTRAINT continuous_agg_raw_hypertable_id_fkey FOREIGN KEY (raw_hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE,
248+
CONSTRAINT continuous_agg_parent_mat_hypertable_id_fkey FOREIGN KEY (parent_mat_hypertable_id)
249+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE
250+
);
251+
252+
INSERT INTO _timescaledb_catalog.continuous_agg
253+
SELECT * FROM _timescaledb_catalog._tmp_continuous_agg;
254+
DROP TABLE _timescaledb_catalog._tmp_continuous_agg;
255+
256+
CREATE INDEX continuous_agg_raw_hypertable_id_idx ON _timescaledb_catalog.continuous_agg (raw_hypertable_id);
257+
258+
SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_agg', '');
259+
260+
GRANT SELECT ON TABLE _timescaledb_catalog.continuous_agg TO PUBLIC;
261+
262+
ALTER TABLE _timescaledb_catalog.continuous_aggs_watermark
263+
ADD CONSTRAINT continuous_aggs_watermark_mat_hypertable_id_fkey
264+
FOREIGN KEY (mat_hypertable_id)
265+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE;
266+
ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
267+
ADD CONSTRAINT continuous_aggs_materialization_invalid_materialization_id_fkey
268+
FOREIGN KEY (materialization_id)
269+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE;
270+
ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_ranges
271+
ADD CONSTRAINT continuous_aggs_materialization_ranges_materialization_id_fkey
272+
FOREIGN KEY (materialization_id)
273+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE;
274+
ALTER TABLE _timescaledb_catalog.continuous_aggs_jobs_refresh_ranges
275+
ADD CONSTRAINT continuous_aggs_jobs_refresh_ranges_materialization_id_fkey
276+
FOREIGN KEY (materialization_id)
277+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE;
278+
279+
ANALYZE _timescaledb_catalog.continuous_agg;
280+
-- end rebuild _timescaledb_catalog.continuous_agg --
281+

‎src/cross_module_fn.c‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -230,6 +230,13 @@ continuous_agg_update_options_default(ContinuousAgg *cagg, WithClauseResult *wit
230230
pg_unreachable();
231231
}
232232

233+
static void
234+
continuous_agg_add_column_default(ContinuousAgg *cagg, AlterTableStmt *stmt)
235+
{
236+
error_no_default_fn_community();
237+
pg_unreachable();
238+
}
239+
233240
static void
234241
continuous_agg_invalidate_raw_ht_all_default(const Hypertable *raw_ht, int64 start, int64 end)
235242
{
@@ -338,6 +345,7 @@ TSDLLEXPORT CrossModuleFunctions ts_cm_functions_default = {
338345
.continuous_agg_invalidate_mat_ht = continuous_agg_invalidate_mat_ht_all_default,
339346
.continuous_agg_dml_invalidate = continuous_agg_dml_invalidate_default,
340347
.continuous_agg_update_options = continuous_agg_update_options_default,
348+
.continuous_agg_add_column = continuous_agg_add_column_default,
341349
.continuous_agg_apply_rewrites_tsl = NULL,
342350
.continuous_agg_validate_query = error_no_default_fn_pg_community,
343351
.continuous_agg_get_bucket_function = error_no_default_fn_pg_community,

‎src/cross_module_fn.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,7 @@ typedef struct CrossModuleFunctions
105105
bool update);
106106
void (*continuous_agg_update_options)(ContinuousAgg *cagg,
107107
WithClauseResult *with_clause_options);
108+
void (*continuous_agg_add_column)(ContinuousAgg *cagg, AlterTableStmt *stmt);
108109
Query *(*continuous_agg_apply_rewrites_tsl)(Query *parse);
109110
PGFunction continuous_agg_validate_query;
110111
PGFunction continuous_agg_get_bucket_function;

‎src/process_utility.c‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5072,6 +5072,40 @@ process_altertable_start_matview(ProcessUtilityArgs *args)
50725072

50735073
continuous_agg_with_clause_perm_check(cagg, view_relid);
50745074

5075+
/*
5076+
* CAgg add column handler.
5077+
* Split `stmt->cmds` into the GENERATED ALWAYS AS and everything else
5078+
* (handled by the switch below). Process all ADD COLUMNs at once since
5079+
* they require a rewrite of the view, and doing them one by one would
5080+
* be inefficient.
5081+
*/
5082+
{
5083+
List *addcol_cmds = NIL;
5084+
List *other_cmds = NIL;
5085+
foreach (lc, stmt->cmds)
5086+
{
5087+
AlterTableCmd *cmd = (AlterTableCmd *) lfirst(lc);
5088+
if (cmd->subtype == AT_AddColumn)
5089+
{
5090+
addcol_cmds = lappend(addcol_cmds, cmd);
5091+
}
5092+
else
5093+
{
5094+
other_cmds = lappend(other_cmds, cmd);
5095+
}
5096+
}
5097+
5098+
if (addcol_cmds != NIL)
5099+
{
5100+
AlterTableStmt addcol_stmt = *stmt;
5101+
addcol_stmt.cmds = addcol_cmds;
5102+
ts_cm_functions->continuous_agg_add_column(cagg, &addcol_stmt);
5103+
CommandCounterIncrement();
5104+
5105+
stmt->cmds = other_cmds;
5106+
}
5107+
}
5108+
50755109
foreach (lc, stmt->cmds)
50765110
{
50775111
AlterTableCmd *cmd = (AlterTableCmd *) lfirst(lc);

‎src/ts_catalog/catalog.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -815,6 +815,7 @@ typedef enum Anum_continuous_agg
815815
Anum_continuous_agg_direct_view_schema,
816816
Anum_continuous_agg_direct_view_name,
817817
Anum_continuous_agg_materialize_only,
818+
Anum_continuous_agg_schema_change_timestamp,
818819
_Anum_continuous_agg_max,
819820
} Anum_continuous_agg;
820821

@@ -832,6 +833,7 @@ typedef struct FormData_continuous_agg
832833
NameData direct_view_schema;
833834
NameData direct_view_name;
834835
bool materialized_only;
836+
int64 schema_change_timestamp;
835837
} FormData_continuous_agg;
836838

837839
typedef FormData_continuous_agg *Form_continuous_agg;

‎src/ts_catalog/continuous_agg.c‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -263,6 +263,16 @@ continuous_agg_formdata_make_tuple(const FormData_continuous_agg *fd, TupleDesc
263263
values[AttrNumberGetAttrOffset(Anum_continuous_agg_materialize_only)] =
264264
BoolGetDatum(fd->materialized_only);
265265

266+
if (fd->schema_change_timestamp == TS_TIME_NOBEGIN)
267+
{
268+
nulls[AttrNumberGetAttrOffset(Anum_continuous_agg_schema_change_timestamp)] = true;
269+
}
270+
else
271+
{
272+
values[AttrNumberGetAttrOffset(Anum_continuous_agg_schema_change_timestamp)] =
273+
Int64GetDatum(fd->schema_change_timestamp);
274+
}
275+
266276
return heap_form_tuple(desc, values, nulls);
267277
}
268278

@@ -315,6 +325,17 @@ continuous_agg_formdata_fill(FormData_continuous_agg *fd, const TupleInfo *ti)
315325

316326
fd->materialized_only =
317327
DatumGetBool(values[AttrNumberGetAttrOffset(Anum_continuous_agg_materialize_only)]);
328+
329+
if (nulls[AttrNumberGetAttrOffset(Anum_continuous_agg_schema_change_timestamp)])
330+
{
331+
fd->schema_change_timestamp = TS_TIME_NOBEGIN;
332+
}
333+
else
334+
{
335+
fd->schema_change_timestamp = DatumGetInt64(
336+
values[AttrNumberGetAttrOffset(Anum_continuous_agg_schema_change_timestamp)]);
337+
}
338+
318339
if (should_free)
319340
{
320341
heap_freetuple(tuple);
@@ -627,6 +648,42 @@ ts_continuous_agg_find_by_mat_hypertable_id(int32 mat_hypertable_id, bool missin
627648
return ca;
628649
}
629650

651+
/*
652+
* Record the threshold below which a newly added column is not yet materialized,
653+
* so the planner rewrite can avoid serving for the cagg. The threshold only ever moves
654+
* forward, so we keep the highest value across multiple ADD COLUMN operations.
655+
*/
656+
void
657+
ts_continuous_agg_set_schema_change_timestamp(int32 mat_hypertable_id, int64 threshold)
658+
{
659+
ScanIterator iterator =
660+
ts_scan_iterator_create(CONTINUOUS_AGG, RowExclusiveLock, CurrentMemoryContext);
661+
CatalogSecurityContext sec_ctx;
662+
663+
init_scan_by_mat_hypertable_id(&iterator, mat_hypertable_id);
664+
ts_scanner_foreach(&iterator)
665+
{
666+
TupleInfo *ti = ts_scan_iterator_tuple_info(&iterator);
667+
FormData_continuous_agg form;
668+
669+
continuous_agg_formdata_fill(&form, ti);
670+
671+
if (form.schema_change_timestamp == TS_TIME_NOBEGIN ||
672+
threshold > form.schema_change_timestamp)
673+
{
674+
form.schema_change_timestamp = threshold;
675+
676+
HeapTuple new_tuple =
677+
continuous_agg_formdata_make_tuple(&form, ts_scanner_get_tupledesc(ti));
678+
ts_catalog_database_info_become_owner(ts_catalog_database_info_get(), &sec_ctx);
679+
ts_catalog_update_tid(ti->scanrel, ts_scanner_get_tuple_tid(ti), new_tuple);
680+
ts_catalog_restore_user(&sec_ctx);
681+
heap_freetuple(new_tuple);
682+
}
683+
}
684+
ts_scan_iterator_close(&iterator);
685+
}
686+
630687
static bool
631688
continuous_agg_find_by_name(const char *schema, const char *name, ContinuousAggViewType type,
632689
FormData_continuous_agg *fd)

‎src/ts_catalog/continuous_agg.h‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -151,6 +151,9 @@ extern TSDLLEXPORT ContinuousAggInfo ts_continuous_agg_get_all_caggs_info(int32
151151
extern TSDLLEXPORT ContinuousAgg *
152152
ts_continuous_agg_find_by_mat_hypertable_id(int32 mat_hypertable_id, bool missing_ok);
153153

154+
extern TSDLLEXPORT void ts_continuous_agg_set_schema_change_timestamp(int32 mat_hypertable_id,
155+
int64 threshold);
156+
154157
extern TSDLLEXPORT ContinuousAggHypertableStatus
155158
ts_continuous_agg_hypertable_status(int32 hypertable_id);
156159
extern TSDLLEXPORT List *ts_continuous_aggs_find_by_raw_table_id(int32 raw_hypertable_id);

0 commit comments

Comments
 (0)