tardis.sql

-- psql -U postgres -h localhost -f /path/to/tardis.sql

SET statement_timeout = 0;
SET lock_timeout = 0;
SET client_encoding = 'UTF8';
SET standard_conforming_strings = on;
SET check_function_bodies = false;
SET client_min_messages = warning;
SET row_security = off;

CREATE DATABASE tardis WITH TEMPLATE = template0 ENCODING = 'UTF8' LC_COLLATE = 'en_US.UTF-8' LC_CTYPE = 'en_US.UTF-8';

\connect tardis

SET statement_timeout = 0;
SET lock_timeout = 0;
SET client_encoding = 'UTF8';
SET standard_conforming_strings = on;
SET check_function_bodies = false;
SET client_min_messages = warning;
SET row_security = off;


CREATE SCHEMA history;

CREATE EXTENSION IF NOT EXISTS plpgsql WITH SCHEMA pg_catalog;

CREATE EXTENSION IF NOT EXISTS btree_gist WITH SCHEMA public;

CREATE EXTENSION IF NOT EXISTS "uuid-ossp" WITH SCHEMA public;

SET search_path = history, pg_catalog;

CREATE FUNCTION log_change() RETURNS trigger
    LANGUAGE plpgsql
    AS $_$
    DECLARE
      c refcursor;
      tt tstzrange;
    BEGIN
        IF TG_OP = 'INSERT' THEN
            EXECUTE 'INSERT INTO history.' || TG_TABLE_NAME ||
              ' SELECT $1.*, tstzrange(now(), $$infinity$$, $$[)$$)' USING NEW;
            RETURN NEW;
        ELSIF TG_OP = 'UPDATE' THEN
            OPEN c FOR EXECUTE 'SELECT _validrange FROM history.' || TG_TABLE_NAME ||
              ' WHERE id = $1 ORDER BY _validrange DESC LIMIT 1 FOR UPDATE'
              USING NEW.id;
            FETCH FROM c INTO tt;

            IF isempty(tstzrange(lower(tt), now(), $$[)$$)) THEN

                -- What edge case are we missing by commenting the following lines of code?
		-- How can we get the lastxid value?

		--IF NOT lastxid = txid_current() THEN
		--    RAISE EXCEPTION 'UPDATE would have empty validity: %d!', OLD;
		--END IF;

		EXECUTE 'DELETE FROM history.' || TG_TABLE_NAME ||
		  ' WHERE CURRENT OF ' || quote_ident(c::text);
            ELSE
                EXECUTE 'UPDATE history.' || TG_TABLE_NAME || ' SET _validrange = tstzrange($1, now(), $$[)$$)' ||
                  ' WHERE CURRENT OF ' || quote_ident(c::text) USING lower(tt);
            END IF;

            EXECUTE 'INSERT INTO history.' || TG_TABLE_NAME ||
              ' SELECT $1.*, tstzrange(now(), $$infinity$$, $$[)$$)' USING NEW;

            RETURN NEW;
        ELSIF TG_OP = 'DELETE' THEN

          OPEN c FOR EXECUTE 'SELECT _validrange FROM history.' || TG_TABLE_NAME ||
            ' WHERE id = $1 ORDER BY _validrange DESC LIMIT 1 FOR UPDATE' USING OLD.id;

          FETCH FROM c into tt;

          IF isempty(tstzrange(lower(tt), now(), $$[)$$)) THEN
            EXECUTE 'DELETE FROM history.' || TG_TABLE_NAME ||
              ' WHERE CURRENT OF ' || quote_ident(c::text);
            RETURN OLD;
          END IF;

          EXECUTE 'UPDATE history.' || TG_TABLE_NAME ||
            ' SET _validrange = tstzrange($1, now(), $$[)$$) WHERE CURRENT OF ' ||
            quote_ident(c::text) USING lower(tt);

          RETURN OLD;
        END IF;
        RETURN NULL;
    END;
$_$;


CREATE TABLE table1 (
    id text NOT NULL,
    name text NOT NULL,
    _validrange tstzrange
);

ALTER TABLE ONLY table1
    ADD CONSTRAINT table1_exclusion EXCLUDE
    USING gist (id WITH =, _validrange WITH &&);


SET search_path = public, pg_catalog;

-- We're using text instead of uuid because btree_gist don't support
-- uuid data type yet and we want to use it for the EXCLUDE constraint
-- See: https://commitfest.postgresql.org/7/332/

CREATE TABLE table1 (
    id text PRIMARY KEY DEFAULT (uuid_generate_v4())::text NOT NULL,
    name text NOT NULL
);

CREATE TRIGGER table1_history BEFORE
  INSERT OR DELETE OR UPDATE ON table1
  FOR EACH ROW EXECUTE PROCEDURE history.log_change();


-- Data

BEGIN;
INSERT INTO table1 (name) VALUES ('name 1');
INSERT INTO table1 (name) VALUES ('name 2');
COMMIT;

BEGIN;
UPDATE table1 SET name = 'name 1.1' WHERE name = 'name 1';
COMMIT;

BEGIN;
-- Since this is a transaction, the first UPDATE is not recorded
-- in the history.table1 (as expected)
UPDATE table1 SET name = 'name 1.2' WHERE name = 'name 1.1';
UPDATE table1 SET name = 'name 1.3' WHERE name = 'name 1.2';
COMMIT;

SELECT * FROM public.table1 WHERE name = 'name 1.3';
SELECT * FROM history.table1 ORDER BY _validrange DESC;