Fix timestamp out of range in CAgg refresh policy

When setting a refresh policy with `end_offset=>NULL` for a CAgg with
variable sized bucket it was erroring out when there was o data to be
refreshed.

This was happening cause we're using the wrong util function to get the
maximum value for the given type. Fixed it using the proper function to
handle both the fixed and variable bucket size.

Author: fabriziomello

.unreleased/pr_8559

@@ -0,0 +1,2 @@
+Fixes: #8559 Fix `timestamp out of range` using `end_offset=NULL` on CAgg refresh policy
+Thanks: @nofalx for reporting the error when using `end_offset=NULL` on CAgg refresh policy

tsl/src/bgw_policy/continuous_aggregate_api.c

@@ -137,7 +137,7 @@ policy_refresh_cagg_get_refresh_end(const Dimension *dim, const Jsonb *config, b
 	int64 res = get_time_from_config(dim, config, POL_REFRESH_CONF_KEY_END_OFFSET, end_isnull);
 
 	if (*end_isnull)
-		return ts_time_get_end_or_max(ts_dimension_get_partition_type(dim));
+		return ts_time_get_noend_or_max(ts_dimension_get_partition_type(dim));
 	return res;
 }
 

tsl/test/expected/cagg_policy.out

@@ -1400,3 +1400,72 @@ SELECT timescaledb_experimental.add_policies('cagg');
  f
 (1 row)
 
+-- Issue #6902
+-- Fix timestamp out of range in a refresh policy when setting `end_offset=>NULL`
+-- for a CAgg with variable sized bucket (i.e: using `time_bucket` with timezone)
+CREATE TABLE issue_6902 (
+  ts  TIMESTAMPTZ NOT NULL,
+  temperature NUMERIC
+) WITH (
+  timescaledb.hypertable,
+  timescaledb.partition_column='ts',
+  timescaledb.chunk_interval='1 day',
+  timescaledb.compress='off'
+);
+INSERT INTO issue_6902
+SELECT t, 1 FROM generate_series(now() - interval '3 hours', now(), interval '1 minute') AS t;
+CREATE MATERIALIZED VIEW issue_6902_by_hour
+WITH (timescaledb.continuous) AS
+SELECT
+  time_bucket(INTERVAL '1 hour', ts, 'America/Sao_Paulo') AS bucket, -- using timezone
+  MAX(temperature),
+  MIN(temperature),
+  COUNT(*)
+FROM issue_6902
+GROUP BY 1
+WITH NO DATA;
+SELECT add_continuous_aggregate_policy (
+  'issue_6902_by_hour',
+  start_offset => INTERVAL '3 hours',
+  end_offset => NULL,
+  schedule_interval => INTERVAL '12 hour',
+  initial_start => now() + INTERVAL '12 hour'
+) AS job_id \gset
+-- 181 rows
+CALL run_job(:job_id);
+SELECT count(*) FROM issue_6902;
+ count 
+-------
+   181
+(1 row)
+
+-- run again without any change, remain 181 rows
+CALL run_job(:job_id);
+SELECT count(*) FROM issue_6902;
+ count 
+-------
+   181
+(1 row)
+
+-- change existing data
+UPDATE issue_6902
+SET temperature = temperature + 1;
+-- run again without any change, remain 181 rows
+CALL run_job(:job_id);
+SELECT count(*) FROM issue_6902;
+ count 
+-------
+   181
+(1 row)
+
+-- insert more data
+INSERT INTO issue_6902
+SELECT t, 1 FROM generate_series(now() - interval '3 hours', now(), interval '1 minute') AS t;
+-- run again without and should have 362 rows
+CALL run_job(:job_id);
+SELECT count(*) FROM issue_6902;
+ count 
+-------
+   362
+(1 row)
+

tsl/test/sql/cagg_policy.sql

@@ -676,3 +676,62 @@ AS SELECT time_bucket(1, a), sum(b)
    FROM t GROUP BY time_bucket(1, a);
 
 SELECT timescaledb_experimental.add_policies('cagg');
+
+-- Issue #6902
+-- Fix timestamp out of range in a refresh policy when setting `end_offset=>NULL`
+-- for a CAgg with variable sized bucket (i.e: using `time_bucket` with timezone)
+CREATE TABLE issue_6902 (
+  ts  TIMESTAMPTZ NOT NULL,
+  temperature NUMERIC
+) WITH (
+  timescaledb.hypertable,
+  timescaledb.partition_column='ts',
+  timescaledb.chunk_interval='1 day',
+  timescaledb.compress='off'
+);
+
+INSERT INTO issue_6902
+SELECT t, 1 FROM generate_series(now() - interval '3 hours', now(), interval '1 minute') AS t;
+
+CREATE MATERIALIZED VIEW issue_6902_by_hour
+WITH (timescaledb.continuous) AS
+SELECT
+  time_bucket(INTERVAL '1 hour', ts, 'America/Sao_Paulo') AS bucket, -- using timezone
+  MAX(temperature),
+  MIN(temperature),
+  COUNT(*)
+FROM issue_6902
+GROUP BY 1
+WITH NO DATA;
+
+SELECT add_continuous_aggregate_policy (
+  'issue_6902_by_hour',
+  start_offset => INTERVAL '3 hours',
+  end_offset => NULL,
+  schedule_interval => INTERVAL '12 hour',
+  initial_start => now() + INTERVAL '12 hour'
+) AS job_id \gset
+
+-- 181 rows
+CALL run_job(:job_id);
+SELECT count(*) FROM issue_6902;
+
+-- run again without any change, remain 181 rows
+CALL run_job(:job_id);
+SELECT count(*) FROM issue_6902;
+
+-- change existing data
+UPDATE issue_6902
+SET temperature = temperature + 1;
+
+-- run again without any change, remain 181 rows
+CALL run_job(:job_id);
+SELECT count(*) FROM issue_6902;
+
+-- insert more data
+INSERT INTO issue_6902
+SELECT t, 1 FROM generate_series(now() - interval '3 hours', now(), interval '1 minute') AS t;
+
+-- run again without and should have 362 rows
+CALL run_job(:job_id);
+SELECT count(*) FROM issue_6902;