Trigger-powered Point-In-Time table (PIT) for persistent staging area (PSA)

pit-psa-postgres-trigger.sql

-- Thanks to: https://stackoverflow.com/questions/58655766/create-trigger-for-update-another-row-automatically-on-same-table-using-postgres

/*
Features:
- DONE Auto-updating xtime.
- DONE Insert custom vtime.
- DONE Insert custom vtime & xtime.
- DONE Idempotency.
- DONE Vtime insertion order is independent.
- DONE Do-not-insert new row when attrs is identical to previous row.
*/

begin;

-- Define table.
create table if not exists plugh (
    nk           text         not null
  , attrs_digest uuid         generated always as (md5(attrs)::uuid) stored
  , vtime        timestamp(0) default now()::timestamp(0)
  , xtime        timestamp(0) default '9999-12-31 23:59:59'::timestamp(0) check (xtime>vtime)
  , attrs        text         not null default ''
  , primary key        (nk, vtime)
  -- , unique      (nk, xtime)
);

-- Define the TRIGGER FUNCTION.
create or replace function plugh__recalc_xtimes()
  returns trigger
  language plpgsql
  as
$$
begin
  delete from plugh
  where nk = new.nk
  and vtime = new.vtime
  and attrs_digest = (
    select attrs_digest
    from plugh
    where nk = new.nk
    and vtime = (select max(vtime) from plugh sub where nk = new.nk and vtime < new.vtime)
  )
  ;
  update plugh
  set xtime = coalesce((select min(vtime) from plugh sub where nk = plugh.nk and vtime > plugh.vtime), '9999-12-31 23:59:59'::timestamp(0))
  where nk = new.nk
  and xtime <> coalesce((select min(vtime) from plugh sub where nk = plugh.nk and vtime > plugh.vtime), '9999-12-31 23:59:59'::timestamp(0))
  ;
  return new;
end;
$$;

-- Define the INSERT & DELETE TRIGGERS.
drop trigger if exists trigger__plugh__insert__recalc_xtimes on plugh;
drop trigger if exists trigger__plugh__delete__recalc_xtimes on plugh;
create trigger trigger__plugh__insert__recalc_xtimes after insert on plugh for each row execute procedure plugh__recalc_xtimes() ;
create trigger trigger__plugh__delete__recalc_xtimes after delete on plugh for each row execute procedure plugh__recalc_xtimes() ;

-- Insert some test records.
insert into plugh (nk,vtime,attrs) values ('a','2003-01-01','103'),   ('b','2003-01-01','203')                                       on conflict (nk, vtime) do update set attrs = excluded.attrs; -- orig_order=5
insert into plugh (nk,vtime,attrs) values ('a','2001-01-01','100'),   ('b','2001-01-01','200')                                       on conflict (nk, vtime) do update set attrs = excluded.attrs; -- orig_order=1
insert into plugh (nk,vtime,attrs) values ('a','2002-01-01','101'),   ('b','2002-01-01','201')                                       on conflict (nk, vtime) do update set attrs = excluded.attrs; -- orig_order=3
insert into plugh (nk,vtime,attrs) values ('a','2002-06-01','101.5'), ('b','2002-06-01','201.5')                                     on conflict (nk, vtime) do update set attrs = excluded.attrs; -- orig_order=4
insert into plugh (nk,vtime,attrs) values ('a','2002-01-01','101'),   ('b','2002-01-01','201')                                       on conflict (nk, vtime) do update set attrs = excluded.attrs; -- orig_order=2
insert into plugh (nk,vtime,xtime,attrs) values ('a','2002-02-10','2002-02-15','101.10'),   ('b','2002-02-10','2002-02-15','201.10') on conflict (nk, vtime) do update set attrs = excluded.attrs; -- orig_order=2

-- Check it out after some inserts.
-- select * from plugh order by nk, vtime;

-- Perform a delete ("redaction").
delete from plugh where nk = 'b' and vtime = '2002-06-01 00:00:00'::timestamp(0);

-- Perform an update ("revision").
update plugh set attrs = '401' where nk = 'a' and vtime = '2002-01-01 00:00:00';

-- Check it out after the redaction and revision.
-- select * from plugh order by nk, vtime;

-- Insert some "current" records (undefined vtime).
insert into plugh (nk,attrs) values ('a','107'), ('b','207') on conflict (nk, vtime) do update set attrs = excluded.attrs; -- orig_order=5

-- Check it out after "current" records were inserted.
-- select * from plugh order by nk, vtime;

-- Feature: No updating of vtime when attrs are identical to previous version's record.
insert into plugh (nk,vtime,attrs) values ('a','2001-01-09','100') on conflict (nk, vtime) do update set attrs = excluded.attrs; -- orig_order=5
insert into plugh (nk,vtime,attrs) values ('a','2001-01-09','100') on conflict (nk, vtime) do update set attrs = excluded.attrs; -- orig_order=5
insert into plugh (nk,vtime,attrs) values ('a','2001-01-09','100') on conflict (nk, vtime) do update set attrs = excluded.attrs; -- orig_order=5
insert into plugh (nk,vtime,attrs) values ('a','2001-01-09','100') on conflict (nk, vtime) do update set attrs = excluded.attrs; -- orig_order=5
insert into plugh (nk,vtime,attrs) values ('a','2001-01-09','100') on conflict (nk, vtime) do update set attrs = excluded.attrs; -- orig_order=5

-- Check it out.
select * from plugh order by nk desc, vtime limit 20;

-- Since we cannot define a unique index on (nk,xtime) due to the pecularities of the trigger logic, let's make sure that no duplicates caused. (The following query *should* return zero rows.)
select nk,xtime,count(*) from plugh group by 1,2 having count(*)>1;

-- rollback; -- Explict but technically unnecessary rollback for clarity.
commit;