Skip to content

Backport to 2.28.x: #10242: Fix incremental refresh skipping the last bucket #10294

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
timescale-automation wants to merge 1 commit into
base: 2.28.x
Choose a base branch
from
Open

Backport to 2.28.x: #10242: Fix incremental refresh skipping the last bucket #10294

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .unreleased/pr_10242
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
Fixes: #10221 Fix incremental refresh skipping the last bucket
Thanks: @FrancescEthon and @ManuelEthon for reporting the issue
4 changes: 2 additions & 2 deletions tsl/src/continuous_aggs/refresh.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1519,11 +1519,11 @@ continuous_agg_split_refresh_window(ContinuousAgg *cagg, InternalTimeRange *orig
}
}

/* Advance past invalidations that end at or before cur.
/* Advance past invalidations that end before cur.
* This is correct because we are reading invalidation entries already ordered by
* lowest_modified_value.
*/
while (inval_idx < inval_count && inval_high <= cur)
while (inval_idx < inval_count && inval_high < cur)
{
inval_idx++;
if (inval_idx < inval_count)
Expand Down
64 changes: 64 additions & 0 deletions tsl/test/expected/cagg_policy_incremental.out
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1537,6 +1537,70 @@ NOTICE: drop cascades to 6 other objects
NOTICE: drop cascades to table _timescaledb_internal._hyper_8_98_chunk
NOTICE: drop cascades to 2 other objects
NOTICE: drop cascades to table _timescaledb_internal._hyper_10_101_chunk
------------------------------------------------------------------------------------------
-- #10221: policy-path variant. An invalidation whose greatest_modified_value sits
-- EXACTLY on a bucket start must not lose its last bucket when the policy refresh
-- window is split into batches. The policy has finite start/end offsets, so no
-- null-restoration can mask a missing batch.
------------------------------------------------------------------------------------------
CREATE TABLE pol_boundary (
time TIMESTAMPTZ NOT NULL,
device INTEGER,
value FLOAT
);
SELECT FROM create_hypertable('pol_boundary', 'time');
--

-- device 1: mid-bucket rows, 2025-03-01 .. 2025-03-08
INSERT INTO pol_boundary
SELECT time, 1, 1.0
FROM generate_series('2025-03-01 12:00:00+00'::timestamptz,
'2025-03-08 12:00:00+00'::timestamptz,
'1 day'::interval) AS time;
CREATE MATERIALIZED VIEW pol_boundary_cagg
WITH (timescaledb.continuous) AS
SELECT time_bucket('1 day', time) AS bucket, device, count(*) AS cnt
FROM pol_boundary
GROUP BY 1, 2
WITH NO DATA;
SELECT add_continuous_aggregate_policy('pol_boundary_cagg',
start_offset => INTERVAL '30 days',
end_offset => INTERVAL '1 day',
schedule_interval => INTERVAL '1 hour',
buckets_per_batch => 1
) AS pol_boundary_job_id \gset
-- First run materializes everything and advances the watermark to 2025-03-09
CALL run_job(:pol_boundary_job_id);
-- device 2: three rows below the watermark whose greatest value (2025-03-04
-- 00:00) lands exactly on a bucket start
INSERT INTO pol_boundary VALUES
('2025-03-02 12:00:00+00', 2, 1.0),
('2025-03-03 12:00:00+00', 2, 1.0),
('2025-03-04 00:00:00+00', 2, 1.0);
-- Second run must refresh all three device-2 buckets, including
-- [2025-03-04, 2025-03-05)
CALL run_job(:pol_boundary_job_id);
SELECT * FROM pol_boundary_cagg WHERE device = 2 ORDER BY bucket;
bucket | device | cnt
------------------------------+--------+-----
Sun Mar 02 00:00:00 2025 UTC | 2 | 1
Mon Mar 03 00:00:00 2025 UTC | 2 | 1
Tue Mar 04 00:00:00 2025 UTC | 2 | 1

-- No stranded invalidation may remain
SELECT _timescaledb_functions.to_timestamp(lowest_modified_value) AS stranded_low,
_timescaledb_functions.to_timestamp(greatest_modified_value) AS stranded_high
FROM _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
WHERE materialization_id = (SELECT mat_hypertable_id FROM _timescaledb_catalog.continuous_agg
WHERE user_view_name = 'pol_boundary_cagg')
AND lowest_modified_value > -210866803200000000
AND greatest_modified_value < 9223372036854775807;
stranded_low | stranded_high
--------------+---------------

DROP TABLE pol_boundary CASCADE;
NOTICE: drop cascades to 2 other objects
NOTICE: drop cascades to table _timescaledb_internal._hyper_18_113_chunk
RESET timezone;
\c :TEST_DBNAME :ROLE_SUPERUSER
REASSIGN OWNED BY test_cagg_refresh_policy_user TO :ROLE_SUPERUSER;
Expand Down
57 changes: 57 additions & 0 deletions tsl/test/expected/cagg_refresh_incremental.out
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -716,6 +716,63 @@ FROM
----------
f

------------------------------------------------------------------------------------------
-- #10221: an invalidation whose greatest_modified_value lands EXACTLY on a bucket
-- start must not lose its last bucket when the refresh window is split into batches.
------------------------------------------------------------------------------------------
SET timezone TO 'UTC';
CREATE TABLE boundary_data (
time TIMESTAMPTZ NOT NULL,
device INTEGER,
value FLOAT
);
SELECT FROM create_hypertable('boundary_data', 'time');
--

-- device 1: mid-bucket rows only, 2025-03-01 .. 2025-03-10
INSERT INTO boundary_data
SELECT time, 1, 1.0
FROM generate_series('2025-03-01 12:00:00+00'::timestamptz,
'2025-03-10 12:00:00+00'::timestamptz,
'1 day'::interval) AS time;
CREATE MATERIALIZED VIEW boundary_cagg
WITH (timescaledb.continuous) AS
SELECT time_bucket('1 day', time) AS bucket, device, count(*) AS cnt
FROM boundary_data
GROUP BY 1, 2
WITH NO DATA;
-- Full refresh advances the watermark to 2025-03-11
CALL refresh_continuous_aggregate('boundary_cagg', NULL, NULL);
-- device 2: single row EXACTLY at a bucket start below the watermark. The
-- hypertable invalidation entry for it is [2025-03-05, 2025-03-05] -- both
-- bounds precisely on the bucket boundary.
INSERT INTO boundary_data VALUES ('2025-03-05 00:00:00+00', 2, 1.0);
-- Batched refresh over a finite window: bucket [2025-03-05, 2025-03-06) must
-- get its own batch.
CALL refresh_continuous_aggregate(
'boundary_cagg', '2025-03-01'::timestamptz, '2025-04-01'::timestamptz,
options => '{"buckets_per_batch": 1}'::jsonb);
-- The 2025-03-05 bucket must be materialized for device 2
SELECT * FROM boundary_cagg WHERE device = 2 ORDER BY bucket;
bucket | device | cnt
------------------------------+--------+-----
Wed Mar 05 00:00:00 2025 UTC | 2 | 1

-- No stranded invalidation may remain
SELECT _timescaledb_functions.to_timestamp(lowest_modified_value) AS stranded_low,
_timescaledb_functions.to_timestamp(greatest_modified_value) AS stranded_high
FROM _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
WHERE materialization_id = (SELECT mat_hypertable_id FROM _timescaledb_catalog.continuous_agg
WHERE user_view_name = 'boundary_cagg')
AND lowest_modified_value > -210866803200000000
AND greatest_modified_value < 9223372036854775807;
stranded_low | stranded_high
--------------+---------------

DROP TABLE boundary_data CASCADE;
NOTICE: drop cascades to 2 other objects
NOTICE: drop cascades to table _timescaledb_internal._hyper_8_97_chunk
RESET timezone;
\c :TEST_DBNAME :ROLE_SUPERUSER
REASSIGN OWNED BY test_cagg_refresh_manual_user TO :ROLE_SUPERUSER;
REVOKE ALL ON SCHEMA public FROM test_cagg_refresh_manual_user;
Expand Down
61 changes: 61 additions & 0 deletions tsl/test/sql/cagg_policy_incremental.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -965,6 +965,67 @@ RESET client_min_messages;
DROP TABLE threshold_inc_data CASCADE;

DROP TABLE sparse_data CASCADE;

------------------------------------------------------------------------------------------
-- #10221: policy-path variant. An invalidation whose greatest_modified_value sits
-- EXACTLY on a bucket start must not lose its last bucket when the policy refresh
-- window is split into batches. The policy has finite start/end offsets, so no
-- null-restoration can mask a missing batch.
------------------------------------------------------------------------------------------
CREATE TABLE pol_boundary (
time TIMESTAMPTZ NOT NULL,
device INTEGER,
value FLOAT
);
SELECT FROM create_hypertable('pol_boundary', 'time');

-- device 1: mid-bucket rows, 2025-03-01 .. 2025-03-08
INSERT INTO pol_boundary
SELECT time, 1, 1.0
FROM generate_series('2025-03-01 12:00:00+00'::timestamptz,
'2025-03-08 12:00:00+00'::timestamptz,
'1 day'::interval) AS time;

CREATE MATERIALIZED VIEW pol_boundary_cagg
WITH (timescaledb.continuous) AS
SELECT time_bucket('1 day', time) AS bucket, device, count(*) AS cnt
FROM pol_boundary
GROUP BY 1, 2
WITH NO DATA;

SELECT add_continuous_aggregate_policy('pol_boundary_cagg',
start_offset => INTERVAL '30 days',
end_offset => INTERVAL '1 day',
schedule_interval => INTERVAL '1 hour',
buckets_per_batch => 1
) AS pol_boundary_job_id \gset

-- First run materializes everything and advances the watermark to 2025-03-09
CALL run_job(:pol_boundary_job_id);

-- device 2: three rows below the watermark whose greatest value (2025-03-04
-- 00:00) lands exactly on a bucket start
INSERT INTO pol_boundary VALUES
('2025-03-02 12:00:00+00', 2, 1.0),
('2025-03-03 12:00:00+00', 2, 1.0),
('2025-03-04 00:00:00+00', 2, 1.0);

-- Second run must refresh all three device-2 buckets, including
-- [2025-03-04, 2025-03-05)
CALL run_job(:pol_boundary_job_id);

SELECT * FROM pol_boundary_cagg WHERE device = 2 ORDER BY bucket;

-- No stranded invalidation may remain
SELECT _timescaledb_functions.to_timestamp(lowest_modified_value) AS stranded_low,
_timescaledb_functions.to_timestamp(greatest_modified_value) AS stranded_high
FROM _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
WHERE materialization_id = (SELECT mat_hypertable_id FROM _timescaledb_catalog.continuous_agg
WHERE user_view_name = 'pol_boundary_cagg')
AND lowest_modified_value > -210866803200000000
AND greatest_modified_value < 9223372036854775807;

DROP TABLE pol_boundary CASCADE;
RESET timezone;

\c :TEST_DBNAME :ROLE_SUPERUSER
Expand Down
56 changes: 56 additions & 0 deletions tsl/test/sql/cagg_refresh_incremental.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -484,6 +484,62 @@ FROM
EXCEPT
(SELECT * FROM conditions_by_day ORDER BY 1, 2)) AS diff;

------------------------------------------------------------------------------------------
-- #10221: an invalidation whose greatest_modified_value lands EXACTLY on a bucket
-- start must not lose its last bucket when the refresh window is split into batches.
------------------------------------------------------------------------------------------
SET timezone TO 'UTC';

CREATE TABLE boundary_data (
time TIMESTAMPTZ NOT NULL,
device INTEGER,
value FLOAT
);
SELECT FROM create_hypertable('boundary_data', 'time');

-- device 1: mid-bucket rows only, 2025-03-01 .. 2025-03-10
INSERT INTO boundary_data
SELECT time, 1, 1.0
FROM generate_series('2025-03-01 12:00:00+00'::timestamptz,
'2025-03-10 12:00:00+00'::timestamptz,
'1 day'::interval) AS time;

CREATE MATERIALIZED VIEW boundary_cagg
WITH (timescaledb.continuous) AS
SELECT time_bucket('1 day', time) AS bucket, device, count(*) AS cnt
FROM boundary_data
GROUP BY 1, 2
WITH NO DATA;

-- Full refresh advances the watermark to 2025-03-11
CALL refresh_continuous_aggregate('boundary_cagg', NULL, NULL);

-- device 2: single row EXACTLY at a bucket start below the watermark. The
-- hypertable invalidation entry for it is [2025-03-05, 2025-03-05] -- both
-- bounds precisely on the bucket boundary.
INSERT INTO boundary_data VALUES ('2025-03-05 00:00:00+00', 2, 1.0);

-- Batched refresh over a finite window: bucket [2025-03-05, 2025-03-06) must
-- get its own batch.
CALL refresh_continuous_aggregate(
'boundary_cagg', '2025-03-01'::timestamptz, '2025-04-01'::timestamptz,
options => '{"buckets_per_batch": 1}'::jsonb);

-- The 2025-03-05 bucket must be materialized for device 2
SELECT * FROM boundary_cagg WHERE device = 2 ORDER BY bucket;

-- No stranded invalidation may remain
SELECT _timescaledb_functions.to_timestamp(lowest_modified_value) AS stranded_low,
_timescaledb_functions.to_timestamp(greatest_modified_value) AS stranded_high
FROM _timescaledb_catalog.continuous_aggs_materialization_invalidation_log
WHERE materialization_id = (SELECT mat_hypertable_id FROM _timescaledb_catalog.continuous_agg
WHERE user_view_name = 'boundary_cagg')
AND lowest_modified_value > -210866803200000000
AND greatest_modified_value < 9223372036854775807;

DROP TABLE boundary_data CASCADE;
RESET timezone;

\c :TEST_DBNAME :ROLE_SUPERUSER
REASSIGN OWNED BY test_cagg_refresh_manual_user TO :ROLE_SUPERUSER;
REVOKE ALL ON SCHEMA public FROM test_cagg_refresh_manual_user;
Expand Down
Loading