pg_incremental

pig build pkg pg_incremental;# build rpm/deb

pig repo add pgsql -u # add both repo and update cache

pig install pg_incremental;# install via package name, for the active PG version  pig install pg_incremental -v 18; # install for PG 18 pig install pg_incremental -v 17; # install for PG 17 pig install pg_incremental -v 16; # install for PG 16

CREATE EXTENSION pg_incremental CASCADE; -- requires pg_cron

CREATE EXTENSION pg_incremental CASCADE; -- depends on pg_cron

SELECT incremental.create_sequence_pipeline('event-aggregation', 'events', $$  INSERT INTO events_agg  SELECT date_trunc('day', event_time), count(*)  FROM events  WHERE event_id BETWEEN $1 AND $2  GROUP BY 1  ON CONFLICT (day) DO UPDATE SET event_count = events_agg.event_count + excluded.event_count $$);

SELECT incremental.create_sequence_pipeline(  'event-aggregation', 'events',  $$ ... $$,  schedule := '* * * * *',  max_batch_size := 10000 );

SELECT incremental.create_time_interval_pipeline('event-aggregation', '1 day', $$  INSERT INTO events_agg  SELECT event_time::date, count(distinct event_id)  FROM events  WHERE event_time >= $1 AND event_time < $2  GROUP BY 1 $$);

SELECT incremental.create_time_interval_pipeline('event-export',  time_interval := '1 day',  batched := false,  start_time := '2024-11-01',  command := $$ SELECT export_events($1, $2) $$ );

SELECT incremental.create_file_list_pipeline('event-import', 's3://mybucket/events/*.csv', $$  SELECT import_events($1) $$);

SELECT * FROM incremental.sequence_pipelines; SELECT * FROM incremental.time_interval_pipelines; SELECT * FROM incremental.processed_files;

SELECT jobname, start_time, status, return_message FROM cron.job_run_details JOIN cron.job USING (jobid) WHERE jobname LIKE 'pipeline:%' ORDER BY 1 DESC LIMIT 5;