Skip to content

Commit bd8fdce

Browse files
fabriziomellofabriziomello
committed
Enable Incremental CAgg Refresh Policy by default
In #7790 we introduced the Incremental CAgg Refresh Policy that is not enabled by default using the following values for the new options: * `buckets_per_batch=0`: single batch execution. * `max_batches_per_execution=10`: maximum of 10 batches per execution. This PR enable the incremental refresh without break the current semantics of process all buckets in a given refresh window by setting new default values: * `buckets_per_batch=1`: one bucket per batch. * `max_batches_per_execution=0`: unlimited number of batches to be processed.
1 parent 2f9c02c commit bd8fdce

13 files changed

Lines changed: 348 additions & 116 deletions

File tree

‎.unreleased/pr_8265‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Implements: #8265 Set incremental CAgg refresh policy on by default

‎tsl/src/bgw_policy/continuous_aggregate_api.c‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,12 @@
3232
/* Default max runtime for a continuous aggregate jobs is unlimited for now */
3333
#define DEFAULT_MAX_RUNTIME \
3434
DatumGetIntervalP(DirectFunctionCall3(interval_in, CStringGetDatum("0"), InvalidOid, -1))
35+
/* Default buckets per batch is 1, which means that the job will refresh 1 bucket at a time */
36+
#define DEFAULT_BUCKETS_PER_BATCH 1
37+
/* Default max batches per execution is 0, which means no limit */
38+
#define DEFAULT_MAX_BATCHES_PER_EXECUTION 0
39+
/* Default refresh newest first is true, which means from newest data to the oldest */
40+
#define DEFAULT_REFRESH_NEWEST_FIRST true
3541

3642
int32
3743
policy_continuous_aggregate_get_mat_hypertable_id(const Jsonb *config)
@@ -151,6 +157,9 @@ policy_refresh_cagg_get_buckets_per_batch(const Jsonb *config)
151157
bool found;
152158
int32 res = ts_jsonb_get_int32_field(config, POL_REFRESH_CONF_KEY_BUCKETS_PER_BATCH, &found);
153159

160+
if (!found)
161+
res = DEFAULT_BUCKETS_PER_BATCH; /* default value */
162+
154163
return res;
155164
}
156165

@@ -162,7 +171,7 @@ policy_refresh_cagg_get_max_batches_per_execution(const Jsonb *config)
162171
ts_jsonb_get_int32_field(config, POL_REFRESH_CONF_KEY_MAX_BATCHES_PER_EXECUTION, &found);
163172

164173
if (!found)
165-
res = 10; /* default value */
174+
res = DEFAULT_MAX_BATCHES_PER_EXECUTION; /* default value */
166175

167176
return res;
168177
}
@@ -174,7 +183,7 @@ policy_refresh_cagg_get_refresh_newest_first(const Jsonb *config)
174183
bool res = ts_jsonb_get_bool_field(config, POL_REFRESH_CONF_KEY_REFRESH_NEWEST_FIRST, &found);
175184

176185
if (!found)
177-
res = true; /* default value */
186+
res = DEFAULT_REFRESH_NEWEST_FIRST; /* default value */
178187

179188
return res;
180189
}

‎tsl/src/bgw_policy/job.c‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -425,7 +425,8 @@ policy_refresh_cagg_execute(int32 job_id, Jsonb *config)
425425
context,
426426
refresh_window->start_isnull,
427427
refresh_window->end_isnull,
428-
false,
428+
(context.callctx != CAGG_REFRESH_POLICY_BATCHED),
429+
false, /* force */
429430
policy_data.process_hypertable_invalidations);
430431
if (processing_batch >= policy_data.max_batches_per_execution &&
431432
processing_batch < context.number_of_batches &&

‎tsl/src/continuous_aggs/create.c‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -954,9 +954,10 @@ tsl_process_continuous_agg_viewstmt(Node *node, const char *query_string, void *
954954
continuous_agg_refresh_internal(cagg,
955955
&refresh_window,
956956
context,
957-
true,
958-
true,
959-
false,
957+
true, /* start_isnull */
958+
true, /* end_isnull */
959+
true, /* bucketing_refresh_window */
960+
false, /* force */
960961
true /* process_hypertable_invalidations */);
961962
}
962963

‎tsl/src/continuous_aggs/refresh.c‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -690,6 +690,7 @@ continuous_agg_refresh(PG_FUNCTION_ARGS)
690690
context,
691691
PG_ARGISNULL(1),
692692
PG_ARGISNULL(2),
693+
true,
693694
force,
694695
process_hypertable_invalidations);
695696

@@ -790,7 +791,7 @@ void
790791
continuous_agg_refresh_internal(const ContinuousAgg *cagg,
791792
const InternalTimeRange *refresh_window_arg,
792793
const CaggRefreshContext context, const bool start_isnull,
793-
const bool end_isnull, bool force,
794+
const bool end_isnull, bool bucketing_refresh_window, bool force,
794795
bool process_hypertable_invalidations)
795796
{
796797
int32 mat_id = cagg->data.mat_hypertable_id;
@@ -842,7 +843,7 @@ continuous_agg_refresh_internal(const ContinuousAgg *cagg,
842843
get_rel_name(cagg->relid));
843844

844845
/* No bucketing when open ended */
845-
if (!(start_isnull && end_isnull))
846+
if (bucketing_refresh_window && !(start_isnull && end_isnull))
846847
{
847848
if (cagg->bucket_function->bucket_fixed_interval == false)
848849
{

‎tsl/src/continuous_aggs/refresh.h‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,8 @@ extern void continuous_agg_refresh_internal(const ContinuousAgg *cagg,
2020
const InternalTimeRange *refresh_window,
2121
const CaggRefreshContext context,
2222
const bool start_isnull, const bool end_isnull,
23-
bool force, bool process_hypertable_invalidations);
23+
bool bucketing_refresh_window, bool force,
24+
bool process_hypertable_invalidations);
2425
extern List *continuous_agg_split_refresh_window(ContinuousAgg *cagg,
2526
InternalTimeRange *original_refresh_window,
2627
int32 buckets_per_batch,

‎tsl/test/expected/cagg_bgw-15.out‎

Lines changed: 105 additions & 33 deletions
Large diffs are not rendered by default.

‎tsl/test/expected/cagg_bgw-16.out‎

Lines changed: 105 additions & 33 deletions
Large diffs are not rendered by default.

‎tsl/test/expected/cagg_bgw-17.out‎

Lines changed: 105 additions & 33 deletions
Large diffs are not rendered by default.

‎tsl/test/expected/cagg_policy.out‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1121,7 +1121,7 @@ INSERT INTO bigint_tab VALUES(5);
11211121
INSERT INTO bigint_tab VALUES(10);
11221122
INSERT INTO bigint_tab VALUES(20);
11231123
CALL run_job(:job_mid);
1124-
SELECT * FROM mat_bigint;
1124+
SELECT * FROM mat_bigint ORDER BY 1;
11251125
a | countb
11261126
----+--------
11271127
5 | 1
@@ -1394,7 +1394,7 @@ ALTER materialized view deals_best_weekly set (timescaledb.materialized_only=tru
13941394
ALTER materialized view deals_best_daily set (timescaledb.materialized_only=true);
13951395
-- we have data from 6 weeks before to May 5 2022 (Thu)
13961396
CALL refresh_continuous_aggregate('deals_best_weekly', '2022-04-24', '2022-05-03');
1397-
SELECT * FROM deals_best_weekly;
1397+
SELECT * FROM deals_best_weekly ORDER BY bucket;
13981398
bucket | avg_temp | max_rating
13991399
------------------------------+------------------+------------
14001400
Sun Apr 24 17:00:00 2022 PDT | 117.764705882353 | 6
@@ -1411,7 +1411,7 @@ SELECT * FROM deals_best_daily ORDER BY bucket LIMIT 2;
14111411
-- expect to get an up-to-date notice
14121412
CALL refresh_continuous_aggregate('deals_best_weekly', '2022-04-24', '2022-05-05');
14131413
NOTICE: continuous aggregate "deals_best_weekly" is already up-to-date
1414-
SELECT * FROM deals_best_weekly;
1414+
SELECT * FROM deals_best_weekly ORDER BY bucket;
14151415
bucket | avg_temp | max_rating
14161416
------------------------------+------------------+------------
14171417
Sun Apr 24 17:00:00 2022 PDT | 117.764705882353 | 6

0 commit comments

Comments
 (0)