Unify handling of CAgg bucket_origin

sql/updates/latest-dev.sql

@@ -141,3 +141,8 @@ DROP FUNCTION IF EXISTS _timescaledb_functions.cagg_get_bucket_function(INTEGER)
 --
 -- End rebuild the catalog table `_timescaledb_catalog.continuous_aggs_bucket_function`
 --
+
+-- Convert _timescaledb_catalog.continuous_aggs_bucket_function.bucket_origin to TimestampTZ
+UPDATE _timescaledb_catalog.continuous_aggs_bucket_function
+   SET bucket_origin = bucket_origin::timestamp::timestamptz::text
+   WHERE length(bucket_origin) > 1;

sql/updates/reverse-dev.sql

@@ -52,3 +52,8 @@ ANALYZE _timescaledb_catalog.continuous_aggs_bucket_function;
 --
 -- End rebuild the catalog table `_timescaledb_catalog.continuous_aggs_bucket_function`
 --
+
+-- Convert _timescaledb_catalog.continuous_aggs_bucket_function.origin back to Timestamp
+UPDATE _timescaledb_catalog.continuous_aggs_bucket_function
+   SET origin = origin::timestamptz::timestamp::text
+   WHERE length(origin) > 1;

src/ts_catalog/continuous_agg.c

@@ -452,7 +452,7 @@ continuous_agg_fill_bucket_function(int32 mat_hypertable_id, ContinuousAggsBucke
 		if (strlen(origin_str) == 0)
 			TIMESTAMP_NOBEGIN(bf->bucket_origin);
 		else
-			bf->bucket_origin = DatumGetTimestamp(DirectFunctionCall3(timestamp_in,
+			bf->bucket_origin = DatumGetTimestamp(DirectFunctionCall3(timestamptz_in,
 																	  CStringGetDatum(origin_str),
 																	  ObjectIdGetDatum(InvalidOid),
 																	  Int32GetDatum(-1)));
@@ -1377,7 +1377,7 @@ generic_time_bucket(const ContinuousAggsBucketFunction *bf, Datum timestamp)
 										   IntervalPGetDatum(bf->bucket_width),
 										   timestamp,
 										   CStringGetTextDatum(bf->timezone),
-										   TimestampTzGetDatum((TimestampTz) bf->bucket_origin));
+										   TimestampTzGetDatum(bf->bucket_origin));
 			}
 		}
 
@@ -1394,7 +1394,7 @@ generic_time_bucket(const ContinuousAggsBucketFunction *bf, Datum timestamp)
 			return DirectFunctionCall3(ts_timestamp_bucket,
 									   IntervalPGetDatum(bf->bucket_width),
 									   timestamp,
-									   TimestampGetDatum(bf->bucket_origin));
+									   TimestampTzGetDatum(bf->bucket_origin));
 		}
 	}
 	else
@@ -1415,7 +1415,7 @@ generic_time_bucket(const ContinuousAggsBucketFunction *bf, Datum timestamp)
 				return DirectFunctionCall4(ts_time_bucket_ng_timezone_origin,
 										   IntervalPGetDatum(bf->bucket_width),
 										   timestamp,
-										   TimestampTzGetDatum((TimestampTz) bf->bucket_origin),
+										   TimestampTzGetDatum(bf->bucket_origin),
 										   CStringGetTextDatum(bf->timezone));
 			}
 		}
@@ -1433,7 +1433,7 @@ generic_time_bucket(const ContinuousAggsBucketFunction *bf, Datum timestamp)
 			return DirectFunctionCall3(ts_time_bucket_ng_timestamp,
 									   IntervalPGetDatum(bf->bucket_width),
 									   timestamp,
-									   TimestampGetDatum(bf->bucket_origin));
+									   TimestampTzGetDatum(bf->bucket_origin));
 		}
 	}
 }

src/ts_catalog/continuous_agg.h

@@ -90,7 +90,7 @@ typedef struct ContinuousAggsBucketFunction
 	 * Custom origin value stored as UTC timestamp.
 	 * If not specified, stores infinity.
 	 */
-	Timestamp bucket_origin;
+	TimestampTz bucket_origin;
 
 	/* `bucket_offset` argument of the function. */
 	Interval *bucket_offest;

tsl/src/continuous_aggs/common.h

@@ -80,7 +80,7 @@ typedef struct CAggTimebucketInfo
 	 * Custom origin value stored as UTC timestamp.
 	 * If not specified, stores infinity.
 	 */
-	Timestamp origin;
+	TimestampTz origin;
 } CAggTimebucketInfo;
 
 #define CAGG_MAKEQUERY(selquery, srcquery)                                                         \

tsl/src/continuous_aggs/create.c

@@ -798,7 +798,7 @@ cagg_create(const CreateTableAsStmt *create_stmt, ViewStmt *stmt, Query *panquer
 		if (!TIMESTAMP_NOT_FINITE(bucket_info->origin))
 		{
 			origin = DatumGetCString(
-				DirectFunctionCall1(timestamp_out, TimestampGetDatum(bucket_info->origin)));
+				DirectFunctionCall1(timestamptz_out, TimestampTzGetDatum(bucket_info->origin)));
 		}
 
 		create_bucket_function_catalog_entry(materialize_hypertable_id,

tsl/test/expected/exp_cagg_origin.out

@@ -773,6 +773,19 @@ SELECT city,
 FROM conditions_timestamptz
 GROUP BY city, bucket;
 NOTICE:  refreshing continuous aggregate "conditions_summary_timestamptz"
+-- Make sure the origin is saved in the catalog table
+SELECT mat_hypertable_id AS cagg_id
+  FROM _timescaledb_catalog.continuous_agg
+  WHERE user_view_name = 'conditions_summary_timestamptz'
+\gset
+SELECT bucket_func, bucket_width, bucket_origin, bucket_timezone, bucket_fixed_width
+  FROM _timescaledb_catalog.continuous_aggs_bucket_function
+  WHERE mat_hypertable_id = :cagg_id;
+                                               bucket_func                                                | bucket_width |        bucket_origin         | bucket_timezone | bucket_fixed_width 
+----------------------------------------------------------------------------------------------------------+--------------+------------------------------+-----------------+--------------------
+ timescaledb_experimental.time_bucket_ng(interval,timestamp with time zone,timestamp with time zone,text) | @ 12 hours   | Mon Jun 01 02:00:00 2020 PDT | Europe/Moscow   | f
+(1 row)
+
 SELECT city, to_char(bucket at time zone 'MSK', 'YYYY-MM-DD HH24:MI:SS') AS b, min, max
 FROM conditions_summary_timestamptz
 ORDER BY b, city;

tsl/test/sql/exp_cagg_origin.sql

@@ -500,6 +500,16 @@ SELECT city,
 FROM conditions_timestamptz
 GROUP BY city, bucket;
 
+-- Make sure the origin is saved in the catalog table
+SELECT mat_hypertable_id AS cagg_id
+  FROM _timescaledb_catalog.continuous_agg
+  WHERE user_view_name = 'conditions_summary_timestamptz'
+\gset
+
+SELECT bucket_func, bucket_width, bucket_origin, bucket_timezone, bucket_fixed_width
+  FROM _timescaledb_catalog.continuous_aggs_bucket_function
+  WHERE mat_hypertable_id = :cagg_id;
+
 SELECT city, to_char(bucket at time zone 'MSK', 'YYYY-MM-DD HH24:MI:SS') AS b, min, max
 FROM conditions_summary_timestamptz
 ORDER BY b, city;