-- The statement trigger of track_changes does not look at the rows of an UPDATE that leaves the key column alone: the replacement of the new version covers the old one. What about the
-- other ways an UPDATE ends an entry? A partial index gets no entry for a version outside its predicate; an expression, a covered column, an equal value (HOT), a row inserted and
-- updated in one transaction, several updates of a row in one transaction, an upsert, a change of the key itself and a row that moves to another partition. Every case: the
-- heap-free read, the read that checks every page and no sift plan give one answer.
CREATE EXTENSION pg_sift;
CREATE EXTENSION pg_sift_test;
SET client_min_messages = warning;
SET max_parallel_workers_per_gather = 0;
-- t_plain: all rows; t_part: the index holds the rows with counter_a > 500; t_expr: the index holds an expression of tenant_id
CREATE TABLE t_plain (id int PRIMARY KEY, tenant_id int NOT NULL, counter_a int, label text) WITH (autovacuum_enabled = false, fillfactor = 100);
CREATE TABLE t_part (LIKE t_plain INCLUDING ALL) WITH (autovacuum_enabled = false, fillfactor = 100);
CREATE TABLE t_expr (LIKE t_plain INCLUDING ALL) WITH (autovacuum_enabled = false, fillfactor = 100);
INSERT INTO t_plain SELECT i, i % 10, i, 'x' FROM generate_series(1, 20000) i;
INSERT INTO t_part SELECT * FROM t_plain;
INSERT INTO t_expr SELECT * FROM t_plain;
CREATE INDEX t_plain_c ON t_plain USING sift_col (id, tenant_id, counter_a) WITH (track_changes = 'id');
CREATE INDEX t_part_c ON t_part USING sift_col (id, tenant_id, counter_a) WITH (track_changes = 'id') WHERE counter_a > 500;
CREATE INDEX t_expr_c ON t_expr USING sift_col (id, (tenant_id + 1), counter_a) WITH (track_changes = 'id');
VACUUM (FREEZE, ANALYZE) t_plain, t_part, t_expr;
-- the question asked of a table: the aggregate its index answers (the partial index only for a query inside its predicate), three ways
CREATE FUNCTION answer(t text, mode int) RETURNS text LANGUAGE plpgsql AS $$
DECLARE plan text := ''; l text; r text;
	q text := format('SELECT %s AS g, count(*) AS n, sum(counter_a) AS s FROM %I %s GROUP BY 1',
	                 CASE WHEN t = 't_expr' THEN 'tenant_id + 1' ELSE 'tenant_id' END, t, CASE WHEN t = 't_part' THEN 'WHERE counter_a > 500' ELSE '' END);
BEGIN
	PERFORM set_config('sift.enable_agg', (mode < 2)::text, true);
	PERFORM set_config('sift.test_no_heapfree', (mode = 1)::text, true);
	FOR l IN EXECUTE 'EXPLAIN (COSTS OFF) ' || q LOOP
		plan := plan || l || E'\n';
	END LOOP;
	IF (plan LIKE '%SiftAgg%') <> (mode < 2) THEN
		RAISE EXCEPTION 'the plan is not the one asked for: %', plan;
	END IF;
	EXECUTE 'SELECT coalesce(string_agg(format(''%s:%s:%s'', g, n, s), '' '' ORDER BY g), ''none'') FROM (' || q || ') q' INTO r;
	RETURN r;
END $$;
CREATE FUNCTION chk(what text, t text, OUT test text, OUT table_name text, OUT free_equals_no_sift bool, OUT checked_equals_no_sift bool) LANGUAGE plpgsql AS $$
BEGIN
	test := what;
	table_name := t;
	free_equals_no_sift := answer(t, 0) = answer(t, 2);
	checked_equals_no_sift := answer(t, 1) = answer(t, 2);
END $$;
-- UPDATEs that leave the key alone, on each kind of index
CREATE FUNCTION same_key(t text) RETURNS void LANGUAGE plpgsql AS $$
BEGIN
	EXECUTE format('UPDATE %I SET counter_a = counter_a + 1 WHERE id %% 7 = 0', t);              -- a covered column
	EXECUTE format('UPDATE %I SET label = ''y'' WHERE id %% 5 = 0', t);                           -- an uncovered one (HOT)
	EXECUTE format('UPDATE %I SET counter_a = counter_a WHERE id %% 3 = 0', t);                   -- an equal value
	EXECUTE format('UPDATE %I SET tenant_id = (tenant_id + 1) %% 10 WHERE id %% 11 = 0', t);     -- the grouping column, and for the expression index its expression
	EXECUTE format('UPDATE %I SET counter_a = counter_a - 400 WHERE id %% 13 = 0 AND counter_a > 600', t);   -- out of the partial predicate
	EXECUTE format('UPDATE %I SET counter_a = counter_a + 500 WHERE id %% 17 = 0 AND counter_a <= 500', t);  -- into it
END $$;
SELECT same_key('t_plain'), same_key('t_part'), same_key('t_expr');
 same_key | same_key | same_key 
----------+----------+----------
          |          | 
(1 row)

SELECT * FROM chk('same key', 't_plain');
   test   | table_name | free_equals_no_sift | checked_equals_no_sift 
----------+------------+---------------------+------------------------
 same key | t_plain    | t                   | t
(1 row)

SELECT * FROM chk('same key', 't_part');
   test   | table_name | free_equals_no_sift | checked_equals_no_sift 
----------+------------+---------------------+------------------------
 same key | t_part     | t                   | t
(1 row)

SELECT * FROM chk('same key', 't_expr');
   test   | table_name | free_equals_no_sift | checked_equals_no_sift 
----------+------------+---------------------+------------------------
 same key | t_expr     | t                   | t
(1 row)

SELECT sift.col_promote('t_plain_c') >= 0 AND sift.col_promote('t_part_c') >= 0 AND sift.col_promote('t_expr_c') >= 0 AS promoted;
 promoted 
----------
 t
(1 row)

SELECT * FROM chk('same key, promoted', 't_plain');
        test        | table_name | free_equals_no_sift | checked_equals_no_sift 
--------------------+------------+---------------------+------------------------
 same key, promoted | t_plain    | t                   | t
(1 row)

SELECT * FROM chk('same key, promoted', 't_part');
        test        | table_name | free_equals_no_sift | checked_equals_no_sift 
--------------------+------------+---------------------+------------------------
 same key, promoted | t_part     | t                   | t
(1 row)

SELECT * FROM chk('same key, promoted', 't_expr');
        test        | table_name | free_equals_no_sift | checked_equals_no_sift 
--------------------+------------+---------------------+------------------------
 same key, promoted | t_expr     | t                   | t
(1 row)

-- one transaction: a row inserted and updated, a row updated several times, an upsert, a key that changes, and a delete after the update
CREATE FUNCTION one_xact(t text) RETURNS void LANGUAGE plpgsql AS $$
BEGIN
	EXECUTE format('INSERT INTO %I SELECT i, i %% 10, 900 + i, ''n'' FROM generate_series(30001, 30400) i', t);
	EXECUTE format('UPDATE %I SET counter_a = counter_a + 1 WHERE id BETWEEN 30001 AND 30300', t);
	EXECUTE format('UPDATE %I SET counter_a = counter_a - 800 WHERE id BETWEEN 30001 AND 30100', t);
	EXECUTE format('UPDATE %I SET counter_a = counter_a + 800 WHERE id BETWEEN 30001 AND 30050', t);
	EXECUTE format('UPDATE %I SET counter_a = counter_a + 2 WHERE id %% 19 = 0', t);
	EXECUTE format('UPDATE %I SET counter_a = counter_a + 3 WHERE id %% 19 = 0', t);
	EXECUTE format('INSERT INTO %I VALUES (19, 9, 1000, ''u''), (38, 8, 5, ''u''), (40001, 1, 700, ''u'') ON CONFLICT (id) DO UPDATE SET counter_a = %I.counter_a + 100', t, t);
	EXECUTE format('UPDATE %I SET id = id + 100000 WHERE id %% 23 = 0', t);
	EXECUTE format('DELETE FROM %I WHERE id BETWEEN 30301 AND 30320', t);
	EXECUTE format('UPDATE %I SET counter_a = 0 WHERE id BETWEEN 30321 AND 30340', t);
END $$;
BEGIN;
SELECT one_xact('t_plain'), one_xact('t_part'), one_xact('t_expr');
 one_xact | one_xact | one_xact 
----------+----------+----------
          |          | 
(1 row)

COMMIT;
SELECT * FROM chk('one transaction', 't_plain');
      test       | table_name | free_equals_no_sift | checked_equals_no_sift 
-----------------+------------+---------------------+------------------------
 one transaction | t_plain    | t                   | t
(1 row)

SELECT * FROM chk('one transaction', 't_part');
      test       | table_name | free_equals_no_sift | checked_equals_no_sift 
-----------------+------------+---------------------+------------------------
 one transaction | t_part     | t                   | t
(1 row)

SELECT * FROM chk('one transaction', 't_expr');
      test       | table_name | free_equals_no_sift | checked_equals_no_sift 
-----------------+------------+---------------------+------------------------
 one transaction | t_expr     | t                   | t
(1 row)

-- and when the transaction is rolled back, or a savepoint
BEGIN;
SELECT same_key('t_plain'), same_key('t_part'), same_key('t_expr');
 same_key | same_key | same_key 
----------+----------+----------
          |          | 
(1 row)

ROLLBACK;
SELECT * FROM chk('rolled back', 't_part');
    test     | table_name | free_equals_no_sift | checked_equals_no_sift 
-------------+------------+---------------------+------------------------
 rolled back | t_part     | t                   | t
(1 row)

BEGIN;
SAVEPOINT a;
SELECT same_key('t_part');
 same_key 
----------
 
(1 row)

ROLLBACK TO a;
SELECT counter_a FROM t_part WHERE id = 14;
 counter_a 
-----------
        15
(1 row)

COMMIT;
SELECT * FROM chk('savepoint', 't_part');
   test    | table_name | free_equals_no_sift | checked_equals_no_sift 
-----------+------------+---------------------+------------------------
 savepoint | t_part     | t                   | t
(1 row)

VACUUM t_plain, t_part, t_expr;
SELECT * FROM chk('vacuumed', 't_plain');
   test   | table_name | free_equals_no_sift | checked_equals_no_sift 
----------+------------+---------------------+------------------------
 vacuumed | t_plain    | t                   | t
(1 row)

SELECT * FROM chk('vacuumed', 't_part');
   test   | table_name | free_equals_no_sift | checked_equals_no_sift 
----------+------------+---------------------+------------------------
 vacuumed | t_part     | t                   | t
(1 row)

SELECT * FROM chk('vacuumed', 't_expr');
   test   | table_name | free_equals_no_sift | checked_equals_no_sift 
----------+------------+---------------------+------------------------
 vacuumed | t_expr     | t                   | t
(1 row)

-- a row that moves to another partition: a statement trigger of a partition does not fire for an UPDATE through the parent, so the option is off there (by default with a NOTICE, an error for a named column)
-- and the reads of such an index check the heap, whatever the rows did
CREATE TABLE hp (id int NOT NULL, tenant_id int NOT NULL, counter_a int, label text) PARTITION BY RANGE (tenant_id);
CREATE TABLE hp0 PARTITION OF hp FOR VALUES FROM (0) TO (5) WITH (autovacuum_enabled = false);
CREATE TABLE hp1 PARTITION OF hp FOR VALUES FROM (5) TO (10) WITH (autovacuum_enabled = false);
CREATE UNIQUE INDEX hp0_id ON hp0 (id);
CREATE UNIQUE INDEX hp1_id ON hp1 (id);
INSERT INTO hp SELECT i, i % 10, i, 'x' FROM generate_series(1, 20000) i;
CREATE INDEX hp0_h ON hp0 USING sift_col (id, tenant_id, counter_a) WITH (track_changes = 'off');
CREATE INDEX hp0_c ON hp0 USING sift_col (id, tenant_id, counter_a);
CREATE INDEX hp1_c ON hp1 USING sift_col (id, tenant_id, counter_a) WITH (track_changes = 'off');
VACUUM (FREEZE, ANALYZE) hp0, hp1;
UPDATE hp SET tenant_id = (tenant_id + 5) % 10 WHERE id % 9 = 0;
SELECT * FROM chk('partition moves', 'hp0');
      test       | table_name | free_equals_no_sift | checked_equals_no_sift 
-----------------+------------+---------------------+------------------------
 partition moves | hp0        | t                   | t
(1 row)

SELECT * FROM chk('partition moves', 'hp1');
      test       | table_name | free_equals_no_sift | checked_equals_no_sift 
-----------------+------------+---------------------+------------------------
 partition moves | hp1        | t                   | t
(1 row)

DELETE FROM hp WHERE id % 31 = 0;
SELECT * FROM chk('partition deletes', 'hp0');
       test        | table_name | free_equals_no_sift | checked_equals_no_sift 
-------------------+------------+---------------------+------------------------
 partition deletes | hp0        | t                   | t
(1 row)

SELECT * FROM chk('partition deletes', 'hp1');
       test        | table_name | free_equals_no_sift | checked_equals_no_sift 
-------------------+------------+---------------------+------------------------
 partition deletes | hp1        | t                   | t
(1 row)

DROP FUNCTION chk(text, text);
DROP FUNCTION answer(text, int);
DROP FUNCTION one_xact(text);
DROP FUNCTION same_key(text);
DROP TABLE hp, t_plain, t_part, t_expr;
DROP EXTENSION pg_sift_test;
DROP EXTENSION pg_sift;