Skip to content

Commit 74093c3

Browse files
melihmutlupnthaogayyappan
committed
Introduce granular refresh related catalogs
Add catalog definitions and helper functions for continuous_aggs_tenant_tracking. - continuous_aggs_tenant_tracking catalog to store tenant-based granular trackings. - hypertable_cagg_settings catalog to store tenant tracking column, start and end offsets to calculate tracking window for granular refresh. Co-authored-by: Thao Pham <pnthao@users.noreply.github.com> Co-authored-by: gayyappan <gayathri@timescale.com>
1 parent 5252684 commit 74093c3

13 files changed

Lines changed: 376 additions & 2 deletions

‎.unreleased/pr_10299‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Implements: #10299 Add continuous_aggs_tenant_tracking and hypertable_cagg_settings catalogs

‎sql/pre_install/tables.sql‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -440,6 +440,35 @@ SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_aggs
440440

441441
CREATE INDEX continuous_aggs_jobs_refresh_ranges_idx ON _timescaledb_catalog.continuous_aggs_jobs_refresh_ranges (materialization_id);
442442

443+
-- Per-tenant invalidation tracking, used in granular refresh of contiuous aggregates.
444+
CREATE TABLE _timescaledb_catalog.continuous_aggs_tenant_tracking (
445+
hypertable_id integer NOT NULL,
446+
tenant_id text,
447+
min_timestamp bigint,
448+
max_timestamp bigint,
449+
seqnum integer NOT NULL,
450+
-- table constraints
451+
CONSTRAINT continuous_aggs_tenant_tracking_hypertable_id_fkey FOREIGN KEY (hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE
452+
);
453+
454+
SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_aggs_tenant_tracking', '');
455+
456+
CREATE INDEX continuous_aggs_tenant_tracking_idx ON _timescaledb_catalog.continuous_aggs_tenant_tracking (hypertable_id, seqnum);
457+
458+
-- Per-hypertable settings for granular refresh of continuous aggregates.
459+
-- Row existence means granular refresh is configured for the hypertable.
460+
CREATE TABLE _timescaledb_catalog.hypertable_cagg_settings (
461+
hypertable_id integer NOT NULL,
462+
granular_refresh_column name NOT NULL,
463+
granular_refresh_start_offset text,
464+
granular_refresh_end_offset text,
465+
-- table constraints
466+
CONSTRAINT hypertable_cagg_settings_pkey PRIMARY KEY (hypertable_id),
467+
CONSTRAINT hypertable_cagg_settings_hypertable_id_fkey FOREIGN KEY (hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE
468+
);
469+
470+
SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.hypertable_cagg_settings', '');
471+
443472
/* the source of this data is the enum from the source code that lists
444473
* the algorithms. This table is NOT dumped.
445474
*/

‎sql/updates/latest-dev.sql‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -147,3 +147,28 @@ GRANT SELECT ON _timescaledb_catalog.chunk TO PUBLIC;
147147
-- END add chunk.relid
148148
--
149149

150+
-- add continuous_aggs_tenant_tracking
151+
CREATE TABLE _timescaledb_catalog.continuous_aggs_tenant_tracking (
152+
hypertable_id integer NOT NULL,
153+
tenant_id text,
154+
min_timestamp bigint,
155+
max_timestamp bigint,
156+
seqnum integer NOT NULL,
157+
CONSTRAINT continuous_aggs_tenant_tracking_hypertable_id_fkey FOREIGN KEY (hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE
158+
);
159+
160+
SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.continuous_aggs_tenant_tracking', '');
161+
162+
CREATE INDEX continuous_aggs_tenant_tracking_idx ON _timescaledb_catalog.continuous_aggs_tenant_tracking (hypertable_id, seqnum);
163+
164+
-- add hypertable_cagg_settings
165+
CREATE TABLE _timescaledb_catalog.hypertable_cagg_settings (
166+
hypertable_id integer NOT NULL,
167+
granular_refresh_column name NOT NULL,
168+
granular_refresh_start_offset text,
169+
granular_refresh_end_offset text,
170+
CONSTRAINT hypertable_cagg_settings_pkey PRIMARY KEY (hypertable_id),
171+
CONSTRAINT hypertable_cagg_settings_hypertable_id_fkey FOREIGN KEY (hypertable_id) REFERENCES _timescaledb_catalog.hypertable (id) ON DELETE CASCADE
172+
);
173+
174+
SELECT pg_catalog.pg_extension_config_dump('_timescaledb_catalog.hypertable_cagg_settings', '');

‎sql/updates/reverse-dev.sql‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -260,3 +260,10 @@ DROP FUNCTION IF EXISTS _timescaledb_functions.policy_compaction_check(JSONB);
260260

261261
DROP FUNCTION IF EXISTS @extschema@.alter_job(job_id INTEGER, schedule_interval INTERVAL, max_runtime INTERVAL, max_retries INTEGER, retry_period INTERVAL, scheduled BOOL, config JSONB, next_start TIMESTAMPTZ, if_exists BOOL, check_config REGPROC, fixed_schedule BOOL, initial_start TIMESTAMPTZ, timezone TEXT, job_name TEXT, config_merge JSONB);
262262

263+
-- drop continuous_aggs_tenant_tracking
264+
ALTER EXTENSION timescaledb DROP TABLE _timescaledb_catalog.continuous_aggs_tenant_tracking;
265+
DROP TABLE _timescaledb_catalog.continuous_aggs_tenant_tracking;
266+
267+
-- drop hypertable_cagg_settings
268+
ALTER EXTENSION timescaledb DROP TABLE _timescaledb_catalog.hypertable_cagg_settings;
269+
DROP TABLE _timescaledb_catalog.hypertable_cagg_settings;

‎src/ts_catalog/CMakeLists.txt‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,9 @@ set(SOURCES
77
${CMAKE_CURRENT_SOURCE_DIR}/compression_settings.c
88
${CMAKE_CURRENT_SOURCE_DIR}/continuous_agg.c
99
${CMAKE_CURRENT_SOURCE_DIR}/continuous_aggs_jobs_refresh_ranges.c
10+
${CMAKE_CURRENT_SOURCE_DIR}/continuous_aggs_tenant_tracking.c
1011
${CMAKE_CURRENT_SOURCE_DIR}/continuous_aggs_watermark.c
12+
${CMAKE_CURRENT_SOURCE_DIR}/hypertable_cagg_settings.c
1113
${CMAKE_CURRENT_SOURCE_DIR}/metadata.c
1214
${CMAKE_CURRENT_SOURCE_DIR}/tablespace.c)
1315
target_sources(${PROJECT_NAME} PRIVATE ${SOURCES})

‎src/ts_catalog/catalog.c‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,14 @@ static const TableInfoDef catalog_table_names[_MAX_CATALOG_TABLES + 1] = {
9393
.schema_name = CATALOG_SCHEMA_NAME,
9494
.table_name = CONTINUOUS_AGGS_JOBS_REFRESH_RANGES_TABLE_NAME,
9595
},
96+
[CONTINUOUS_AGGS_TENANT_TRACKING] = {
97+
.schema_name = CATALOG_SCHEMA_NAME,
98+
.table_name = CONTINUOUS_AGGS_TENANT_TRACKING_TABLE_NAME,
99+
},
100+
[HYPERTABLE_CAGG_SETTINGS] = {
101+
.schema_name = CATALOG_SCHEMA_NAME,
102+
.table_name = HYPERTABLE_CAGG_SETTINGS_TABLE_NAME,
103+
},
96104
[COMPRESSION_SETTINGS] = {
97105
.schema_name = CATALOG_SCHEMA_NAME,
98106
.table_name = COMPRESSION_SETTINGS_TABLE_NAME,
@@ -242,6 +250,18 @@ static const TableIndexDef catalog_table_index_definitions[_MAX_CATALOG_TABLES]
242250
[CONTINUOUS_AGGS_JOBS_REFRESH_RANGES_IDX] = "continuous_aggs_jobs_refresh_ranges_idx",
243251
},
244252
},
253+
[CONTINUOUS_AGGS_TENANT_TRACKING] = {
254+
.length = _MAX_CONTINUOUS_AGGS_TENANT_TRACKING_INDEX,
255+
.names = (char *[]) {
256+
[CONTINUOUS_AGGS_TENANT_TRACKING_IDX] = "continuous_aggs_tenant_tracking_idx",
257+
},
258+
},
259+
[HYPERTABLE_CAGG_SETTINGS] = {
260+
.length = _MAX_HYPERTABLE_CAGG_SETTINGS_INDEX,
261+
.names = (char *[]) {
262+
[HYPERTABLE_CAGG_SETTINGS_PKEY] = "hypertable_cagg_settings_pkey",
263+
},
264+
},
245265
[CONTINUOUS_AGGS_WATERMARK] = {
246266
.length = _MAX_CONTINUOUS_AGGS_WATERMARK_INDEX,
247267
.names = (char *[]) {

‎src/ts_catalog/catalog.h‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,8 @@ typedef enum CatalogTable
4949
CONTINUOUS_AGGS_MATERIALIZATION_INVALIDATION_LOG,
5050
CONTINUOUS_AGGS_MATERIALIZATION_RANGES,
5151
CONTINUOUS_AGGS_JOBS_REFRESH_RANGES,
52+
CONTINUOUS_AGGS_TENANT_TRACKING,
53+
HYPERTABLE_CAGG_SETTINGS,
5254
COMPRESSION_SETTINGS,
5355
COMPRESSION_CHUNK_SIZE,
5456
CONTINUOUS_AGGS_BUCKET_FUNCTION,
@@ -1080,6 +1082,84 @@ enum
10801082
_MAX_CONTINUOUS_AGGS_JOBS_REFRESH_RANGES_INDEX,
10811083
};
10821084

1085+
/****** CONTINUOUS_AGGS_TENANT_TRACKING definitions */
1086+
#define CONTINUOUS_AGGS_TENANT_TRACKING_TABLE_NAME "continuous_aggs_tenant_tracking"
1087+
1088+
typedef enum Anum_continuous_aggs_tenant_tracking
1089+
{
1090+
Anum_continuous_aggs_tenant_tracking_hypertable_id = 1,
1091+
Anum_continuous_aggs_tenant_tracking_tenant_id,
1092+
Anum_continuous_aggs_tenant_tracking_min_timestamp,
1093+
Anum_continuous_aggs_tenant_tracking_max_timestamp,
1094+
Anum_continuous_aggs_tenant_tracking_seqnum,
1095+
_Anum_continuous_aggs_tenant_tracking_max,
1096+
} Anum_continuous_aggs_tenant_tracking;
1097+
1098+
#define Natts_continuous_aggs_tenant_tracking (_Anum_continuous_aggs_tenant_tracking_max - 1)
1099+
1100+
typedef struct FormData_continuous_aggs_tenant_tracking
1101+
{
1102+
int32 hypertable_id;
1103+
text *tenant_id;
1104+
int64 min_timestamp;
1105+
int64 max_timestamp;
1106+
int32 seqnum;
1107+
} FormData_continuous_aggs_tenant_tracking;
1108+
1109+
typedef FormData_continuous_aggs_tenant_tracking *Form_continuous_aggs_tenant_tracking;
1110+
1111+
enum
1112+
{
1113+
CONTINUOUS_AGGS_TENANT_TRACKING_IDX = 0,
1114+
_MAX_CONTINUOUS_AGGS_TENANT_TRACKING_INDEX,
1115+
};
1116+
typedef enum Anum_continuous_aggs_tenant_tracking_idx
1117+
{
1118+
Anum_continuous_aggs_tenant_tracking_idx_hypertable_id = 1,
1119+
Anum_continuous_aggs_tenant_tracking_idx_seqnum,
1120+
_Anum_continuous_aggs_tenant_tracking_idx_max,
1121+
} Anum_continuous_aggs_tenant_tracking_idx;
1122+
1123+
#define Natts_continuous_aggs_tenant_tracking_idx \
1124+
(_Anum_continuous_aggs_tenant_tracking_idx_max - 1)
1125+
1126+
/****** HYPERTABLE_CAGG_SETTINGS definitions */
1127+
#define HYPERTABLE_CAGG_SETTINGS_TABLE_NAME "hypertable_cagg_settings"
1128+
1129+
typedef enum Anum_hypertable_cagg_settings
1130+
{
1131+
Anum_hypertable_cagg_settings_hypertable_id = 1,
1132+
Anum_hypertable_cagg_settings_granular_refresh_column,
1133+
Anum_hypertable_cagg_settings_granular_refresh_start_offset,
1134+
Anum_hypertable_cagg_settings_granular_refresh_end_offset,
1135+
_Anum_hypertable_cagg_settings_max,
1136+
} Anum_hypertable_cagg_settings;
1137+
1138+
#define Natts_hypertable_cagg_settings (_Anum_hypertable_cagg_settings_max - 1)
1139+
1140+
typedef struct FormData_hypertable_cagg_settings
1141+
{
1142+
int32 hypertable_id;
1143+
NameData granular_refresh_column;
1144+
text *granular_refresh_start_offset;
1145+
text *granular_refresh_end_offset;
1146+
} FormData_hypertable_cagg_settings;
1147+
1148+
typedef FormData_hypertable_cagg_settings *Form_hypertable_cagg_settings;
1149+
1150+
enum
1151+
{
1152+
HYPERTABLE_CAGG_SETTINGS_PKEY = 0,
1153+
_MAX_HYPERTABLE_CAGG_SETTINGS_INDEX,
1154+
};
1155+
typedef enum Anum_hypertable_cagg_settings_pkey
1156+
{
1157+
Anum_hypertable_cagg_settings_pkey_hypertable_id = 1,
1158+
_Anum_hypertable_cagg_settings_pkey_max,
1159+
} Anum_hypertable_cagg_settings_pkey;
1160+
1161+
#define Natts_hypertable_cagg_settings_pkey (_Anum_hypertable_cagg_settings_pkey_max - 1)
1162+
10831163
typedef enum Anum_continuous_aggs_jobs_refresh_ranges_idx
10841164
{
10851165
Anum_continuous_aggs_jobs_refresh_ranges_idx_materialization_id = 1,

‎src/ts_catalog/continuous_aggs_tenant_tracking.c‎

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
/*
2+
* This file and its contents are licensed under the Apache License 2.0.
3+
* Please see the included NOTICE for copyright information and
4+
* LICENSE-APACHE for a copy of the license.
5+
*/
6+
7+
#include <postgres.h>
8+
#include <access/table.h>
9+
#include <utils/rel.h>
10+
11+
#include "scan_iterator.h"
12+
#include "ts_catalog/catalog.h"
13+
#include "ts_catalog/continuous_aggs_tenant_tracking.h"
14+
15+
static void
16+
init_scan_by_hypertable_id(ScanIterator *iterator, const int32 hypertable_id)
17+
{
18+
iterator->ctx.index = catalog_get_index(ts_catalog_get(),
19+
CONTINUOUS_AGGS_TENANT_TRACKING,
20+
CONTINUOUS_AGGS_TENANT_TRACKING_IDX);
21+
22+
ts_scan_iterator_scan_key_init(iterator,
23+
Anum_continuous_aggs_tenant_tracking_idx_hypertable_id,
24+
BTEqualStrategyNumber,
25+
F_INT4EQ,
26+
Int32GetDatum(hypertable_id));
27+
}
28+
29+
/*
30+
* Insert one tenant tracking row. tenant_id is copied into a bytea holding the
31+
* exact tenant key bytes.
32+
*/
33+
TSDLLEXPORT void
34+
ts_cagg_tenant_tracking_insert(int32 hypertable_id, const char *tenant_id, int tenant_id_len,
35+
int64 min_timestamp, int64 max_timestamp, int32 seqnum)
36+
{
37+
Catalog *catalog = ts_catalog_get();
38+
Relation rel = table_open(catalog_get_table_id(catalog, CONTINUOUS_AGGS_TENANT_TRACKING),
39+
RowExclusiveLock);
40+
TupleDesc desc = RelationGetDescr(rel);
41+
Datum values[Natts_continuous_aggs_tenant_tracking];
42+
bool nulls[Natts_continuous_aggs_tenant_tracking] = { false };
43+
CatalogSecurityContext sec_ctx;
44+
bytea *tenant_bytea = (bytea *) palloc(VARHDRSZ + tenant_id_len);
45+
46+
SET_VARSIZE(tenant_bytea, VARHDRSZ + tenant_id_len);
47+
memcpy(VARDATA(tenant_bytea), tenant_id, tenant_id_len);
48+
49+
values[AttrNumberGetAttrOffset(Anum_continuous_aggs_tenant_tracking_hypertable_id)] =
50+
Int32GetDatum(hypertable_id);
51+
values[AttrNumberGetAttrOffset(Anum_continuous_aggs_tenant_tracking_tenant_id)] =
52+
PointerGetDatum(tenant_bytea);
53+
values[AttrNumberGetAttrOffset(Anum_continuous_aggs_tenant_tracking_min_timestamp)] =
54+
Int64GetDatum(min_timestamp);
55+
values[AttrNumberGetAttrOffset(Anum_continuous_aggs_tenant_tracking_max_timestamp)] =
56+
Int64GetDatum(max_timestamp);
57+
values[AttrNumberGetAttrOffset(Anum_continuous_aggs_tenant_tracking_seqnum)] =
58+
Int32GetDatum(seqnum);
59+
60+
ts_catalog_database_info_become_owner(ts_catalog_database_info_get(), &sec_ctx);
61+
ts_catalog_insert_values(rel, desc, values, nulls);
62+
ts_catalog_restore_user(&sec_ctx);
63+
64+
table_close(rel, NoLock);
65+
}
66+
67+
/*
68+
* Insert the "invalid" marker row <null, null, null, seqnum>, signalling that
69+
* tenant tracking for this seqnum is incomplete and the refresh must fall back
70+
* to the full invalidation log.
71+
*/
72+
TSDLLEXPORT void
73+
ts_cagg_tenant_tracking_insert_invalid_marker(int32 hypertable_id, int32 seqnum)
74+
{
75+
Catalog *catalog = ts_catalog_get();
76+
Relation rel = table_open(catalog_get_table_id(catalog, CONTINUOUS_AGGS_TENANT_TRACKING),
77+
RowExclusiveLock);
78+
TupleDesc desc = RelationGetDescr(rel);
79+
Datum values[Natts_continuous_aggs_tenant_tracking] = { 0 };
80+
bool nulls[Natts_continuous_aggs_tenant_tracking] = { false };
81+
CatalogSecurityContext sec_ctx;
82+
83+
values[AttrNumberGetAttrOffset(Anum_continuous_aggs_tenant_tracking_hypertable_id)] =
84+
Int32GetDatum(hypertable_id);
85+
nulls[AttrNumberGetAttrOffset(Anum_continuous_aggs_tenant_tracking_tenant_id)] = true;
86+
nulls[AttrNumberGetAttrOffset(Anum_continuous_aggs_tenant_tracking_min_timestamp)] = true;
87+
nulls[AttrNumberGetAttrOffset(Anum_continuous_aggs_tenant_tracking_max_timestamp)] = true;
88+
values[AttrNumberGetAttrOffset(Anum_continuous_aggs_tenant_tracking_seqnum)] =
89+
Int32GetDatum(seqnum);
90+
91+
ts_catalog_database_info_become_owner(ts_catalog_database_info_get(), &sec_ctx);
92+
ts_catalog_insert_values(rel, desc, values, nulls);
93+
ts_catalog_restore_user(&sec_ctx);
94+
95+
table_close(rel, NoLock);
96+
}
97+
98+
/* Delete all tracking rows for a hypertable. */
99+
TSDLLEXPORT void
100+
ts_cagg_tenant_tracking_delete_by_hypertable_id(int32 hypertable_id)
101+
{
102+
ScanIterator iterator = ts_scan_iterator_create(CONTINUOUS_AGGS_TENANT_TRACKING,
103+
RowExclusiveLock,
104+
CurrentMemoryContext);
105+
106+
init_scan_by_hypertable_id(&iterator, hypertable_id);
107+
108+
ts_scanner_foreach(&iterator)
109+
{
110+
TupleInfo *ti = ts_scan_iterator_tuple_info(&iterator);
111+
112+
ts_catalog_delete_tid(ti->scanrel, ts_scanner_get_tuple_tid(ti));
113+
}
114+
ts_scan_iterator_close(&iterator);
115+
}

‎src/ts_catalog/continuous_aggs_tenant_tracking.h‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
/*
2+
* This file and its contents are licensed under the Apache License 2.0.
3+
* Please see the included NOTICE for copyright information and
4+
* LICENSE-APACHE for a copy of the license.
5+
*/
6+
#pragma once
7+
8+
#include <postgres.h>
9+
10+
#include "export.h"
11+
12+
/*
13+
* Catalog access for _timescaledb_catalog.continuous_aggs_tenant_tracking.
14+
*
15+
* Rows are flushed here from the shared-memory per-tenant invalidation tracker
16+
* during a continuous aggregate refresh (Transaction 2). tenant_id holds the
17+
* exact tenant key bytes; min/max_timestamp are internal time values. The
18+
* "invalid marker" row (tenant_id/min/max NULL) signals that tenant tracking
19+
* for that seqnum is incomplete, forcing a fall back to the full invalidation
20+
* log.
21+
*/
22+
23+
extern TSDLLEXPORT void ts_cagg_tenant_tracking_insert(int32 hypertable_id, const char *tenant_id,
24+
int tenant_id_len, int64 min_timestamp,
25+
int64 max_timestamp, int32 seqnum);
26+
27+
extern TSDLLEXPORT void ts_cagg_tenant_tracking_insert_invalid_marker(int32 hypertable_id,
28+
int32 seqnum);
29+
30+
extern TSDLLEXPORT void ts_cagg_tenant_tracking_delete_by_hypertable_id(int32 hypertable_id);

0 commit comments

Comments
 (0)