Optimize cagg refresh for small invalidations

[erimatnor]

This PR includes two changes:

Change 1:

The refreshing of a continuous aggregate is slow when many small
invalidations are generated by frequent single row insert
backfills. This change adds an optimization that merges small
invalidations by first expanding invalidations to full bucket
boundaries. There is really no reason to maintain invalidations that
aren't covering full buckets since refresh windows are already aligned
to buckets anyway.

Change 2:

When there are many small (e.g., single timestamp) invalidations that
cannot be merged despite expanding invalidations to full buckets
(e.g., invalidations are spread across every second bucket in the
worst case), it might no longer be beneficial to materialize every
invalidation separately.

Instead, this change adds a threshold for the number of invalidations
used by the refresh (currently 10 by default) above which
invalidations are merged into one range based on the lowest and
greatest invalidated time value.

The limit can be controlled by an anonymous session variable for
debugging and tweaking purposes. It might be considered for promotion
to an official GUC in the future.

Fixes #2867

[codecov]

Codecov Report

Merging #2926 (0fcdf01) into master (d43e54a) will increase coverage by 0.12%.
The diff coverage is 94.88%.

@@            Coverage Diff             @@
##           master    #2926      +/-   ##
==========================================
+ Coverage   90.15%   90.28%   +0.12%     
==========================================
  Files         212      212              
  Lines       34799    34894      +95     
==========================================
+ Hits        31373    31503     +130     
+ Misses       3426     3391      -35     

Impacted Files	Coverage Δ
src/compat.h	100.00% <ø> (ø)
tsl/src/remote/connection_cache.c	90.40% <ø> (ø)
tsl/src/bgw_policy/continuous_aggregate_api.c	93.64% <90.75%> (-4.01%)	⬇️
tsl/src/continuous_aggs/invalidation.c	97.94% <97.14%> (-0.17%)	⬇️
tsl/src/continuous_aggs/refresh.c	97.58% <98.14%> (-0.16%)	⬇️
src/continuous_agg.c	90.99% <100.00%> (+0.34%)	⬆️
src/process_utility.c	93.85% <100.00%> (-0.02%)	⬇️
tsl/src/bgw_policy/job.c	97.34% <100.00%> (-0.02%)	⬇️
tsl/src/continuous_aggs/create.c	96.92% <100.00%> (-0.01%)	⬇️
tsl/src/continuous_aggs/insert.c	85.71% <100.00%> (ø)
... and 4 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 7a711fc...0fcdf01. Read the comment docs.

[gayyappan]

Possible optimization: cache min_bucket_start and max_bucket_end so that we don't have to recompute this while processing invalidations in a loop.

[erimatnor]

I thought about it, but figured it wasn't worth it. Perhaps you have a suggestion of how to easily cache it without making this unnecessarily complex?

[gayyappan]

If the buckets are big, this could be a lot of data and we might run into some of the issues that were previously solved by setting max_interval_per_job. Probably makes sense to add a guc as part of the PR.

[erimatnor]

The semantics here are different from max_interval_per_job. The old max_interval_per_job setting is actually already present in a similar form as the user- or policy-specified refresh window; i.e., the user controls the max amount materialized by specifying a specific refresh window size. So, if there's a lot of data to materialize, the way to deal with that is to specify a smaller refresh window, instead doing several refreshes. This becomes clear if you think about the worst case situation, i.e., when every bucket in the refresh window is invalid. In that case, this max invalidatoins per refresh value doesn't matter because you need to refresh the whole window anyway.

This max value here is about how we optimize the materialization of the specified refresh window when we are not in the worst case and it is possible to break the window down in smaller regions. Pre-2.0, everything from the earliest invalidation and forward was re-materialized no matter whether the entire range needed re-materialization or not. Post-2.0 we break that down so that we only materialize the regions/buckets that are actually invalid. However, if there are a lot of small regions that are invalidated (e.g., every other bucket in the worst case), it is cheaper to merge that into a large range. The session variable below controls the threshold.

[mkindahl]

Found a few things that are not clear, could be, or is a problem. If you do not think they are, please respond to them.

I also think that is not a good idea to have the max_invalidations_per_refresh for the user function and especially having it to default to 10 (which it does, from my understanding). For the policy runs, this is just a way to throttle the background workers, which is fine, but when calling this from the command line the result will be that the entire range is not refreshed and the user will have to type the command several times, which will be annoying. I think it might be better to add a parameter to the function which defaults to do a full refresh and ensure that the policy set a value for this parameter instead. This will make sure that policies can be throttled and have a good default value, while command-line calls will refresh the entire range.

[pmwkaa]

Looks good overall, don't have much to add, just a couple of comments

[erimatnor]

I also think that is not a good idea to have the max_invalidations_per_refresh for the user function and especially having it to default to 10 (which it does, from my understanding). For the policy runs, this is just a way to throttle the background workers, which is fine, but when calling this from the command line the result will be that the entire range is not refreshed and the user will have to type the command several times, which will be annoying.

So, this seems to be a misunderstanding of what the changes do, and that's on me for failing to make clear and perhaps a badly named variable.

Two things that I'd like to clarify:

• The changes in this PR in no way changes the semantics of refresh_continuous_aggregate. After a call to this function, everything in the given window will be up-to-date. This hasn't changed. What happens internally, though, is that we break a refresh down into a set of multiple sub-refreshes (materialization) on the ranges (buckets) that are actually out-of-date. But if that turns out to be 100 separate refreshes, this is likely to be slower than one single refresh across the range determined by the min invalidated time and max invalidated time. The max_invalidations_per_refresh just sets the threshold of where we merge those separate ranges into one big range. It does not mean that we leave some ranges out-of-date.
• None of the changes in this PR has anything to do with throttling policies. Not sure where that impression came from.

I've tried to change the comments and variable names to make this more clear.

[mkindahl]

I also think that is not a good idea to have the max_invalidations_per_refresh for the user function and especially having it to default to 10 (which it does, from my understanding). For the policy runs, this is just a way to throttle the background workers, which is fine, but when calling this from the command line the result will be that the entire range is not refreshed and the user will have to type the command several times, which will be annoying.

So, this seems to be a misunderstanding of what the changes do, and that's on me for failing to make clear and perhaps a badly named variable.

Two things that I'd like to clarify:

* The changes in this PR in no way changes the semantics of `refresh_continuous_aggregate`. After a call to this function, everything in the given window will be up-to-date. This hasn't changed. What happens internally, though, is that we break a refresh down into a set of multiple sub-refreshes (materialization) on the ranges (buckets) that are actually out-of-date. But if that turns out to be 100 separate refreshes, this is likely to be slower than one single refresh across the range determined by the min invalidated time and max invalidated time. The `max_invalidations_per_refresh` just sets the threshold of where we merge those separate ranges into one big range. It does _not_ mean that we leave some ranges out-of-date.

* None of the changes in this PR has anything to do with throttling policies. Not sure where that impression came from.

I've tried to change the comments and variable names to make this more clear.

I understand what the variable do, but IMHO the problem is that the default behavior is to not process the entire range when it is "too complicated" (when we have a lot of buckets that are not connected). We had several comments before when REFRESH MATERIALIZED VIEW didn't refresh everything, so I think we should make sure that we, by default, refresh the entire range even if that happens to take a long time.

[erimatnor]

I understand what the variable do, but IMHO the problem is that the default behavior is to not process the entire range when it is "too complicated" (when we have a lot of buckets that are not connected). We had several comments before when REFRESH MATERIALIZED VIEW didn't refresh everything, so I think we should make sure that we, by default, refresh the entire range even if that happens to take a long time.

Above the threshold we actually process more than necessary, so it is actually the opposite of your understanding. When you say we had comments about REFRESH MATERIALIZED VIEW, I assume you are referring to pre-2.0 behavior. That functionality was entirely different from what we do here, so there is no clear comparison. The semantics of refresh_continuous_aggregate here is to leave everything in the user-specified window/range up-to-date after the call completes. So, your statement about not processing the entire range is not the way it works.

[mkindahl]

Thanks for the clarification that the changes does not affect the range that the refresh function process. LGTM.

[erimatnor]

@mkindahl @pmwkaa Changed the following based on feedback.

• Updated comments for min/max bucket calculation (also made a small tweak to calculating the end of the last bucket)
• Added warning for the session variable parsing, and also added support for trailing whitespaces
• Added tests for non-parsable input of session variable

[mkindahl]

I understand what the variable do, but IMHO the problem is that the default behavior is to not process the entire range when it is "too complicated" (when we have a lot of buckets that are not connected). We had several comments before when REFRESH MATERIALIZED VIEW didn't refresh everything, so I think we should make sure that we, by default, refresh the entire range even if that happens to take a long time.

Above the threshold we actually process more than necessary, so it is actually the opposite of your understanding. When you say we had comments about REFRESH MATERIALIZED VIEW, I assume you are referring to pre-2.0 behavior. That functionality was entirely different from what we do here, so there is no clear comparison. The semantics of refresh_continuous_aggregate here is to leave everything in the user-specified window/range up-to-date after the call completes. So, your statement about not processing the entire range is not the way it works.

Good to hear. Then I misunderstood how it worked.