-- sift_table system columns xmin, xmax, cmin and cmax against a heap twin. The ids change from run to run, so the checks are relations: the
-- rows of one inserting transaction share an xmin, a row's xmin is its inserter (a subtransaction included), an updated row has a new xmin,
-- a locked or deleted row has the xmax of its locker or deleter, VACUUM FREEZE leaves FrozenTransactionId. The command ids are the same
-- for the same statements of one transaction. Every statement runs on both tables; the groups printed are the heap's and must be the twin's.
CREATE EXTENSION pg_sift;
CREATE TABLE h (id int PRIMARY KEY, v text) USING heap;
CREATE TABLE s (id int PRIMARY KEY, v text) USING sift_table;
-- the ids that share a value of the column of the table, groups separated by '|'; the twin must print the same
CREATE FUNCTION groups(t regclass, col text) RETURNS text LANGUAGE plpgsql AS $$
DECLARE r text;
BEGIN
	EXECUTE format('SELECT string_agg(ids, '' | '' ORDER BY lo) FROM (SELECT string_agg(id::text, '','' ORDER BY id) ids, min(id) lo FROM %s GROUP BY %s::text) x', t, col) INTO r;
	RETURN r;
END $$;
CREATE FUNCTION same(col text) RETURNS text LANGUAGE sql AS $$
	SELECT CASE WHEN groups('h', col) = groups('s', col) THEN col || ': ' || groups('h', col)
		ELSE format('%s: DIFFERENT heap %s, sift_table %s', col, groups('h', col), groups('s', col)) END
$$;
-- two inserting transactions: each row's xmin is the inserter's, the rows of one share it
BEGIN;
INSERT INTO h VALUES (1, 'a'), (2, 'a'), (3, 'a');
INSERT INTO s VALUES (1, 'a'), (2, 'a'), (3, 'a');
SELECT (SELECT bool_and(xmin = pg_current_xact_id()::xid) FROM h) AS heap_xmin_is_inserter,
	(SELECT bool_and(xmin = pg_current_xact_id()::xid) FROM s) AS sift_xmin_is_inserter;
 heap_xmin_is_inserter | sift_xmin_is_inserter 
-----------------------+-----------------------
 t                     | t
(1 row)

COMMIT;
BEGIN;
INSERT INTO h VALUES (4, 'b'), (5, 'b'), (6, 'b');
INSERT INTO s VALUES (4, 'b'), (5, 'b'), (6, 'b');
COMMIT;
SELECT same('xmin');
        same         
---------------------
 xmin: 1,2,3 | 4,5,6
(1 row)

SELECT same('xmax');
       same        
-------------------
 xmax: 1,2,3,4,5,6
(1 row)

-- through a sequential scan, an index scan, a TID scan and a bitmap scan
SELECT count(DISTINCT xmin::text) AS xmins_seq FROM s;
 xmins_seq 
-----------
         2
(1 row)

SET enable_seqscan = off;
SET enable_bitmapscan = off;
SELECT count(DISTINCT xmin::text) AS xmins_index FROM s WHERE id IN (1, 2, 4);
 xmins_index 
-------------
           2
(1 row)

SET enable_bitmapscan = on;
SET enable_indexscan = off;
SELECT count(DISTINCT xmin::text) AS xmins_bitmap FROM s WHERE id IN (1, 2, 4);
 xmins_bitmap 
--------------
            2
(1 row)

RESET enable_seqscan;
RESET enable_indexscan;
SELECT xmin = (SELECT xmin FROM s WHERE id = 4) AS tid_scan_same FROM s WHERE ctid = (SELECT ctid FROM s WHERE id = 5);
 tid_scan_same 
---------------
 t
(1 row)

-- an update: the new version has the updater's xmin
UPDATE h SET v = 'u' WHERE id = 2;
UPDATE s SET v = 'u' WHERE id = 2;
SELECT same('xmin');
         same          
-----------------------
 xmin: 1,3 | 2 | 4,5,6
(1 row)

SELECT same('xmax');
       same        
-------------------
 xmax: 1,2,3,4,5,6
(1 row)

-- a row lock: xmax is the locker's, in its transaction and after it
BEGIN;
SELECT id FROM h WHERE id = 1 FOR UPDATE;
 id 
----
  1
(1 row)

SELECT id FROM s WHERE id = 1 FOR UPDATE;
 id 
----
  1
(1 row)

SELECT (SELECT xmax = pg_current_xact_id()::xid FROM h WHERE id = 1) AS heap_locker,
	(SELECT xmax = pg_current_xact_id()::xid FROM s WHERE id = 1) AS sift_locker;
 heap_locker | sift_locker 
-------------+-------------
 t           | t
(1 row)

SELECT same('xmax');
        same         
---------------------
 xmax: 1 | 2,3,4,5,6
(1 row)

COMMIT;
SELECT same('xmax');
        same         
---------------------
 xmax: 1 | 2,3,4,5,6
(1 row)

SELECT same('xmin');
         same          
-----------------------
 xmin: 1,3 | 2 | 4,5,6
(1 row)

-- RETURNING reads the columns of the row just written
BEGIN;
WITH r AS (INSERT INTO s VALUES (7, 'r') RETURNING xmin) SELECT bool_and(xmin = pg_current_xact_id()::xid) AS insert_returning FROM r;
 insert_returning 
------------------
 t
(1 row)

UPDATE s SET v = 'r2' WHERE id = 7 RETURNING xmin = pg_current_xact_id()::xid AS update_returning, xmax;
 update_returning | xmax 
------------------+------
 t                |    0
(1 row)

ROLLBACK;
-- command ids: the same statements give the same cmin in a transaction of each table
BEGIN;
INSERT INTO h VALUES (10, 'c');
INSERT INTO h VALUES (11, 'c');
SELECT count(*) FROM h;
 count 
-------
     8
(1 row)

INSERT INTO h VALUES (12, 'c');
SELECT id, cmin::text AS cmin FROM h WHERE id >= 10 ORDER BY id;
 id | cmin 
----+------
 10 | 0
 11 | 1
 12 | 2
(3 rows)

ROLLBACK;
BEGIN;
INSERT INTO s VALUES (10, 'c');
INSERT INTO s VALUES (11, 'c');
SELECT count(*) FROM s;
 count 
-------
     8
(1 row)

INSERT INTO s VALUES (12, 'c');
SELECT id, cmin::text AS cmin FROM s WHERE id >= 10 ORDER BY id;
 id | cmin 
----+------
 10 | 0
 11 | 1
 12 | 2
(3 rows)

ROLLBACK;
-- a row a cursor still sees after a later command of its transaction deleted it: xmax and cmax are the deleter's
BEGIN;
INSERT INTO h VALUES (99, 'x');
DECLARE ch CURSOR FOR SELECT id, xmax = pg_current_xact_id()::xid AS deleted_here, cmax::text AS cmax FROM h WHERE id > 3;
FETCH 1 FROM ch;
 id | deleted_here | cmax 
----+--------------+------
  4 | f            | 0
(1 row)

DELETE FROM h WHERE id = 5;
FETCH ALL FROM ch;
 id | deleted_here | cmax 
----+--------------+------
  5 | t            | 1
  6 | f            | 0
 99 | f            | 0
(3 rows)

ROLLBACK;
BEGIN;
INSERT INTO s VALUES (99, 'x');
DECLARE cs CURSOR FOR SELECT id, xmax = pg_current_xact_id()::xid AS deleted_here, cmax::text AS cmax FROM s WHERE id > 3;
FETCH 1 FROM cs;
 id | deleted_here | cmax 
----+--------------+------
  4 | f            | 0
(1 row)

DELETE FROM s WHERE id = 5;
FETCH ALL FROM cs;
 id | deleted_here | cmax 
----+--------------+------
  5 | t            | 1
  6 | f            | 0
 99 | f            | 0
(3 rows)

ROLLBACK;
-- subtransactions: a row's xmin is the subtransaction that inserted it, after RELEASE too; a rolled back one leaves nothing
BEGIN;
INSERT INTO h VALUES (20, 'top');
SAVEPOINT a;
INSERT INTO h VALUES (21, 'sub');
SAVEPOINT b;
INSERT INTO h VALUES (22, 'sub2');
ROLLBACK TO b;
INSERT INTO h VALUES (23, 'sub3');
RELEASE a;
INSERT INTO h VALUES (24, 'top');
SELECT groups('h', 'xmin') AS heap_groups;
            heap_groups            
-----------------------------------
 1,3 | 2 | 4,5,6 | 20,24 | 21 | 23
(1 row)

SELECT bool_and(CASE WHEN id IN (20, 24) THEN xmin = pg_current_xact_id()::xid ELSE xmin <> pg_current_xact_id()::xid END) AS subs_differ FROM h WHERE id >= 20;
 subs_differ 
-------------
 t
(1 row)

COMMIT;
DELETE FROM h WHERE id >= 20;
BEGIN;
INSERT INTO s VALUES (20, 'top');
SAVEPOINT a;
INSERT INTO s VALUES (21, 'sub');
SAVEPOINT b;
INSERT INTO s VALUES (22, 'sub2');
ROLLBACK TO b;
INSERT INTO s VALUES (23, 'sub3');
RELEASE a;
INSERT INTO s VALUES (24, 'top');
SELECT groups('s', 'xmin') AS sift_groups;
            sift_groups            
-----------------------------------
 1,3 | 2 | 4,5,6 | 20,24 | 21 | 23
(1 row)

SELECT bool_and(CASE WHEN id IN (20, 24) THEN xmin = pg_current_xact_id()::xid ELSE xmin <> pg_current_xact_id()::xid END) AS subs_differ FROM s WHERE id >= 20;
 subs_differ 
-------------
 t
(1 row)

COMMIT;
DELETE FROM s WHERE id >= 20;
-- a read through the aggregate pushdown path and a sort
SELECT count(xmin::text) AS n, count(DISTINCT xmin::text) AS xmins FROM s;
 n | xmins 
---+-------
 6 |     3
(1 row)

SELECT (SELECT string_agg(id::text, ',' ORDER BY xmin::text::bigint, id) FROM h) = (SELECT string_agg(id::text, ',' ORDER BY xmin::text::bigint, id) FROM s) AS same_order;
 same_order 
------------
 t
(1 row)

-- the aggregate pushdown reads the columns of the table only: a query that reads a system column keeps the scan, and answers as without it
CREATE FUNCTION pushed(q text) RETURNS bool LANGUAGE plpgsql AS $$
DECLARE l text;
BEGIN
	FOR l IN EXECUTE 'EXPLAIN (COSTS OFF) ' || q LOOP
		IF l LIKE '%SiftAgg%' THEN RETURN true; END IF;
	END LOOP;
	RETURN false;
END $$;
SET max_parallel_workers_per_gather = 0;
SET cpu_tuple_cost = 0.5;
SELECT pushed('SELECT v, count(*) FROM s GROUP BY v') AS control_pushed,
	pushed('SELECT v, count(xmin::text) FROM s GROUP BY v') AS xmin_aggregate_pushed,
	pushed('SELECT v, count(*) FROM s WHERE xmax::text = ''0'' GROUP BY v') AS xmax_filter_pushed;
 control_pushed | xmin_aggregate_pushed | xmax_filter_pushed 
----------------+-----------------------+--------------------
 t              | f                     | f
(1 row)

SELECT v, count(*), count(DISTINCT xmin::text) AS xmins, count(*) FILTER (WHERE xmax::text <> '0') AS with_xmax FROM s GROUP BY v ORDER BY v;
 v | count | xmins | with_xmax 
---+-------+-------+-----------
 a |     2 |     1 |         1
 b |     3 |     1 |         1
 u |     1 |     1 |         0
(3 rows)

SET sift.enable_agg = off;
SELECT v, count(*), count(DISTINCT xmin::text) AS xmins, count(*) FILTER (WHERE xmax::text <> '0') AS with_xmax FROM s GROUP BY v ORDER BY v;
 v | count | xmins | with_xmax 
---+-------+-------+-----------
 a |     2 |     1 |         1
 b |     3 |     1 |         1
 u |     1 |     1 |         0
(3 rows)

RESET sift.enable_agg;
RESET cpu_tuple_cost;
RESET max_parallel_workers_per_gather;
-- VACUUM FREEZE: a heap keeps the inserter's xid in a frozen row (the frozen flag is a hint bit), a sift_table row reads FrozenTransactionId (2)
VACUUM FREEZE h;
VACUUM FREEZE s;
SELECT (SELECT count(*) FROM s WHERE xmin::text = '2') AS sift_frozen, (SELECT count(*) FROM s) AS rows;
 sift_frozen | rows 
-------------+------
           6 |    6
(1 row)

SELECT same('xmax');
       same        
-------------------
 xmax: 1,2,3,4,5,6
(1 row)

-- the rows promoted to a column group: an update and a lock after the freeze
UPDATE h SET v = 'after' WHERE id = 3;
UPDATE s SET v = 'after' WHERE id = 3;
BEGIN;
SELECT id FROM h WHERE id = 4 FOR UPDATE;
 id 
----
  4
(1 row)

SELECT id FROM s WHERE id = 4 FOR UPDATE;
 id 
----
  4
(1 row)

SELECT same('xmax');
        same         
---------------------
 xmax: 1,2,3,5,6 | 4
(1 row)

ROLLBACK;
SELECT same('xmax');
        same         
---------------------
 xmax: 1,2,3,5,6 | 4
(1 row)

SELECT groups('s', 'xmin') AS sift_xmin;
   sift_xmin   
---------------
 1,2,4,5,6 | 3
(1 row)

DROP FUNCTION pushed(text);
DROP FUNCTION same(text);
DROP FUNCTION groups(regclass, text);
DROP TABLE h, s;
DROP EXTENSION pg_sift;