Refactor cagg bucket alignment code (#9729)

Refactor the bucket alignment code

Refactor the bucket alignment code to reduce duplication
    - Extract the logics that computes the bucket start and next bucket
    start into corresponding functions to use at a few places
    - Consolidate functions that compute bucket start for both fix and
    variable bucket.
    - Clean up the inscribe/circumscribe functions

Disable-check: force-changelog-file

Author: pnthao

src/ts_catalog/continuous_agg.c

@@ -1451,28 +1451,12 @@ void
 ts_compute_inscribed_bucketed_refresh_window_variable(int64 *start, int64 *end,
 													  const ContinuousAggBucketFunction *bf)
 {
-	Datum start_old, end_old, start_aligned, end_aliged;
-
-	/*
-	 * It's OK to use TIMESTAMPOID here. Variable-sized buckets can be used
-	 * only for dates, timestamps and timestamptz's. For all these types our
-	 * internal time representation is microseconds relative the UNIX epoch.
-	 * So the results will be correct regardless of the actual type used in
-	 * the CAGG. For more details see ts_internal_to_time_value() implementation.
-	 */
-	start_old = ts_internal_to_time_value(*start, TIMESTAMPOID);
-	end_old = ts_internal_to_time_value(*end, TIMESTAMPOID);
-
-	start_aligned = generic_time_bucket(bf, start_old);
-	end_aliged = generic_time_bucket(bf, end_old);
-
-	if (DatumGetTimestamp(start_aligned) != DatumGetTimestamp(start_old))
+	int64 start_aligned = ts_cagg_variable_current_bucket_start(*start, bf);
+	if (start_aligned != *start)
 	{
-		start_aligned = generic_add_interval(bf, start_aligned);
+		*start = ts_cagg_variable_next_bucket_start(*start, bf);
 	}
-
-	*start = ts_time_value_to_internal(start_aligned, TIMESTAMPOID);
-	*end = ts_time_value_to_internal(end_aliged, TIMESTAMPOID);
+	*end = ts_cagg_variable_current_bucket_start(*end, bf);
 }
 
 /*
@@ -1489,28 +1473,20 @@ void
 ts_compute_circumscribed_bucketed_refresh_window_variable(int64 *start, int64 *end,
 														  const ContinuousAggBucketFunction *bf)
 {
-	Datum start_old, end_old, start_new, end_new;
-
-	/*
-	 * It's OK to use TIMESTAMPOID here.
-	 * See the comment in ts_compute_inscribed_bucketed_refresh_window_variable()
-	 */
-	start_old = ts_internal_to_time_value(*start, TIMESTAMPOID);
-	end_old = ts_internal_to_time_value(*end, TIMESTAMPOID);
-	start_new = generic_time_bucket(bf, start_old);
-	end_new = generic_time_bucket(bf, end_old);
+	*start = ts_cagg_variable_current_bucket_start(*start, bf);
+	int64 end_new = ts_cagg_variable_current_bucket_start(*end, bf);
 
 	/* Add interval to expand to next bucket if:
 	 * 1. end wasn't at a bucket boundary (end moved during bucketing), OR
 	 * 2. we have a single-point at a bucket boundary (start == end after bucketing) */
-	if (DatumGetTimestamp(end_new) != DatumGetTimestamp(end_old) ||
-		DatumGetTimestamp(start_new) == DatumGetTimestamp(end_new))
+	if (end_new != *end || *start == end_new)
 	{
-		end_new = generic_add_interval(bf, end_new);
+		*end = ts_cagg_variable_next_bucket_start(*end, bf);
+	}
+	else
+	{
+		*end = end_new;
 	}
-
-	*start = ts_time_value_to_internal(start_new, TIMESTAMPOID);
-	*end = ts_time_value_to_internal(end_new, TIMESTAMPOID);
 }
 
 /*
@@ -1521,15 +1497,14 @@ ts_compute_circumscribed_bucketed_refresh_window_variable(int64 *start, int64 *e
  * val = time_bucket(bucket_size, val) + interval bucket_size
  */
 int64
-ts_compute_beginning_of_the_next_bucket_variable(int64 timeval,
-												 const ContinuousAggBucketFunction *bf)
+ts_cagg_variable_next_bucket_start(int64 timeval, const ContinuousAggBucketFunction *bf)
 {
 	Datum val_new;
 	Datum val_old;
 
 	/*
 	 * It's OK to use TIMESTAMPOID here.
-	 * See the comment in ts_compute_inscribed_bucketed_refresh_window_variable()
+	 * See the comment in ts_cagg_variable_current_bucket_start()
 	 */
 	val_old = ts_internal_to_time_value(timeval, TIMESTAMPOID);
 
@@ -1547,11 +1522,14 @@ ts_compute_beginning_of_the_next_bucket_variable(int64 timeval,
  * val = time_bucket(bucket_size, val)
  */
 int64
-ts_compute_start_of_current_bucket_variable(int64 timeval, const ContinuousAggBucketFunction *bf)
+ts_cagg_variable_current_bucket_start(int64 timeval, const ContinuousAggBucketFunction *bf)
 {
 	/*
-	 * It's OK to use TIMESTAMPOID here.
-	 * See the comment in ts_compute_inscribed_bucketed_refresh_window_variable()
+	 * It's OK to use TIMESTAMPOID here. Variable-sized buckets can be used
+	 * only for dates, timestamps and timestamptz's. For all these types our
+	 * internal time representation is microseconds relative the UNIX epoch.
+	 * So the results will be correct regardless of the actual type used in
+	 * the CAGG. For more details see ts_internal_to_time_value() implementation.
 	 */
 	Datum val_beg = ts_internal_to_time_value(timeval, TIMESTAMPOID);
 	return ts_time_value_to_internal(generic_time_bucket(bf, val_beg), TIMESTAMPOID);

src/ts_catalog/continuous_agg.h

@@ -186,10 +186,10 @@ ts_compute_inscribed_bucketed_refresh_window_variable(int64 *start, int64 *end,
 extern TSDLLEXPORT void
 ts_compute_circumscribed_bucketed_refresh_window_variable(int64 *start, int64 *end,
 														  const ContinuousAggBucketFunction *bf);
-extern TSDLLEXPORT int64 ts_compute_beginning_of_the_next_bucket_variable(
-	int64 timeval, const ContinuousAggBucketFunction *bf);
+extern TSDLLEXPORT int64 ts_cagg_variable_next_bucket_start(int64 timeval,
+															const ContinuousAggBucketFunction *bf);
 extern TSDLLEXPORT int64
-ts_compute_start_of_current_bucket_variable(int64 timeval, const ContinuousAggBucketFunction *bf);
+ts_cagg_variable_current_bucket_start(int64 timeval, const ContinuousAggBucketFunction *bf);
 
 extern TSDLLEXPORT Query *ts_continuous_agg_get_query(ContinuousAgg *cagg);
 

src/ts_catalog/continuous_aggs_watermark.c

@@ -142,11 +142,10 @@ cagg_compute_watermark(ContinuousAgg *cagg, int64 watermark, bool isnull)
 		{
 			/*
 			 * Since `value` is already bucketed, `bucketed = true` flag can
-			 * be added to ts_compute_beginning_of_the_next_bucket_variable() as
+			 * be added to ts_cagg_variable_next_bucket_start() as
 			 * an optimization, if necessary.
 			 */
-			watermark =
-				ts_compute_beginning_of_the_next_bucket_variable(watermark, cagg->bucket_function);
+			watermark = ts_cagg_variable_next_bucket_start(watermark, cagg->bucket_function);
 		}
 		else
 		{

tsl/src/continuous_aggs/invalidation.c

@@ -512,10 +512,6 @@ invalidation_expand_to_bucket_boundaries(Invalidation *inv, Oid time_type_oid,
 	int64 bucket_width = ts_continuous_agg_fixed_bucket_width(bucket_function);
 	Assert(bucket_width > 0);
 
-	NullableDatum offset = INIT_NULL_DATUM;
-	NullableDatum origin = INIT_NULL_DATUM;
-	fill_bucket_offset_origin(bucket_function, time_type_oid, &offset, &origin);
-
 	/* Compute the start of the "first" bucket for the type. The min value
 	 * must be at the start of the "first" bucket or somewhere in the
 	 * bucket. If the min value falls on the exact start of the bucket we are
@@ -555,11 +551,9 @@ invalidation_expand_to_bucket_boundaries(Invalidation *inv, Oid time_type_oid,
 	}
 	else
 	{
-		inv->lowest_modified_value = ts_time_bucket_by_type_extended(bucket_width,
-																	 inv->lowest_modified_value,
+		inv->lowest_modified_value = cagg_fixed_current_bucket_start(inv->lowest_modified_value,
 																	 time_type_oid,
-																	 offset,
-																	 origin);
+																	 bucket_function);
 	}
 
 	if (inv->greatest_modified_value < min_bucket_start)
@@ -574,11 +568,9 @@ invalidation_expand_to_bucket_boundaries(Invalidation *inv, Oid time_type_oid,
 	}
 	else
 	{
-		inv->greatest_modified_value = ts_time_bucket_by_type_extended(bucket_width,
-																	   inv->greatest_modified_value,
+		inv->greatest_modified_value = cagg_fixed_current_bucket_start(inv->greatest_modified_value,
 																	   time_type_oid,
-																	   offset,
-																	   origin);
+																	   bucket_function);
 		inv->greatest_modified_value =
 			ts_time_saturating_add(inv->greatest_modified_value, bucket_width - 1, time_type_oid);
 	}

tsl/src/continuous_aggs/invalidation_threshold.c

@@ -264,27 +264,7 @@ invalidation_threshold_compute(const ContinuousAgg *cagg, const InternalTimeRang
 		}
 		else
 		{
-			if (cagg->bucket_function->bucket_fixed_interval == false)
-			{
-				return ts_compute_beginning_of_the_next_bucket_variable(maxval,
-																		cagg->bucket_function);
-			}
-
-			int64 bucket_width = ts_continuous_agg_fixed_bucket_width(cagg->bucket_function);
-			Assert(bucket_width > 0);
-			NullableDatum offset = INIT_NULL_DATUM;
-			NullableDatum origin = INIT_NULL_DATUM;
-			fill_bucket_offset_origin(cagg->bucket_function,
-									  refresh_window->type,
-									  &offset,
-									  &origin);
-			int64 bucket_start = ts_time_bucket_by_type_extended(bucket_width,
-																 maxval,
-																 refresh_window->type,
-																 offset,
-																 origin);
-			/* Add one bucket to get to the end of the last bucket */
-			return ts_time_saturating_add(bucket_start, bucket_width, refresh_window->type);
+			return cagg_next_bucket_start(maxval, refresh_window->type, cagg->bucket_function);
 		}
 	}