citusdata / citusdata/citus

Add rollup generation counter functions

Open
#1,935 0 comments 0 reactions 0 assignees View on GitHub
feature
Dominant language
C
Stars
12.8k
Forks
794
Avg merge
2d 14h
Merged PRs (30d)
31

Description

It would be useful if Citus (or another extension) provided support for doing rollups with late data and concurrent ingestion by adding a generation counter that can be used to keep track of whether raw data was rolled up and whether any transactions of a particular generation are in progress.

When creating the table, you would add a column with as a default value a call to the `current_rollup_generation` function. This function returns the current generation counter and takes a share lock on the counter value, and returns the same value for the remainder of the transaction.

For example, you would create a table as follows:

```sql
CREATE TABLE events (
tenant_id bigint,
payload text not null,
rollup_generation bigint default current_rollup_generation('my_rollups')
);
CREATE INDEX ON events USING BRIN (rollup_generation);
SELECT create_distributed_table('events','tenant_id');
```

Then, when ingesting data, the current generation counter is used for all new rows.

```sql
\COPY events (tenant_id,payload) FROM STDIN
1,hello
2,world
\.

SELECT * FROM events;
tenant_id | payload | rollup_generation
-----------+---------+-------------------
3 | world | 1
2 | hello | 1
```

While the COPY command is running, a share lock is held on the generation counter. That means we can guarantee that all current transactions of a generation are finished by taking an exclusive lock on that generation.

The final step is to compute a rollup for a series of generations. Here we take the exclusive lock on a series of generations to make sure ingest commands have finished, and track which generations have been rolled up in some metadata table.

```sql
CREATE OR REPLACE FUNCTION compute_rollups()
RETURNS void LANGUAGE PLPGSQL AS $function$
DECLARE
start_generation bigint;
end_generation bigint;
BEGIN
/*
* Construct a safe rollup window.
*
* - Get the current generation counter
* - Increase the the generation counter for new transactions (?)
* - Find the oldest generation that has not been rolled up
* - Take exclusive locks on all generations from start to end to ensure that
* insert/copy commands for each generation have finished
* - Update the metadata to indicate that we have done the rollup up to
* end_generation (rolls back on failure)
*/
SELECT start, end INTO start_generation, end_generation
FROM safe_rollup_window('my_rollup');

INSERT INTO rollup_5min
SELECT date_trunc('5 minutes'), count(*) FROM events
WHERE rollup_generation BETWEEN start_generation AND end_generation GROUP BY 1
ON CONFLICT (time_window)
DO UPDATE SET counter = rollup_5min.counter + EXCLUDED.counter;
END;
$function$;
```

This approach has several benefits:
- If data arrives after a rollup has already been performed it can simply be upserted, we know how to distinguish it from data that has already been rolled up.
- Can perform rollup at a higher frequency (more real-time), since we can run a rollup concurrently with ingestions in the same time window
- Data is always roughly ordered along the `rollup_generation` column, which allows efficient reads of data that has not been rolled up using the index
- Because we know precisely which data has been rolled up and can efficiently read that data, we can create real-time views that combine the rollup with the new data

The main downside of this approach is the extra storage space required for the rollup_generation columns.

(?) One issue is that when we increase the generation counter, the new value needs to become visible to new transactions before we take an exclusive lock on the current generation counter. Otherwise, we would block the ingestion pipeline. This may require using a separate transaction for increasing the counter or using a sequence underneath.

A more transparent approach is given in #43, but this has so far proven too costly to become a priority and the lack of a solution for late and concurrent data significantly increases the adoption cost for real-time analytics users.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.