Incremental CAgg refresh using a work queue

[feikesteenbergen]

Continous Aggregates, incremental parallel setup

This code is exploring the possibilities to do incremental CAgg refreshes in
parallel. The setup it uses is as following.

At a very high level these are the components:

• a table that acts as a work queue:
  _timescaledb_additional.incremental_continuous_aggregate_refreshes
• one (or more) producer jobs that schedule CAgg refreshes
• one (or more) consumer jobs that process the jobs based on priority

The producer jobs can be scheduled very frequently, as no duplicate tasks will
be written to the work queue.

Producer

We have a producer procedure
(schedule_refresh_continuous_aggregate_incremental), which schedules tasks to
be picked up by the consumers.

The configuration for this call contains the following keys:

{
    "end_offset": "similar to end-offset in the policy",
    "start_offset": "similar to start-offset in the policy",
    "continuous_aggregate": "regclass / fully qualified name of the user view for the CAgg",
    "increment_size": "the size of each individual task, default: chunk_interval",
    "priority": "priority for these tasks. Lower numbers get processed earlier, default: 100"
}

Producer Examples

Schedule multiple jobs for this cagg, with increments of 1 week

We schedule 2 sets

CALL _timescaledb_additional.schedule_refresh_continuous_aggregate_incremental(
    job_id => null,
    config => '
{
    "end_offset": "6 weeks",
    "start_offset": "3 years",
    "continuous_aggregate": "public.test_cagg_incr_refresh_cagg",
    "increment_size": "3 days"
}');

with the most recent data having the highest priority:

CALL _timescaledb_additional.schedule_refresh_continuous_aggregate_incremental(
    job_id => null,
    config => '
{
    "end_offset": "1 day",
    "start_offset": "6 weeks",
    "continuous_aggregate": "public.test_cagg_incr_refresh_cagg",
    "increment_size": "1 week",
    "priority": 1
}');

Consumer

For the consumer(s), we schedule as many jobs as we want to be able to run in
parallel. Likely, a reasonable maximum for these is not too high, for example,
4-6. While we can do incremental CAgg refreshes, we cannot (as of december
2024) schedule parallel refreshes for the same CAgg. This should therefore never
be higher than your number of CAggs.

These jobs will be consuming a connection all the time, as they are designed to
run all the time.

SELECT
    public.add_job(
        proc => '_timescaledb_additional.task_refresh_continuous_aggregate_incremental_runner'::regproc,
        -- This isn't really needed, but this ensures the workers do not run forever,
        -- but once they terminate, they will be restarted within 15 minutes or so.
        schedule_interval => interval '15 minutes',
        config => '{"max_runtime": "11 hours"}',
        initial_start => now()
    )
FROM
    generate_series(1, 4);

[fabriziomello]

Shouldn't we need a column to know when it was processed??? Or it will be deleted?

[feikesteenbergen]

Shouldn't we need a column to know when it was processed

Let's do that, then we can create some reports afterwards.
We do need to think about retention that way though, especiallly if the number of tasks is large.

[feikesteenbergen]

    started timestamptz,
    finished timestamptz,

Adding these, and changing unique index:

) WHERE worker_pid IS NULL AND finished IS NULL;

[fabriziomello]

We can turn it into a hypertable then to make our life easier to apply retention policies

[feikesteenbergen]

Alas, then my trick with the unique index won't work!

[zilder]

increment_size is always NULL in this context, right?

-            coalesce(increment_size, interval_length * interval '1 microsecond')
+            interval_length * interval '1 microsecond'

[feikesteenbergen]

@fabriziomello we can also decide to RETURN here, so that we are not occupying a max_workers slot / max_background_workers slot

[fabriziomello]

yes, this is something that definitely we need to thing about... then our scheduler can launch again the worker

[feikesteenbergen]

My be worthy of a comment, but the tx that takes this lock is always short lived.
It doesn't do heavy lifting, it populates worker_pid and commits.

[fabriziomello]

Do we really need this?? We're using FOR UPDATE when picking an item from the queue. Asking cause it makes FOR UPDATE SKIP LOCKED useless no?

[fabriziomello]

Do we really need a subtransaction here?