Use a user-defined action to downsample and compress chunks

TimescaleDB lets you downsample and compress chunks by combining a continuous aggregate refresh policy with a compression policy.

If you want to implement features not supported by those policies, you can write a user-defined action to downsample and compress chunks instead. The following example downsamples raw data to an average over hourly data. This is an illustrative example, which can be done more simply with a continuous aggregate policy. But you can make the query arbitrarily complex.

Using a user-defined action to downsample and compress chunks

  1. Create a procedure that first queries the chunks of a hypertable to determine if they are older than the lag parameter. The hypertable in this example is named metrics. If the chunk is not already compressed, downsample it by taking the average of the raw data. Then compress it. A temporary table is used to store the data while calculating the average.

``` 1. CREATE OR REPLACE PROCEDURE downsample_compress (job_id int, config jsonb)

  1. LANGUAGE PLPGSQL

  2. AS $$

  3. DECLARE

  4. lag interval;

  5. chunk REGCLASS;

  6. tmp_name name;

  7. BEGIN

  8. SELECT jsonb_object_field_text (config, 'lag')::interval INTO STRICT lag;

  9. IF lag IS NULL THEN

  10. RAISE EXCEPTION 'Config must have lag';

  11. END IF;

  12. FOR chunk IN

  13. SELECT show.oid

  14. FROM show_chunks('metrics', older_than => lag) SHOW (oid)

  15. INNER JOIN pg_class pgc ON pgc.oid = show.oid

  16. INNER JOIN pg_namespace pgns ON pgc.relnamespace = pgns.oid

  17. INNER JOIN timescaledb_information.chunks chunk ON chunk.chunk_name = pgc.relname

  18. AND chunk.chunk_schema = pgns.nspname

  19. WHERE chunk.is_compressed::bool = FALSE

  20. LOOP

  21. RAISE NOTICE 'Processing chunk: %', chunk::text;

  22. -- build name for temp table

  23. SELECT '_tmp' || relname

  24. FROM pg_class

  25. WHERE oid = chunk INTO STRICT tmp_name;

  26. -- copy downsampled chunk data into temp table

  27. EXECUTE format($sql$ CREATE UNLOGGED TABLE %I AS

  28. SELECT time_bucket('1h', time), device_id, avg(value) FROM %s GROUP BY 1, 2;

  29. $sql$, tmp_name, chunk);

  30. -- clear original chunk

  31. EXECUTE format('TRUNCATE %s;', chunk);

  32. -- copy downsampled data back into chunk

  33. EXECUTE format('INSERT INTO %s(time, device_id, value) SELECT * FROM %I;', chunk, tmp_name);

  34. -- drop temp table

  35. EXECUTE format('DROP TABLE %I;', tmp_name);

  36. PERFORM compress_chunk (chunk);

  37. COMMIT;

  38. END LOOP;

  39. END

  40. $$; ```

  41. Register the job to run daily. In the config, set lag to 12 months to drop chunks containing data older than 12 months.

1. SELECT add_job('downsample_compress','1d', config => '{"lag":"12 month"}');