-- sift.table_find_stale_versions lists the primary key values that are on more than one row, reading the table by a sequential scan that neither the aggregate
-- engine nor the primary key index can answer. A deferred primary key lets a transaction hold copies of a key, as the old versions of updated rows were after a VACUUM
-- of 4.0.0 to 4.2.1 made them live again.
CREATE EXTENSION pg_sift;
CREATE TABLE sv_sift (id int PRIMARY KEY DEFERRABLE INITIALLY DEFERRED, v int) USING sift_table;
CREATE TABLE sv_heap (id int PRIMARY KEY DEFERRABLE INITIALLY DEFERRED, v int);
CREATE TABLE sv_comp (a int, b text, v int, PRIMARY KEY (a, b) DEFERRABLE INITIALLY DEFERRED) USING sift_table;
CREATE TABLE sv_part (id int, v int, PRIMARY KEY (id) DEFERRABLE INITIALLY DEFERRED) PARTITION BY RANGE (id);
CREATE TABLE sv_part_1 PARTITION OF sv_part FOR VALUES FROM (0) TO (1000) USING sift_table;
CREATE TABLE sv_part_2 PARTITION OF sv_part FOR VALUES FROM (1000) TO (2000) USING sift_table;
CREATE TABLE sv_nopk (id int) USING sift_table;
INSERT INTO sv_sift SELECT g, 0 FROM generate_series(1, 3000) g;
INSERT INTO sv_heap SELECT g, 0 FROM generate_series(1, 3000) g;
INSERT INTO sv_comp SELECT g, 'k' || g % 7, 0 FROM generate_series(1, 500) g;
INSERT INTO sv_part SELECT g, 0 FROM generate_series(0, 1999) g;
-- no copies
SELECT * FROM sift.table_find_stale_versions('sv_sift');
 duplicate_keys | sample_keys 
----------------+-------------
              0 | 
(1 row)

SELECT * FROM sift.table_find_stale_versions('sv_heap');
 duplicate_keys | sample_keys 
----------------+-------------
              0 | 
(1 row)

SELECT * FROM sift.table_find_stale_versions('sv_comp');
 duplicate_keys | sample_keys 
----------------+-------------
              0 | 
(1 row)

SELECT * FROM sift.table_find_stale_versions('sv_part');
 duplicate_keys | sample_keys 
----------------+-------------
              0 | 
(1 row)

-- copies, in the transaction that made them; the aggregate engine and the index see none
BEGIN;
INSERT INTO sv_sift SELECT g, 1 FROM generate_series(1, 25) g;
INSERT INTO sv_heap SELECT g, 1 FROM generate_series(1, 25) g;
INSERT INTO sv_comp SELECT g, 'k' || g % 7, 1 FROM generate_series(1, 25) g;
INSERT INTO sv_part SELECT g, 1 FROM generate_series(990, 1010) g;
SELECT * FROM sift.table_find_stale_versions('sv_sift');
 duplicate_keys |                    sample_keys                     
----------------+----------------------------------------------------
             25 | {(1),(10),(11),(12),(13),(14),(15),(16),(17),(18)}
(1 row)

SELECT * FROM sift.table_find_stale_versions('sv_sift', 3);
 duplicate_keys |   sample_keys   
----------------+-----------------
             25 | {(1),(10),(11)}
(1 row)

SELECT * FROM sift.table_find_stale_versions('sv_heap');
 duplicate_keys |                    sample_keys                     
----------------+----------------------------------------------------
             25 | {(1),(10),(11),(12),(13),(14),(15),(16),(17),(18)}
(1 row)

SELECT * FROM sift.table_find_stale_versions('sv_comp', 4);
 duplicate_keys |               sample_keys                
----------------+------------------------------------------
             25 | {"(1,k1)","(10,k3)","(11,k4)","(12,k5)"}
(1 row)

SELECT * FROM sift.table_find_stale_versions('sv_part', 2);
 duplicate_keys |   sample_keys   
----------------+-----------------
             21 | {(1000),(1001)}
(1 row)

SELECT count(*) AS rows_seen, count(DISTINCT id) AS keys_seen FROM sv_sift;
 rows_seen | keys_seen 
-----------+-----------
      3025 |      3025
(1 row)

SELECT current_setting('sift.enable_agg') AS enable_agg_after, current_setting('enable_indexscan') AS indexscan_after;
 enable_agg_after | indexscan_after 
------------------+-----------------
 on               | on
(1 row)

ROLLBACK;
-- a repeatable read snapshot taken before the copies: it finds the copies of its own transaction
BEGIN ISOLATION LEVEL REPEATABLE READ;
SELECT count(*) FROM sv_sift;
 count 
-------
  3000
(1 row)

INSERT INTO sv_sift SELECT g, 2 FROM generate_series(1, 5) g;
SELECT * FROM sift.table_find_stale_versions('sv_sift');
 duplicate_keys |      sample_keys      
----------------+-----------------------
              5 | {(1),(2),(3),(4),(5)}
(1 row)

ROLLBACK;
-- a table without a primary key
\set VERBOSITY terse
SELECT * FROM sift.table_find_stale_versions('sv_nopk');
ERROR:  table sv_nopk has no primary key
DROP TABLE sv_sift, sv_heap, sv_comp, sv_part, sv_nopk;
DROP EXTENSION pg_sift;