Skip to content

Commit 14dd2ff

Browse files
melihmutlumelihmutlu
committed
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.
1 parent 14583bd commit 14dd2ff

33 files changed

Lines changed: 3638 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
@@ -341,6 +341,7 @@ CREATE TABLE _timescaledb_catalog.continuous_agg (
341341
direct_view_schema name NOT NULL,
342342
direct_view_name name NOT NULL,
343343
materialized_only bool NOT NULL DEFAULT FALSE,
344+
schema_change_timestamp bigint,
344345
-- table constraints
345346
CONSTRAINT continuous_agg_pkey PRIMARY KEY (mat_hypertable_id),
346347
CONSTRAINT continuous_agg_partial_view_schema_partial_view_name_key UNIQUE (partial_view_schema, partial_view_name),

‎sql/updates/latest-dev.sql‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -158,3 +158,97 @@ ALTER EXTENSION timescaledb DROP SEQUENCE _timescaledb_catalog.chunk_constraint_
158158
DROP TABLE _timescaledb_catalog.chunk_constraint;
159159
DROP SEQUENCE _timescaledb_catalog.chunk_constraint_name;
160160

161+
-- Rebuild the catalog table `_timescaledb_catalog.continuous_agg` to add the
162+
-- `schema_change_timestamp` column.
163+
164+
-- Drop views and foreign keys that depend on the catalog table.
165+
DROP VIEW IF EXISTS timescaledb_experimental.policies;
166+
DROP VIEW IF EXISTS timescaledb_information.hypertables;
167+
DROP VIEW IF EXISTS timescaledb_information.continuous_aggregates;
168+
DROP VIEW IF EXISTS timescaledb_information.jobs;
169+
170+
ALTER TABLE _timescaledb_catalog.continuous_aggs_watermark
171+
DROP CONSTRAINT continuous_aggs_watermark_mat_hypertable_id_fkey;
172+
ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
173+
DROP CONSTRAINT continuous_aggs_materialization_invalid_materialization_id_fkey;
174+
ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_ranges
175+
DROP CONSTRAINT continuous_aggs_materialization_ranges_materialization_id_fkey;
176+
ALTER TABLE _timescaledb_catalog.continuous_aggs_jobs_refresh_ranges
177+
DROP CONSTRAINT continuous_aggs_jobs_refresh_ranges_materialization_id_fkey;
178+
179+
ALTER EXTENSION timescaledb
180+
DROP TABLE _timescaledb_catalog.continuous_agg;
181+
182+
CREATE TABLE _timescaledb_catalog._tmp_continuous_agg AS
183+
SELECT
184+
mat_hypertable_id,
185+
raw_hypertable_id,
186+
parent_mat_hypertable_id,
187+
user_view_schema,
188+
user_view_name,
189+
partial_view_schema,
190+
partial_view_name,
191+
direct_view_schema,
192+
direct_view_name,
193+
materialized_only
194+
FROM
195+
_timescaledb_catalog.continuous_agg
196+
ORDER BY
197+
mat_hypertable_id;
198+
199+
DROP TABLE _timescaledb_catalog.continuous_agg;
200+
201+
CREATE TABLE _timescaledb_catalog.continuous_agg (
202+
mat_hypertable_id integer NOT NULL,
203+
raw_hypertable_id integer NOT NULL,
204+
parent_mat_hypertable_id integer,
205+
user_view_schema name NOT NULL,
206+
user_view_name name NOT NULL,
207+
partial_view_schema name NOT NULL,
208+
partial_view_name name NOT NULL,
209+
direct_view_schema name NOT NULL,
210+
direct_view_name name NOT NULL,
211+
materialized_only bool NOT NULL DEFAULT FALSE,
212+
schema_change_timestamp bigint,
213+
-- table constraints
214+
CONSTRAINT continuous_agg_pkey PRIMARY KEY (mat_hypertable_id),
215+
CONSTRAINT continuous_agg_partial_view_schema_partial_view_name_key UNIQUE (partial_view_schema, partial_view_name),
216+
CONSTRAINT continuous_agg_user_view_schema_user_view_name_key UNIQUE (user_view_schema, user_view_name),
217+
CONSTRAINT continuous_agg_mat_hypertable_id_fkey FOREIGN KEY (mat_hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE,
218+
CONSTRAINT continuous_agg_raw_hypertable_id_fkey FOREIGN KEY (raw_hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE,
219+
CONSTRAINT continuous_agg_parent_mat_hypertable_id_fkey FOREIGN KEY (parent_mat_hypertable_id)
220+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE
221+
);
222+
223+
INSERT INTO _timescaledb_catalog.continuous_agg
224+
(mat_hypertable_id, raw_hypertable_id, parent_mat_hypertable_id, user_view_schema,
225+
user_view_name, partial_view_schema, partial_view_name, direct_view_schema,
226+
direct_view_name, materialized_only)
227+
SELECT * FROM _timescaledb_catalog._tmp_continuous_agg;
228+
DROP TABLE _timescaledb_catalog._tmp_continuous_agg;
229+
230+
CREATE INDEX continuous_agg_raw_hypertable_id_idx ON _timescaledb_catalog.continuous_agg (raw_hypertable_id);
231+
232+
SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_agg', '');
233+
234+
GRANT SELECT ON TABLE _timescaledb_catalog.continuous_agg TO PUBLIC;
235+
236+
ALTER TABLE _timescaledb_catalog.continuous_aggs_watermark
237+
ADD CONSTRAINT continuous_aggs_watermark_mat_hypertable_id_fkey
238+
FOREIGN KEY (mat_hypertable_id)
239+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE;
240+
ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
241+
ADD CONSTRAINT continuous_aggs_materialization_invalid_materialization_id_fkey
242+
FOREIGN KEY (materialization_id)
243+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE;
244+
ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_ranges
245+
ADD CONSTRAINT continuous_aggs_materialization_ranges_materialization_id_fkey
246+
FOREIGN KEY (materialization_id)
247+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE;
248+
ALTER TABLE _timescaledb_catalog.continuous_aggs_jobs_refresh_ranges
249+
ADD CONSTRAINT continuous_aggs_jobs_refresh_ranges_materialization_id_fkey
250+
FOREIGN KEY (materialization_id)
251+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE;
252+
253+
ANALYZE _timescaledb_catalog.continuous_agg;
254+
-- end rebuild _timescaledb_catalog.continuous_agg --

‎sql/updates/reverse-dev.sql‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -184,3 +184,92 @@ GRANT SELECT ON _timescaledb_catalog.dimension_slice TO PUBLIC;
184184
GRANT SELECT ON _timescaledb_catalog.dimension_slice_id_seq TO PUBLIC;
185185
-- end rebuild _timescaledb_catalog.dimension_slice table --
186186

187+
-- Rebuild the catalog table `_timescaledb_catalog.continuous_agg` to drop the
188+
-- `schema_change_timestamp` column.
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+
-- table constraints
237+
CONSTRAINT continuous_agg_pkey PRIMARY KEY (mat_hypertable_id),
238+
CONSTRAINT continuous_agg_partial_view_schema_partial_view_name_key UNIQUE (partial_view_schema, partial_view_name),
239+
CONSTRAINT continuous_agg_user_view_schema_user_view_name_key UNIQUE (user_view_schema, user_view_name),
240+
CONSTRAINT continuous_agg_mat_hypertable_id_fkey FOREIGN KEY (mat_hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE,
241+
CONSTRAINT continuous_agg_raw_hypertable_id_fkey FOREIGN KEY (raw_hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE,
242+
CONSTRAINT continuous_agg_parent_mat_hypertable_id_fkey FOREIGN KEY (parent_mat_hypertable_id)
243+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE
244+
);
245+
246+
INSERT INTO _timescaledb_catalog.continuous_agg
247+
SELECT * FROM _timescaledb_catalog._tmp_continuous_agg;
248+
DROP TABLE _timescaledb_catalog._tmp_continuous_agg;
249+
250+
CREATE INDEX continuous_agg_raw_hypertable_id_idx ON _timescaledb_catalog.continuous_agg (raw_hypertable_id);
251+
252+
SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_agg', '');
253+
254+
GRANT SELECT ON TABLE _timescaledb_catalog.continuous_agg TO PUBLIC;
255+
256+
ALTER TABLE _timescaledb_catalog.continuous_aggs_watermark
257+
ADD CONSTRAINT continuous_aggs_watermark_mat_hypertable_id_fkey
258+
FOREIGN KEY (mat_hypertable_id)
259+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE;
260+
ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
261+
ADD CONSTRAINT continuous_aggs_materialization_invalid_materialization_id_fkey
262+
FOREIGN KEY (materialization_id)
263+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE;
264+
ALTER TABLE _timescaledb_catalog.continuous_aggs_materialization_ranges
265+
ADD CONSTRAINT continuous_aggs_materialization_ranges_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_jobs_refresh_ranges
269+
ADD CONSTRAINT continuous_aggs_jobs_refresh_ranges_materialization_id_fkey
270+
FOREIGN KEY (materialization_id)
271+
REFERENCES _timescaledb_catalog.continuous_agg (mat_hypertable_id) ON DELETE CASCADE;
272+
273+
ANALYZE _timescaledb_catalog.continuous_agg;
274+
-- end rebuild _timescaledb_catalog.continuous_agg --
275+

‎src/cross_module_fn.c‎

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

232+
static void
233+
continuous_agg_add_column_default(ContinuousAgg *cagg, AlterTableStmt *stmt)
234+
{
235+
error_no_default_fn_community();
236+
pg_unreachable();
237+
}
238+
232239
static void
233240
continuous_agg_invalidate_raw_ht_all_default(const Hypertable *raw_ht, int64 start, int64 end)
234241
{
@@ -337,6 +344,7 @@ TSDLLEXPORT CrossModuleFunctions ts_cm_functions_default = {
337344
.continuous_agg_invalidate_mat_ht = continuous_agg_invalidate_mat_ht_all_default,
338345
.continuous_agg_dml_invalidate = continuous_agg_dml_invalidate_default,
339346
.continuous_agg_update_options = continuous_agg_update_options_default,
347+
.continuous_agg_add_column = continuous_agg_add_column_default,
340348
.continuous_agg_apply_rewrites_tsl = NULL,
341349
.continuous_agg_validate_query = error_no_default_fn_pg_community,
342350
.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
@@ -5080,6 +5080,40 @@ process_altertable_start_matview(ProcessUtilityArgs *args)
50805080

50815081
continuous_agg_with_clause_perm_check(cagg, view_relid);
50825082

5083+
/*
5084+
* CAgg add column handler.
5085+
* Split `stmt->cmds` into the GENERATED ALWAYS AS and everything else
5086+
* (handled by the switch below). Process all ADD COLUMNs at once since
5087+
* they require a rewrite of the view, and doing them one by one would
5088+
* be inefficient.
5089+
*/
5090+
{
5091+
List *addcol_cmds = NIL;
5092+
List *other_cmds = NIL;
5093+
foreach (lc, stmt->cmds)
5094+
{
5095+
AlterTableCmd *cmd = (AlterTableCmd *) lfirst(lc);
5096+
if (cmd->subtype == AT_AddColumn)
5097+
{
5098+
addcol_cmds = lappend(addcol_cmds, cmd);
5099+
}
5100+
else
5101+
{
5102+
other_cmds = lappend(other_cmds, cmd);
5103+
}
5104+
}
5105+
5106+
if (addcol_cmds != NIL)
5107+
{
5108+
AlterTableStmt addcol_stmt = *stmt;
5109+
addcol_stmt.cmds = addcol_cmds;
5110+
ts_cm_functions->continuous_agg_add_column(cagg, &addcol_stmt);
5111+
CommandCounterIncrement();
5112+
5113+
stmt->cmds = other_cmds;
5114+
}
5115+
}
5116+
50835117
foreach (lc, stmt->cmds)
50845118
{
50855119
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 == PG_INT64_MIN)
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 = PG_INT64_MIN;
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 == PG_INT64_MIN ||
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)