Skip to content

Commit 2db2db7

Browse files
melihmutlumelihmutlu
committed
test fixes
1 parent 096616e commit 2db2db7

34 files changed

Lines changed: 1944 additions & 1066 deletions

‎tsl/src/continuous_aggs/common.c‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2102,3 +2102,21 @@ cagg_find_groupingcols(ContinuousAgg *agg, Hypertable *mat_ht)
21022102
}
21032103
return retlist;
21042104
}
2105+
2106+
void
2107+
emit_up_to_date_notice(const ContinuousAgg *cagg, const ContinuousAggRefreshContext context)
2108+
{
2109+
switch (context.callctx)
2110+
{
2111+
case CAGG_REFRESH_WINDOW:
2112+
case CAGG_REFRESH_CREATION:
2113+
case CAGG_REFRESH_WINDOW_BATCHED:
2114+
elog(NOTICE,
2115+
"continuous aggregate \"%s\" is already up-to-date",
2116+
NameStr(cagg->data.user_view_name));
2117+
break;
2118+
case CAGG_REFRESH_POLICY:
2119+
case CAGG_REFRESH_POLICY_BATCHED:
2120+
break;
2121+
}
2122+
}

‎tsl/src/continuous_aggs/common.h‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@
3838

3939
#define DEFAULT_MATPARTCOLUMN_NAME "time_partition_col"
4040
#define CAGG_INVALIDATION_THRESHOLD_NAME "invalidation threshold watermark"
41-
#define CAGG_INVALIDATION_WRONG_GREATEST_VALUE ((int64) - 210866803200000001)
41+
#define CAGG_INVALIDATION_WRONG_GREATEST_VALUE ((int64) -210866803200000001)
4242

4343
typedef struct FinalizeQueryInfo
4444
{
@@ -179,3 +179,5 @@ extern bool caggtimebucket_validate_common(ContinuousAggBucketFunction *bf, List
179179
List *targetList, List *rtable, int ht_partcolno,
180180
StringInfo msg, bool is_cagg_create,
181181
const bool for_rewrites);
182+
extern void emit_up_to_date_notice(const ContinuousAgg *cagg,
183+
const ContinuousAggRefreshContext context);

‎tsl/src/continuous_aggs/create.c‎

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -857,6 +857,7 @@ tsl_process_continuous_agg_viewstmt(Node *node, const char *query_string, void *
857857

858858
if (!stmt->into->skipData)
859859
{
860+
bool refreshed = false;
860861
InternalTimeRange refresh_window = {
861862
.type = InvalidOid,
862863
};
@@ -887,13 +888,17 @@ tsl_process_continuous_agg_viewstmt(Node *node, const char *query_string, void *
887888
refresh_window.end = ts_time_get_noend_or_max(refresh_window.type);
888889

889890
ContinuousAggRefreshContext context = { .callctx = CAGG_REFRESH_CREATION };
890-
continuous_agg_refresh_internal(cagg,
891-
&refresh_window,
892-
context,
893-
true, /* start_isnull */
894-
true, /* end_isnull */
895-
true, /* bucketing_refresh_window */
896-
false /*extend_last_bucket*/);
891+
refreshed = continuous_agg_refresh_internal(cagg,
892+
&refresh_window,
893+
context,
894+
true, /* start_isnull */
895+
true, /* end_isnull */
896+
true, /* bucketing_refresh_window */
897+
false /*extend_last_bucket*/);
898+
if (!refreshed)
899+
{
900+
emit_up_to_date_notice(cagg, context);
901+
}
897902
}
898903

899904
return DDL_DONE;

‎tsl/src/continuous_aggs/refresh.c‎

Lines changed: 17 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -80,8 +80,6 @@ static void continuous_agg_refresh_with_window(const ContinuousAgg *cagg,
8080
const InvalidationStore *invalidations,
8181
const ContinuousAggRefreshContext context,
8282
bool bucketing_refresh_window);
83-
static void emit_up_to_date_notice(const ContinuousAgg *cagg,
84-
const ContinuousAggRefreshContext context);
8583
static bool process_cagg_invalidations_and_refresh(const ContinuousAgg *cagg,
8684
const InternalTimeRange *refresh_window,
8785
const ContinuousAggRefreshContext context,
@@ -682,6 +680,7 @@ continuous_agg_refresh_batched(ContinuousAgg *cagg, InternalTimeRange *refresh_w
682680
int32 batch_start = context.refresh_newest_first ? nbatches - 1 : 0;
683681
int32 batch_end = context.refresh_newest_first ? -1 : nbatches;
684682
int32 batch_step = context.refresh_newest_first ? -1 : 1;
683+
bool any_refreshed = false;
685684
for (int32 batch_idx = batch_start; batch_idx != batch_end; batch_idx += batch_step)
686685
{
687686
InternalTimeRange *batch_window =
@@ -704,13 +703,13 @@ continuous_agg_refresh_batched(ContinuousAgg *cagg, InternalTimeRange *refresh_w
704703
(context.refresh_newest_first ? processing_batch == 1 :
705704
processing_batch == context.number_of_batches);
706705

707-
continuous_agg_refresh_internal(cagg,
708-
batch_window,
709-
context,
710-
batch_window->start_isnull,
711-
batch_window->end_isnull,
712-
!batched, /* bucketing_refresh_window */
713-
apply_extend);
706+
any_refreshed |= continuous_agg_refresh_internal(cagg,
707+
batch_window,
708+
context,
709+
batch_window->start_isnull,
710+
batch_window->end_isnull,
711+
!batched, /* bucketing_refresh_window */
712+
apply_extend);
714713
DEBUG_ERROR_INJECTION(psprintf("cagg_policy_batch_%d_after_refresh", processing_batch));
715714

716715
if (context.max_batches_per_execution > 0 &&
@@ -724,6 +723,11 @@ continuous_agg_refresh_batched(ContinuousAgg *cagg, InternalTimeRange *refresh_w
724723
break;
725724
}
726725
}
726+
727+
if (!any_refreshed)
728+
{
729+
emit_up_to_date_notice(cagg, context);
730+
}
727731
}
728732

729733
/*
@@ -839,24 +843,6 @@ continuous_agg_refresh(PG_FUNCTION_ARGS)
839843
PG_RETURN_VOID();
840844
}
841845

842-
static void
843-
emit_up_to_date_notice(const ContinuousAgg *cagg, const ContinuousAggRefreshContext context)
844-
{
845-
switch (context.callctx)
846-
{
847-
case CAGG_REFRESH_WINDOW:
848-
case CAGG_REFRESH_CREATION:
849-
elog(NOTICE,
850-
"continuous aggregate \"%s\" is already up-to-date",
851-
NameStr(cagg->data.user_view_name));
852-
break;
853-
case CAGG_REFRESH_WINDOW_BATCHED:
854-
case CAGG_REFRESH_POLICY:
855-
case CAGG_REFRESH_POLICY_BATCHED:
856-
break;
857-
}
858-
}
859-
860846
static bool
861847
process_cagg_invalidations_and_refresh(const ContinuousAgg *cagg,
862848
const InternalTimeRange *refresh_window,
@@ -1002,7 +988,7 @@ rollback_and_error(const ContinuousAgg *cagg, CaggRefreshSpiContext *cagg_spi_ct
1002988
ThrowErrorData(edata);
1003989
}
1004990

1005-
void
991+
bool
1006992
continuous_agg_refresh_internal(const ContinuousAgg *cagg_arg,
1007993
const InternalTimeRange *refresh_window_arg,
1008994
const ContinuousAggRefreshContext context, const bool start_isnull,
@@ -1211,11 +1197,7 @@ continuous_agg_refresh_internal(const ContinuousAgg *cagg_arg,
12111197
if (edata)
12121198
{
12131199
rollback_and_error(cagg, &cagg_spi_ctx, (ErrorData *) edata);
1214-
return;
1215-
}
1216-
if (!refreshed)
1217-
{
1218-
emit_up_to_date_notice(cagg, context);
1200+
return false;
12191201
}
12201202

12211203
cleanup_before_cagg_refresh_exit(cagg, &cagg_spi_ctx);
@@ -1226,6 +1208,8 @@ continuous_agg_refresh_internal(const ContinuousAgg *cagg_arg,
12261208
{
12271209
elog(ERROR, "SPI_finish failed: %s", SPI_result_code_string(rc));
12281210
}
1211+
1212+
return refreshed;
12291213
}
12301214

12311215
static void

‎tsl/src/continuous_aggs/refresh.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ extern Datum continuous_agg_refresh(PG_FUNCTION_ARGS);
2121
extern void continuous_agg_refresh_batched(ContinuousAgg *cagg, InternalTimeRange *refresh_window,
2222
ContinuousAggRefreshContext context,
2323
bool extend_last_bucket);
24-
extern void continuous_agg_refresh_internal(const ContinuousAgg *cagg_arg,
24+
extern bool continuous_agg_refresh_internal(const ContinuousAgg *cagg_arg,
2525
const InternalTimeRange *refresh_window,
2626
const ContinuousAggRefreshContext context,
2727
const bool start_isnull, const bool end_isnull,

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1445,7 +1445,6 @@ insert into raw_data select '2000-05-01 00:00+0','Q3', 0, 0;
14451445
\set ON_ERROR_STOP 0
14461446
CALL refresh_continuous_aggregate('search_query_count_3', NULL, '2000-06-01 00:00+0'::timestamptz);
14471447
CALL refresh_continuous_aggregate('search_query_count_3', '2000-05-01 00:00+0'::timestamptz, '2000-06-01 00:00+0'::timestamptz);
1448-
NOTICE: continuous aggregate "search_query_count_3" is already up-to-date
14491448
\set ON_ERROR_STOP 1
14501449
--insert row
14511450
insert into raw_data select '2001-05-10 00:00+0','Q3', 100, 100;
@@ -1471,8 +1470,9 @@ WHERE materialization_id = :'MAT_HTID' ORDER BY 1, 2,3;
14711470
materialization_id | lowest_modified_value | greatest_modified_value
14721471
--------------------+-----------------------+-------------------------
14731472
41 | -9223372036854775808 | -210866803200000001
1474-
41 | 959817600000000 | 988675199999999
1475-
41 | 991353600000000 | 9223372036854775807
1473+
41 | 947376060000000 | 956447999999999
1474+
41 | 959817600000000 | 988847999999999
1475+
41 | 990144000000000 | 9223372036854775807
14761476

14771477
SELECT * from search_query_count_3
14781478
WHERE bucket > '2001-01-01'

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1445,7 +1445,6 @@ insert into raw_data select '2000-05-01 00:00+0','Q3', 0, 0;
14451445
\set ON_ERROR_STOP 0
14461446
CALL refresh_continuous_aggregate('search_query_count_3', NULL, '2000-06-01 00:00+0'::timestamptz);
14471447
CALL refresh_continuous_aggregate('search_query_count_3', '2000-05-01 00:00+0'::timestamptz, '2000-06-01 00:00+0'::timestamptz);
1448-
NOTICE: continuous aggregate "search_query_count_3" is already up-to-date
14491448
\set ON_ERROR_STOP 1
14501449
--insert row
14511450
insert into raw_data select '2001-05-10 00:00+0','Q3', 100, 100;
@@ -1471,8 +1470,9 @@ WHERE materialization_id = :'MAT_HTID' ORDER BY 1, 2,3;
14711470
materialization_id | lowest_modified_value | greatest_modified_value
14721471
--------------------+-----------------------+-------------------------
14731472
41 | -9223372036854775808 | -210866803200000001
1474-
41 | 959817600000000 | 988675199999999
1475-
41 | 991353600000000 | 9223372036854775807
1473+
41 | 947376060000000 | 956447999999999
1474+
41 | 959817600000000 | 988847999999999
1475+
41 | 990144000000000 | 9223372036854775807
14761476

14771477
SELECT * from search_query_count_3
14781478
WHERE bucket > '2001-01-01'

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1445,7 +1445,6 @@ insert into raw_data select '2000-05-01 00:00+0','Q3', 0, 0;
14451445
\set ON_ERROR_STOP 0
14461446
CALL refresh_continuous_aggregate('search_query_count_3', NULL, '2000-06-01 00:00+0'::timestamptz);
14471447
CALL refresh_continuous_aggregate('search_query_count_3', '2000-05-01 00:00+0'::timestamptz, '2000-06-01 00:00+0'::timestamptz);
1448-
NOTICE: continuous aggregate "search_query_count_3" is already up-to-date
14491448
\set ON_ERROR_STOP 1
14501449
--insert row
14511450
insert into raw_data select '2001-05-10 00:00+0','Q3', 100, 100;
@@ -1471,8 +1470,9 @@ WHERE materialization_id = :'MAT_HTID' ORDER BY 1, 2,3;
14711470
materialization_id | lowest_modified_value | greatest_modified_value
14721471
--------------------+-----------------------+-------------------------
14731472
41 | -9223372036854775808 | -210866803200000001
1474-
41 | 959817600000000 | 988675199999999
1475-
41 | 991353600000000 | 9223372036854775807
1473+
41 | 947376060000000 | 956447999999999
1474+
41 | 959817600000000 | 988847999999999
1475+
41 | 990144000000000 | 9223372036854775807
14761476

14771477
SELECT * from search_query_count_3
14781478
WHERE bucket > '2001-01-01'

‎tsl/test/expected/cagg-18.out‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1445,7 +1445,6 @@ insert into raw_data select '2000-05-01 00:00+0','Q3', 0, 0;
14451445
\set ON_ERROR_STOP 0
14461446
CALL refresh_continuous_aggregate('search_query_count_3', NULL, '2000-06-01 00:00+0'::timestamptz);
14471447
CALL refresh_continuous_aggregate('search_query_count_3', '2000-05-01 00:00+0'::timestamptz, '2000-06-01 00:00+0'::timestamptz);
1448-
NOTICE: continuous aggregate "search_query_count_3" is already up-to-date
14491448
\set ON_ERROR_STOP 1
14501449
--insert row
14511450
insert into raw_data select '2001-05-10 00:00+0','Q3', 100, 100;
@@ -1471,8 +1470,9 @@ WHERE materialization_id = :'MAT_HTID' ORDER BY 1, 2,3;
14711470
materialization_id | lowest_modified_value | greatest_modified_value
14721471
--------------------+-----------------------+-------------------------
14731472
41 | -9223372036854775808 | -210866803200000001
1474-
41 | 959817600000000 | 988675199999999
1475-
41 | 991353600000000 | 9223372036854775807
1473+
41 | 947376060000000 | 956447999999999
1474+
41 | 959817600000000 | 988847999999999
1475+
41 | 990144000000000 | 9223372036854775807
14761476

14771477
SELECT * from search_query_count_3
14781478
WHERE bucket > '2001-01-01'

‎tsl/test/expected/cagg_direct_compress.out‎

Lines changed: 12 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,8 @@ SELECT
2626
FROM conditions
2727
GROUP BY 1, 2
2828
WITH NO DATA;
29-
CALL refresh_continuous_aggregate('conditions_hourly', NULL, NULL);
29+
-- Setting buckets_per_batch to a high value to bypass "disabling direct compress because of too small batch size" situation
30+
CALL refresh_continuous_aggregate('conditions_hourly', NULL, NULL, options => '{"buckets_per_batch": 10000}'::jsonb);
3031
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('conditions_hourly') chunk;
3132
chunk_status_text
3233
-------------------
@@ -38,7 +39,7 @@ ALTER MATERIALIZED VIEW conditions_hourly SET (timescaledb.compress);
3839
NOTICE: defaulting compress_orderby to bucket,device_id
3940
-- Enable direct compress on cagg refresh
4041
SET timescaledb.enable_direct_compress_on_cagg_refresh TO on;
41-
CALL refresh_continuous_aggregate('conditions_hourly', NULL, NULL);
42+
CALL refresh_continuous_aggregate('conditions_hourly', NULL, NULL, options => '{"buckets_per_batch": 10000}'::jsonb);
4243
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('conditions_hourly') chunk;
4344
chunk_status_text
4445
-------------------
@@ -48,7 +49,7 @@ SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks
4849
INSERT INTO conditions
4950
SELECT t, d::text, 1, 1 FROM generate_series('2025-12-15 00:00:00+00'::timestamptz - interval '1 year', '2025-12-15 00:00:00+00'::timestamptz, interval '1 hour') AS t, generate_series(1, 10) AS d;
5051
SET timescaledb.enable_direct_compress_on_cagg_refresh TO off;
51-
CALL refresh_continuous_aggregate('conditions_hourly', NULL, NULL);
52+
CALL refresh_continuous_aggregate('conditions_hourly', NULL, NULL, options => '{"buckets_per_batch": 10000}'::jsonb);
5253
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('conditions_hourly') chunk;
5354
chunk_status_text
5455
----------------------
@@ -71,7 +72,7 @@ SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks
7172
INSERT INTO conditions
7273
SELECT t, d::text, 1, 1 FROM generate_series('2025-12-15 00:00:00+00'::timestamptz - interval '1 year', '2025-12-15 00:00:00+00'::timestamptz, interval '1 hour') AS t, generate_series(1, 10) AS d;
7374
SET timescaledb.enable_direct_compress_on_cagg_refresh TO on;
74-
CALL refresh_continuous_aggregate('conditions_hourly', NULL, NULL);
75+
CALL refresh_continuous_aggregate('conditions_hourly', NULL, NULL, options => '{"buckets_per_batch": 10000}'::jsonb);
7576
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('conditions_hourly') chunk;
7677
chunk_status_text
7778
-------------------
@@ -98,14 +99,14 @@ INSERT INTO conditions
9899
SELECT t, d::text, 1, 1 FROM generate_series('2025-12-15 00:00:00+00'::timestamptz - interval '1 year', '2025-12-15 00:00:00+00'::timestamptz, interval '1 hour') AS t, generate_series(1, 10) AS d;
99100
SET timescaledb.enable_direct_compress_on_cagg_refresh TO on;
100101
-- Refresh the base CAgg
101-
CALL refresh_continuous_aggregate('conditions_hourly', NULL, NULL);
102+
CALL refresh_continuous_aggregate('conditions_hourly', NULL, NULL, options => '{"buckets_per_batch": 10000}'::jsonb);
102103
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('conditions_hourly') chunk;
103104
chunk_status_text
104105
-------------------
105106
{COMPRESSED}
106107

107108
-- Refresh the hierarchical CAgg
108-
CALL refresh_continuous_aggregate('conditions_daily', NULL, NULL);
109+
CALL refresh_continuous_aggregate('conditions_daily', NULL, NULL, options => '{"buckets_per_batch": 10000}'::jsonb);
109110
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('conditions_daily') chunk;
110111
chunk_status_text
111112
-------------------
@@ -115,24 +116,24 @@ SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks
115116
INSERT INTO conditions
116117
SELECT t, d::text, 1, 1 FROM generate_series('2025-12-15 00:00:00+00'::timestamptz - interval '1 year', '2025-12-15 00:00:00+00'::timestamptz, interval '1 hour') AS t, generate_series(1, 10) AS d;
117118
-- Refresh the base CAgg
118-
CALL refresh_continuous_aggregate('conditions_hourly', NULL, NULL);
119+
CALL refresh_continuous_aggregate('conditions_hourly', NULL, NULL, options => '{"buckets_per_batch": 10000}'::jsonb);
119120
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('conditions_hourly') chunk;
120121
chunk_status_text
121122
-------------------
122123
{COMPRESSED}
123124

124125
-- Refreshing again the base CAgg is a no-op since everything is up to date
125-
CALL refresh_continuous_aggregate('conditions_hourly', NULL, NULL);
126+
CALL refresh_continuous_aggregate('conditions_hourly', NULL, NULL, options => '{"buckets_per_batch": 10000}'::jsonb);
126127
NOTICE: continuous aggregate "conditions_hourly" is already up-to-date
127128
-- Refresh the hierarchical CAgg with invalidations procuded by the base CAgg
128-
CALL refresh_continuous_aggregate('conditions_daily', NULL, NULL);
129+
CALL refresh_continuous_aggregate('conditions_daily', NULL, NULL, options => '{"buckets_per_batch": 10000}'::jsonb);
129130
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('conditions_daily') chunk;
130131
chunk_status_text
131132
-------------------
132133
{COMPRESSED}
133134

134135
-- Refreshing again the hierarchical CAgg is a no-op since everything is up to date
135-
CALL refresh_continuous_aggregate('conditions_daily', NULL, NULL);
136+
CALL refresh_continuous_aggregate('conditions_daily', NULL, NULL, options => '{"buckets_per_batch": 10000}'::jsonb);
136137
NOTICE: continuous aggregate "conditions_daily" is already up-to-date
137138
-- Tests with custom segmentby and orderby
138139
CREATE MATERIALIZED VIEW conditions_weekly
@@ -148,7 +149,7 @@ FROM conditions
148149
GROUP BY 1, 2, 3
149150
WITH NO DATA;
150151
ALTER MATERIALIZED VIEW conditions_weekly SET (timescaledb.compress_segmentby = 'device_id, location_id', timescaledb.compress_orderby = 'max, min, bucket DESC');
151-
CALL refresh_continuous_aggregate('conditions_weekly', NULL, NULL);
152+
CALL refresh_continuous_aggregate('conditions_weekly', NULL, NULL, options => '{"buckets_per_batch": 10000}'::jsonb);
152153
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('conditions_weekly') chunk;
153154
chunk_status_text
154155
-------------------

0 commit comments

Comments
 (0)