Skip to content

Commit ca6925d

Browse files
melihmutlupnthaogayyappan
authored
New catalogs for CAgg granular refresh (#10299)
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 83f61f2 commit ca6925d

13 files changed

Lines changed: 578 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 NOT NULL,
448+
max_timestamp bigint NOT NULL,
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 NOT NULL,
155+
max_timestamp bigint NOT NULL,
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,

0 commit comments

Comments
 (0)