Create and manage jobs

Create and manage custom Postgres functions and procedures that run on a schedule in TimescaleDB

s in are custom Postgres functions or procedures that run on a schedule you define. They help you automate routine database maintenance tasks, data processing operations, and custom workflows that go beyond the built-in automation policies.

While includes native scheduling policies for common operations like refreshing s, compressing data with , dropping old data, and reordering data within s, you may need custom s to:

• Perform complex data transformations that aren’t covered by built-in policies
• Trigger external processes or notifications based on database events
• Coordinate multiple operations in a specific sequence
• Run custom business logic on a schedule
• Implement organization-specific maintenance routines

How jobs work

s in follow a simple lifecycle: you create a function or procedure, register it with the job scheduler, and the system handles execution on your defined schedule.

The job scheduler runs as a background process and automatically executes your s according to the schedule you define. You can monitor execution history through the timescaledb_information.jobs and timescaledb_information.job_stats views, and manually trigger jobs for testing or immediate execution.

Prerequisites

To follow the steps on this page:

• Create a target with Real-time analytics enabled.

  You need your connection details. This procedure also works for .

Create a job

To create a , create a function or procedure that you want your database to execute, then set it up to run on a schedule.

1. Define a function or procedure in the language of your choice

   Wrap it in a CREATE statement:

   CREATE FUNCTION <function_name> (job_id INT DEFAULT NULL, config JSONB DEFAULT NULL)
   RETURNS VOID
   DECLARE
       <declaration>;
   BEGIN
       <function_body>;
   END;
   $<variable_name>$ LANGUAGE <language>;
   

   For example, to create a function that reindexes a table within your database:

   CREATE FUNCTION reindex_mytable(job_id INT DEFAULT NULL, config JSONB DEFAULT NULL)
   RETURNS VOID
   AS $$
   BEGIN
      REINDEX TABLE mytable;
   END;
   $$ LANGUAGE plpgsql;
   

   job_id and config are required arguments in the function signature. This returns CREATE FUNCTION to indicate that the function has successfully been created.

2. Call the function to validate

   For example:

   select reindex_mytable();
   

   The result looks like this:

    reindex_mytable
   -----------------
   
   (1 row)
   

3. Register your job with add_job

   Pass the name of your , the schedule you want it to run on, and the content of your config. For the config value, if you don’t need any special configuration parameters, set to NULL. For example, to run the reindex_mytable function every hour:

   SELECT add_job('reindex_mytable', '1h', config => NULL);
   

   The call returns a job_id and stores it along with config in the catalog.

   The runs on the schedule you set. You can also run it manually with run_job passing job_id. When the runs, job_id and config are passed as arguments.

4. Validate the job

   List all currently registered s with timescaledb_information.jobs:

   SELECT * FROM timescaledb_information.jobs;
   

   The result looks like this:

   job_id |      application_name      | schedule_interval | max_runtime | max_retries | retry_period |      proc_schema      |    proc_name     |   owner   | scheduled |         config         |          next_start           | hypertable_schema | hypertable_name
   --------+----------------------------+-------------------+-------------+-------------+--------------+-----------------------+------------------+-----------+-----------+------------------------+-------------------------------+-------------------+-----------------
   1 | Telemetry Reporter [1]     | 24:00:00          | 00:01:40    |          -1 | 01:00:00     | _timescaledb_internal | policy_telemetry | postgres  | t         |                        | 2022-08-18 06:26:39.524065+00 |                   |
   1000 | User-Defined Action [1000] | 01:00:00          | 00:00:00    |          -1 | 00:05:00     | public                | reindex_mytable  | tsdbadmin | t         |                        | 2022-08-17 07:17:24.831698+00 |                   |
   (2 rows)
   

Test and debug a job

To debug a , increase the log level and run the manually with run_job in the foreground. Because run_job is a stored procedure and not a function, run it with CALL instead of SELECT.

1. Set the minimum log level to DEBUG1

   SET client_min_messages TO DEBUG1;
   

2. Run the job

   Replace 1000 with your job_id:

   CALL run_job(1000);
   

Alter and delete a job

Alter an existing with alter_job. You can change both the config and the schedule on which the runs.

1. Change a job’s config

   To replace the entire JSON config for a , call alter_job with a new config object. For example, replace the JSON config for a with ID 1000:

   SELECT alter_job(1000, config => '{"hypertable":"metrics"}');
   

2. Turn off job scheduling

   To turn off automatic scheduling of a , call alter_job and set scheduledto false. You can still run the manually with run_job. For example, turn off the scheduling for a with ID 1000:

   SELECT alter_job(1000, scheduled => false);
   

3. Re-enable automatic scheduling of a job

   To re-enable automatic scheduling of a , call alter_job and set scheduled to true. For example, re-enable scheduling for a with ID 1000:

   SELECT alter_job(1000, scheduled => true);
   

4. Delete a job with delete_job

   For example, to delete a with ID 1000:

   SELECT delete_job(1000);
   

Samples

Downsample and compress chunks

lets you downsample and compress s by combining a refresh policy with . If you want to implement features not supported by those policies, you can write a to downsample and convert s to columnstore 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 policy. But you can make the query arbitrarily complex.

1. Create a procedure to downsample chunks and convert them to columnstore

   This procedure that first queries the s of a to determine if they are older than the lag parameter. The in this example is named metrics. If the is not already compressed, downsample it by taking the average of the raw data. Then compress by converting to the columnstore. This procedure uses a temporary table to store the data while calculating the average.

   CREATE OR REPLACE PROCEDURE downsample_compress (job_id int, config jsonb)
   LANGUAGE PLPGSQL
   AS $$
   DECLARE
     lag interval;
     chunk REGCLASS;
     tmp_name name;
   BEGIN
     SELECT jsonb_object_field_text (config, 'lag')::interval INTO STRICT lag;
   
     IF lag IS NULL THEN
       RAISE EXCEPTION 'Config must have lag';
     END IF;
   
     FOR chunk IN
       SELECT show.oid
       FROM show_chunks('metrics', older_than => lag) SHOW (oid)
         INNER JOIN pg_class pgc ON pgc.oid = show.oid
         INNER JOIN pg_namespace pgns ON pgc.relnamespace = pgns.oid
         INNER JOIN timescaledb_information.chunks chunk ON chunk.chunk_name = pgc.relname
           AND chunk.chunk_schema = pgns.nspname
       WHERE chunk.is_compressed::bool = FALSE
     LOOP
       RAISE NOTICE 'Processing chunk: %', chunk::text;
   
       -- build name for temp table
       SELECT '_tmp' || relname
       FROM pg_class
       WHERE oid = chunk INTO STRICT tmp_name;
   
       -- copy downsampled chunk data into temp table
       EXECUTE format($sql$ CREATE UNLOGGED TABLE %I AS
         SELECT time_bucket('1h', time), device_id, avg(value) FROM %s GROUP BY 1, 2;
       $sql$, tmp_name, chunk);
   
       -- clear original chunk
       EXECUTE format('TRUNCATE %s;', chunk);
   
       -- copy downsampled data back into chunk
       EXECUTE format('INSERT INTO %s(time, device_id, value) SELECT * FROM %I;', chunk, tmp_name);
   
       -- drop temp table
       EXECUTE format('DROP TABLE %I;', tmp_name);
   
       PERFORM convert_to_columnstore (chunk);
   
       COMMIT;
     END LOOP;
   END
   $$;
   

2. Register the job to run daily

   In the config, set lag to 12 months to drop s containing data older than 12 months.

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

Generic retention policy

natively supports adding a data retention policy to a . If you want to add a generic data retention policy to all s, you can create a custom .

1. Create a procedure that drops chunks from any hypertable

   This procedure drops s from any if they are older than the drop_after parameter. To get all s, the timescaledb_information.hypertables table is queried.

   CREATE OR REPLACE PROCEDURE generic_retention (job_id int, config jsonb)
   LANGUAGE PLPGSQL
   AS $$
   DECLARE
     drop_after interval;
   BEGIN
     SELECT jsonb_object_field_text (config, 'drop_after')::interval
       INTO STRICT drop_after;
   
     IF drop_after IS NULL THEN
       RAISE EXCEPTION 'Config must have drop_after';
     END IF;
   
     PERFORM drop_chunks(
       format('%I.%I', hypertable_schema, hypertable_name),
       older_than => drop_after
     ) FROM timescaledb_information.hypertables;
   END
   $$;
   

2. Register the job to run daily

   In the config, set drop_after to 12 months to drop s containing data older than 12 months.

   SELECT add_job('generic_retention','1d', config => '{"drop_after":"12 month"}');
   

You can further refine this policy by adding filters to your procedure. For example, add a WHERE clause to the PERFORM query to only drop s from particular s.

Automatic tablespace management

Moving older data to a different tablespace can help you save on storage costs. supports automatic tablespace management by providing the move_chunk function to move chunks between tablespaces. To schedule the moves automatically, you can write a custom .

On , use tiered storage which handles this by providing a tiering policy API to move data to low-cost object storage backed by Amazon S3.

To implement automatic moving with a :

1. Create a procedure that moves chunks to a different tablespace

   This procedure moves s to a different tablespace if they contain data older than the lag parameter.

   CREATE OR REPLACE PROCEDURE move_chunks (job_id int, config jsonb)
   LANGUAGE PLPGSQL
   AS $$
   DECLARE
      ht REGCLASS;
      lag interval;
      destination_tablespace name;
      index_destination_tablespace name;
      reorder_index REGCLASS;
      chunk REGCLASS;
      tmp_name name;
   BEGIN
      SELECT jsonb_object_field_text (config, 'hypertable')::regclass INTO STRICT ht;
      SELECT jsonb_object_field_text (config, 'lag')::interval INTO STRICT lag;
      SELECT jsonb_object_field_text (config, 'destination_tablespace') INTO STRICT destination_tablespace;
      SELECT jsonb_object_field_text (config, 'index_destination_tablespace') INTO STRICT index_destination_tablespace;
      SELECT jsonb_object_field_text (config, 'reorder_index') INTO STRICT reorder_index;
   
    IF ht IS NULL OR lag IS NULL OR destination_tablespace IS NULL THEN
      RAISE EXCEPTION 'Config must have hypertable, lag and destination_tablespace';
    END IF;
   
    IF index_destination_tablespace IS NULL THEN
      index_destination_tablespace := destination_tablespace;
    END IF;
   
    FOR chunk IN
       SELECT c.oid
       FROM pg_class AS c
         LEFT JOIN pg_tablespace AS t ON (c.reltablespace = t.oid)
         JOIN pg_namespace AS n ON (c.relnamespace = n.oid)
         JOIN (SELECT * FROM show_chunks(ht, older_than => lag) SHOW (oid)) AS chunks ON (chunks.oid::text = n.nspname || '.' || c.relname)
       WHERE t.spcname != destination_tablespace OR t.spcname IS NULL
    LOOP
      RAISE NOTICE 'Moving chunk: %', chunk::text;
      PERFORM move_chunk(
          chunk => chunk,
          destination_tablespace => destination_tablespace,
          index_destination_tablespace => index_destination_tablespace,
          reorder_index => reorder_index
      );
    END LOOP;
   END
   $$;
   

2. Register the job to run daily

   In the config, set hypertable to metrics to implement automatic moves on the metrics . Set lag to 12 months to move s containing data older than 12 months. Set tablespace to the destination tablespace.

   SELECT add_job(
     'move_chunks',
     '1d',
     config => '{"hypertable":"metrics","lag":"12 month","destination_tablespace":"old_chunks"}'
   );