Skip to content

Refactor cagg bucket alignment code #9729

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

Merged
pnthao merged 1 commit into from
May 8, 2026
Merged

Refactor cagg bucket alignment code #9729

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
62 changes: 20 additions & 42 deletions src/ts_catalog/continuous_agg.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}

/*
Expand All @@ -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);
}

/*
Expand All @@ -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);

Expand All @@ -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);
Expand Down
6 changes: 3 additions & 3 deletions src/ts_catalog/continuous_agg.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);

Expand Down
5 changes: 2 additions & 3 deletions src/ts_catalog/continuous_aggs_watermark.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
{
Expand Down
16 changes: 4 additions & 12 deletions tsl/src/continuous_aggs/invalidation.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand All @@ -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);
}
Expand Down
22 changes: 1 addition & 21 deletions tsl/src/continuous_aggs/invalidation_threshold.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}

Expand Down
Loading
Loading