diff --git a/src/ts_catalog/continuous_agg.c b/src/ts_catalog/continuous_agg.c index 341d06c6d5d..4a50d8ebf5c 100644 --- a/src/ts_catalog/continuous_agg.c +++ b/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); diff --git a/src/ts_catalog/continuous_agg.h b/src/ts_catalog/continuous_agg.h index e8241d2ce4a..ef68f31792a 100644 --- a/src/ts_catalog/continuous_agg.h +++ b/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); diff --git a/src/ts_catalog/continuous_aggs_watermark.c b/src/ts_catalog/continuous_aggs_watermark.c index 21814764f32..502ce2a2703 100644 --- a/src/ts_catalog/continuous_aggs_watermark.c +++ b/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 { diff --git a/tsl/src/continuous_aggs/invalidation.c b/tsl/src/continuous_aggs/invalidation.c index 9e1d8ae53d0..7c918f85e71 100644 --- a/tsl/src/continuous_aggs/invalidation.c +++ b/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); } diff --git a/tsl/src/continuous_aggs/invalidation_threshold.c b/tsl/src/continuous_aggs/invalidation_threshold.c index a29dcd6ea93..f77dfb2d855 100644 --- a/tsl/src/continuous_aggs/invalidation_threshold.c +++ b/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); } } diff --git a/tsl/src/continuous_aggs/refresh.c b/tsl/src/continuous_aggs/refresh.c index 797057df214..7bb55bcca56 100644 --- a/tsl/src/continuous_aggs/refresh.c +++ b/tsl/src/continuous_aggs/refresh.c @@ -58,9 +58,8 @@ typedef struct CaggRefreshSpiContext static Hypertable *cagg_get_hypertable_or_fail(int32 hypertable_id); static InternalTimeRange get_largest_bucketed_window(Oid timetype, int64 bucket_width); static InternalTimeRange -compute_inscribed_bucketed_refresh_window(const ContinuousAgg *cagg, - const InternalTimeRange *const refresh_window, - const int64 bucket_width); +compute_inscribed_bucketed_refresh_window(const InternalTimeRange *const refresh_window, + const ContinuousAggBucketFunction *bucket_function); static void continuous_agg_refresh_init(ContinuousAggRefreshState *refresh, const ContinuousAgg *cagg, const InternalTimeRange *refresh_window, @@ -154,25 +153,27 @@ get_largest_bucketed_window(Oid timetype, int64 bucket_width) * where part of its data were dropped by a retention policy. See #2198 for details. */ static InternalTimeRange -compute_inscribed_bucketed_refresh_window(const ContinuousAgg *cagg, - const InternalTimeRange *const refresh_window, - const int64 bucket_width) +compute_inscribed_bucketed_refresh_window(const InternalTimeRange *const refresh_window, + const ContinuousAggBucketFunction *bucket_function) { - Assert(cagg != NULL); - Assert(cagg->bucket_function != NULL); + Assert(bucket_function != NULL); + + if (bucket_function->bucket_fixed_interval == false) + { + InternalTimeRange result = *refresh_window; + ts_compute_inscribed_bucketed_refresh_window_variable(&result.start, + &result.end, + bucket_function); + return result; + } + + int64 bucket_width = ts_continuous_agg_fixed_bucket_width(bucket_function); + Assert(bucket_width > 0); InternalTimeRange result = *refresh_window; InternalTimeRange largest_bucketed_window = get_largest_bucketed_window(refresh_window->type, bucket_width); - /* Get offset and origin for bucket function */ - NullableDatum offset = INIT_NULL_DATUM; - NullableDatum origin = INIT_NULL_DATUM; - fill_bucket_offset_origin(cagg->bucket_function, refresh_window->type, &offset, &origin); - - /* Defined offset and origin in one function is not supported */ - Assert(offset.isnull == true || origin.isnull == true); - if (refresh_window->start <= largest_bucketed_window.start) { result.start = largest_bucketed_window.start; @@ -186,11 +187,8 @@ compute_inscribed_bucketed_refresh_window(const ContinuousAgg *cagg, int64 included_bucket = ts_time_saturating_add(refresh_window->start, bucket_width - 1, refresh_window->type); /* Get the start of the included bucket. */ - result.start = ts_time_bucket_by_type_extended(bucket_width, - included_bucket, - refresh_window->type, - offset, - origin); + result.start = + cagg_fixed_current_bucket_start(included_bucket, refresh_window->type, bucket_function); } if (refresh_window->end >= largest_bucketed_window.end) @@ -201,11 +199,9 @@ compute_inscribed_bucketed_refresh_window(const ContinuousAgg *cagg, { /* The window is reduced to the beginning of the bucket, which contains the exclusive * end of the refresh window. */ - result.end = ts_time_bucket_by_type_extended(bucket_width, - refresh_window->end, + result.end = cagg_fixed_current_bucket_start(refresh_window->end, refresh_window->type, - offset, - origin); + bucket_function); } return result; } @@ -235,7 +231,7 @@ int_bucket_offset_to_datum(Oid type, const ContinuousAggBucketFunction *bucket_f /* * Get a NullableDatum for offset and origin based on the CAgg information */ -void +static void fill_bucket_offset_origin(const ContinuousAggBucketFunction *bucket_function, Oid type, NullableDatum *offset, NullableDatum *origin) { @@ -280,6 +276,76 @@ fill_bucket_offset_origin(const ContinuousAggBucketFunction *bucket_function, Oi } } +/* + * Compute the start of the bucket containing the given timestamp, accounting + * for the CAgg's offset and origin. + * + * This is a convenience wrapper that combines fill_bucket_offset_origin and + * ts_time_bucket_by_type_extended for fixed-interval buckets. + */ +int64 +cagg_fixed_current_bucket_start(int64 timestamp, Oid type, + const ContinuousAggBucketFunction *bucket_function) +{ + 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, type, &offset, &origin); + Assert(offset.isnull == true || origin.isnull == true); + return ts_time_bucket_by_type_extended(bucket_width, timestamp, type, offset, origin); +} + +/* + * Compute the start of the bucket immediately following the bucket that + * contains the given timestamp. Equivalently, this is the exclusive end of + * the bucket containing timestamp. + * + * This is a convenience wrapper used when the caller needs to advance past the + * current bucket (e.g. computing an invalidation threshold or the exclusive + * upper bound of a circumscribed refresh window). + */ +int64 +cagg_fixed_next_bucket_start(int64 timestamp, Oid type, + const ContinuousAggBucketFunction *bucket_function) +{ + int64 bucket_width = ts_continuous_agg_fixed_bucket_width(bucket_function); + Assert(bucket_width > 0); + int64 bucket_start = cagg_fixed_current_bucket_start(timestamp, type, bucket_function); + return ts_time_saturating_add(bucket_start, bucket_width, type); +} + +/* + * Compute the start of the bucket containing the given timestamp, dispatching + * to the fixed-interval or variable-interval implementation as appropriate. + */ +int64 +cagg_current_bucket_start(int64 timestamp, Oid type, + const ContinuousAggBucketFunction *bucket_function) +{ + if (bucket_function->bucket_fixed_interval) + { + return cagg_fixed_current_bucket_start(timestamp, type, bucket_function); + } + return ts_cagg_variable_current_bucket_start(timestamp, bucket_function); +} + +/* + * Compute the start of the bucket immediately following the bucket containing + * the given timestamp, dispatching to the fixed-interval or variable-interval + * implementation as appropriate. + */ +int64 +cagg_next_bucket_start(int64 timestamp, Oid type, + const ContinuousAggBucketFunction *bucket_function) +{ + if (bucket_function->bucket_fixed_interval) + { + return cagg_fixed_next_bucket_start(timestamp, type, bucket_function); + } + return ts_cagg_variable_next_bucket_start(timestamp, bucket_function); +} + /* * Adjust the refresh window to align with circumscribed buckets, so it includes buckets, which * fully cover the refresh window. @@ -308,12 +374,10 @@ fill_bucket_offset_origin(const ContinuousAggBucketFunction *bucket_function, Oi * dropping chunks manually or as part of retention policy. */ InternalTimeRange -compute_circumscribed_bucketed_refresh_window(const ContinuousAgg *cagg, - const InternalTimeRange *const refresh_window, +compute_circumscribed_bucketed_refresh_window(const InternalTimeRange *const refresh_window, const ContinuousAggBucketFunction *bucket_function) { - Assert(cagg != NULL); - Assert(cagg->bucket_function != NULL); + Assert(bucket_function != NULL); if (bucket_function->bucket_fixed_interval == false) { @@ -332,14 +396,6 @@ compute_circumscribed_bucketed_refresh_window(const ContinuousAgg *cagg, InternalTimeRange largest_bucketed_window = get_largest_bucketed_window(refresh_window->type, bucket_width); - /* Get offset and origin for bucket function */ - NullableDatum offset = INIT_NULL_DATUM; - NullableDatum origin = INIT_NULL_DATUM; - fill_bucket_offset_origin(cagg->bucket_function, refresh_window->type, &offset, &origin); - - /* Defined offset and origin in one function is not supported */ - Assert(offset.isnull == true || origin.isnull == true); - if (refresh_window->start <= largest_bucketed_window.start) { result.start = largest_bucketed_window.start; @@ -348,11 +404,9 @@ compute_circumscribed_bucketed_refresh_window(const ContinuousAgg *cagg, { /* For alignment with a bucket, which includes the start of the refresh window, we just * need to get start of the bucket. */ - result.start = ts_time_bucket_by_type_extended(bucket_width, - refresh_window->start, + result.start = cagg_fixed_current_bucket_start(refresh_window->start, refresh_window->type, - offset, - origin); + bucket_function); } if (refresh_window->end >= largest_bucketed_window.end) @@ -361,24 +415,16 @@ compute_circumscribed_bucketed_refresh_window(const ContinuousAgg *cagg, } else { - int64 exclusive_end; - int64 bucketed_end; - Assert(refresh_window->end > result.start); + int64 exclusive_end; + /* The end of the window is non-inclusive so subtract one before * bucketing in case we're already at the end of the bucket (we don't * want to add an extra bucket). */ exclusive_end = ts_time_saturating_sub(refresh_window->end, 1, refresh_window->type); - bucketed_end = ts_time_bucket_by_type_extended(bucket_width, - exclusive_end, - refresh_window->type, - offset, - origin); - - /* We get the time value for the start of the bucket, so need to add - * bucket_width to get the end of it. */ - result.end = ts_time_saturating_add(bucketed_end, bucket_width, refresh_window->type); + result.end = + cagg_fixed_next_bucket_start(exclusive_end, refresh_window->type, bucket_function); } return result; } @@ -516,9 +562,7 @@ continuous_agg_scan_refresh_window_ranges(const ContinuousAgg *cagg, if (refresh->bucketing_refresh_window) { bucketed_refresh_window = - compute_circumscribed_bucketed_refresh_window(cagg, - &invalidation, - cagg->bucket_function); + compute_circumscribed_bucketed_refresh_window(&invalidation, cagg->bucket_function); } (*exec_func)(&bucketed_refresh_window, context, count, func_arg1); @@ -820,20 +864,8 @@ continuous_agg_refresh_internal(const ContinuousAgg *cagg_arg, /* No bucketing when open ended */ if (bucketing_refresh_window && !(start_isnull && end_isnull)) { - if (cagg->bucket_function->bucket_fixed_interval == false) - { - refresh_window = *refresh_window_arg; - ts_compute_inscribed_bucketed_refresh_window_variable(&refresh_window.start, - &refresh_window.end, - cagg->bucket_function); - } - else - { - int64 bucket_width = ts_continuous_agg_fixed_bucket_width(cagg->bucket_function); - Assert(bucket_width > 0); - refresh_window = - compute_inscribed_bucketed_refresh_window(cagg, refresh_window_arg, bucket_width); - } + refresh_window = + compute_inscribed_bucketed_refresh_window(refresh_window_arg, cagg->bucket_function); } /* If there is no other policy defined after this, the inscribed bucket calculated above @@ -845,18 +877,8 @@ continuous_agg_refresh_internal(const ContinuousAgg *cagg_arg, */ if (extend_last_bucket && !(start_isnull && end_isnull)) { - if (cagg->bucket_function->bucket_fixed_interval == false) - { - refresh_window.end = - ts_compute_beginning_of_the_next_bucket_variable(refresh_window.end, - cagg->bucket_function); - } - else - { - int64 bucket_width = ts_continuous_agg_fixed_bucket_width(cagg->bucket_function); - refresh_window.end = - ts_time_saturating_add(refresh_window.end, bucket_width, refresh_window.type); - } + refresh_window.end = + cagg_next_bucket_start(refresh_window.end, refresh_window.type, cagg->bucket_function); } if (refresh_window.start >= refresh_window.end) @@ -965,28 +987,10 @@ continuous_agg_refresh_internal(const ContinuousAgg *cagg_arg, if (invalidation_threshold > ts_time_get_min(refresh_window.type) && invalidation_threshold < ts_time_get_max(refresh_window.type)) { - if (cagg->bucket_function->bucket_fixed_interval) - { - NullableDatum offset = INIT_NULL_DATUM; - NullableDatum origin = INIT_NULL_DATUM; - fill_bucket_offset_origin(cagg->bucket_function, + computed_invalidation_threshold_for_cagg = + cagg_current_bucket_start(invalidation_threshold, refresh_window.type, - &offset, - &origin); - int64 bucket_width = ts_continuous_agg_fixed_bucket_width(cagg->bucket_function); - computed_invalidation_threshold_for_cagg = - ts_time_bucket_by_type_extended(bucket_width, - invalidation_threshold, - refresh_window.type, - offset, - origin); - } - else - { - computed_invalidation_threshold_for_cagg = - ts_compute_start_of_current_bucket_variable(invalidation_threshold, - cagg->bucket_function); - } + cagg->bucket_function); } if (refresh_window.end > computed_invalidation_threshold_for_cagg) @@ -1152,17 +1156,8 @@ continuous_agg_split_refresh_window(ContinuousAgg *cagg, InternalTimeRange *orig /* Compute the inscribed bucket for the capped refresh window range */ const int64 bucket_width = ts_continuous_agg_bucket_width(cagg->bucket_function); - if (cagg->bucket_function->bucket_fixed_interval == false) - { - ts_compute_inscribed_bucketed_refresh_window_variable(&refresh_window.start, - &refresh_window.end, - cagg->bucket_function); - } - else - { - refresh_window = - compute_inscribed_bucketed_refresh_window(cagg, &refresh_window, bucket_width); - } + refresh_window = + compute_inscribed_bucketed_refresh_window(&refresh_window, cagg->bucket_function); /* Check if the refresh size is large enough to produce bathes, if not then return no batches */ const int64 refresh_window_size = i64abs(refresh_window.end - refresh_window.start); diff --git a/tsl/src/continuous_aggs/refresh.h b/tsl/src/continuous_aggs/refresh.h index c17f8c47fa1..48b07f96aad 100644 --- a/tsl/src/continuous_aggs/refresh.h +++ b/tsl/src/continuous_aggs/refresh.h @@ -24,9 +24,14 @@ extern List *continuous_agg_split_refresh_window(ContinuousAgg *cagg, int32 buckets_per_batch, bool refresh_newest_first); InternalTimeRange -compute_circumscribed_bucketed_refresh_window(const ContinuousAgg *cagg, - const InternalTimeRange *const refresh_window, +compute_circumscribed_bucketed_refresh_window(const InternalTimeRange *const refresh_window, const ContinuousAggBucketFunction *bucket_function); -extern void fill_bucket_offset_origin(const ContinuousAggBucketFunction *bucket_function, Oid type, - NullableDatum *offset, NullableDatum *origin); +extern int64 cagg_fixed_current_bucket_start(int64 timestamp, Oid type, + const ContinuousAggBucketFunction *bucket_function); +extern int64 cagg_fixed_next_bucket_start(int64 timestamp, Oid type, + const ContinuousAggBucketFunction *bucket_function); +extern int64 cagg_current_bucket_start(int64 timestamp, Oid type, + const ContinuousAggBucketFunction *bucket_function); +extern int64 cagg_next_bucket_start(int64 timestamp, Oid type, + const ContinuousAggBucketFunction *bucket_function);