-- count(*) of one element test that the point directory answers exactly: a segment whose units are all clean and all-visible is counted from the list
-- ("counted": no row of the segment is scanned); every unit that is not (a dead row, a page not all-visible, an element too common to be listed, rows of the
-- write buffer) is read, and the count is the heap's after every kind of write. The output names how the count was answered, not the number of units, which follows the
-- size of the compressed units of the runner; after an INSERT it shows the count only: whether the last page of the segment took rows, and so is still
-- all-visible, depends on the layout of the heap.
CREATE EXTENSION pg_sift;
SET max_parallel_workers_per_gather = 0;
SET seq_page_cost = 1000;
SET random_page_cost = 1000;
SET client_min_messages = warning;
SET sift.async_maintenance = off;
SET max_parallel_maintenance_workers = 0;
CREATE FUNCTION gen(a int, b int) RETURNS TABLE(id int, tags text[], ids int[]) LANGUAGE sql AS $$
	SELECT i, CASE WHEN i % 13 = 0 THEN NULL WHEN i % 17 = 0 THEN '{}' WHEN i % 19 = 0 THEN ARRAY['t1', NULL, 'u1', 't1']
	               ELSE ARRAY['t' || i % 500, 'u' || i % 3, 'w' || i % 1000] END,
	       CASE WHEN i % 11 = 0 THEN NULL WHEN i % 23 = 0 THEN '{}' ELSE ARRAY[i % 400, i % 5 + 1000, i % 7 + 2000] END
	FROM generate_series(a, b) i $$;
CREATE TABLE ac (id int, tags text[], ids int[]) WITH (autovacuum_enabled = false);
INSERT INTO ac SELECT * FROM gen(1, 30000);
CREATE INDEX ac_col ON ac USING sift_col (id, tags, ids) WITH (paths = 'tags[] text, ids[] int4');
VACUUM (FREEZE, ANALYZE) ac;
-- the count with the pushdown against the heap's, and how it was answered
CREATE FUNCTION cnt(w text, show boolean) RETURNS text LANGUAGE plpgsql AS $$
DECLARE l text; listed boolean := false; scanned bigint := 0; a bigint; b bigint;
BEGIN
	SET sift.enable_agg = on;
	FOR l IN EXECUTE 'EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF, BUFFERS OFF) SELECT count(*) FROM ac WHERE ' || w LOOP
		IF l ~ '^ *Units:.*from point directory' THEN listed := true; END IF;
		IF l ~ '^ *Rows:' THEN scanned := (regexp_match(l, 'Rows: (\d+) scanned'))[1]::bigint; END IF;
	END LOOP;
	EXECUTE 'SELECT count(*) FROM ac WHERE ' || w INTO a;
	SET sift.enable_agg = off;
	EXECUTE 'SELECT count(*) FROM ac WHERE ' || w INTO b;
	RESET sift.enable_agg;
	RETURN (CASE WHEN a = b THEN 'same ' ELSE 'DIFF ' END) || b ||
		CASE WHEN NOT show THEN '' ELSE (CASE WHEN scanned < 5000 THEN ' counted' ELSE ' read' END) || CASE WHEN listed THEN ' listed' ELSE '' END END;
END $$;
CREATE FUNCTION shapes(tag text, show boolean DEFAULT true) RETURNS TABLE(shape text, result text) LANGUAGE sql AS $$
	SELECT tag || ' ' || s.name, cnt(s.w, show) FROM (VALUES
		('text', $w$tags @> ARRAY['t7']$w$),
		('text any', $w$'t8' = ANY (tags)$w$),
		('repeated', $w$tags @> ARRAY['t1']$w$),
		('int', $w$ids @> ARRAY[7]$w$),
		('absent', $w$tags @> ARRAY['zz']$w$),
		('too common', $w$tags @> ARRAY['u1']$w$),
		('two', $w$tags @> ARRAY['t7', 'w7']$w$),
		('overlap', $w$tags && ARRAY['t7', 't8']$w$),
		('not', $w$NOT (tags @> ARRAY['t7'])$w$),
		('with filter', $w$tags @> ARRAY['t7'] AND id > 100$w$)) s(name, w) $$;
SELECT * FROM shapes('built');
       shape       |       result        
-------------------+---------------------
 built text        | same 49 counted
 built text any    | same 49 counted
 built repeated    | same 1422 counted
 built int         | same 66 counted
 built absent      | same 0 counted
 built too common  | same 9604 read
 built two         | same 25 read listed
 built overlap     | same 98 read listed
 built not         | same 27644 counted
 built with filter | same 48 read listed
(10 rows)

-- rows of the write buffer: the segment is counted from the list, the buffer is read
INSERT INTO ac SELECT * FROM gen(30001, 31000);
SELECT * FROM shapes('insert', false);
       shape        |   result   
--------------------+------------
 insert text        | same 51
 insert text any    | same 51
 insert repeated    | same 1469
 insert int         | same 69
 insert absent      | same 0
 insert too common  | same 9924
 insert two         | same 26
 insert overlap     | same 102
 insert not         | same 28565
 insert with filter | same 50
(10 rows)

VACUUM ac;
SELECT * FROM shapes('vacuum');
       shape        |        result        
--------------------+----------------------
 vacuum text        | same 51 counted
 vacuum text any    | same 51 counted
 vacuum repeated    | same 1469 counted
 vacuum int         | same 69 counted
 vacuum absent      | same 0 counted
 vacuum too common  | same 9924 read
 vacuum two         | same 26 read listed
 vacuum overlap     | same 102 read listed
 vacuum not         | same 28565 counted
 vacuum with filter | same 50 read listed
(10 rows)

-- a deleted row is dead in the index until it is vacuumed and rewritten, and its page is not all-visible
DELETE FROM ac WHERE id % 1000 = 7;
SELECT * FROM shapes('delete');
       shape        |        result         
--------------------+-----------------------
 delete text        | same 25 read listed
 delete text any    | same 51 read listed
 delete repeated    | same 1468 read listed
 delete int         | same 55 read listed
 delete absent      | same 0 counted
 delete too common  | same 9912 read
 delete two         | same 0 read listed
 delete overlap     | same 76 read listed
 delete not         | same 28562 counted
 delete with filter | same 25 read listed
(10 rows)

VACUUM ac;
SELECT * FROM shapes('delete vacuum');
           shape           |        result         
---------------------------+-----------------------
 delete vacuum text        | same 25 read listed
 delete vacuum text any    | same 51 read listed
 delete vacuum repeated    | same 1468 read listed
 delete vacuum int         | same 55 read listed
 delete vacuum absent      | same 0 counted
 delete vacuum too common  | same 9912 read
 delete vacuum two         | same 0 read listed
 delete vacuum overlap     | same 76 read listed
 delete vacuum not         | same 28562 counted
 delete vacuum with filter | same 25 read listed
(10 rows)

-- an updated row: the old version is dead, the new one is a row of the buffer
UPDATE ac SET tags = ARRAY['t7', 'x1'] WHERE id % 701 = 0;
SELECT * FROM shapes('update');
       shape        |        result         
--------------------+-----------------------
 update text        | same 69 read listed
 update text any    | same 51 read listed
 update repeated    | same 1466 read listed
 update int         | same 55 read listed
 update absent      | same 0 counted
 update too common  | same 9898 read
 update two         | same 0 read listed
 update overlap     | same 120 read listed
 update not         | same 28521 counted
 update with filter | same 69 read listed
(10 rows)

VACUUM (FREEZE) ac;
SELECT * FROM shapes('update vacuum');
           shape           |        result         
---------------------------+-----------------------
 update vacuum text        | same 69 read listed
 update vacuum text any    | same 51 read listed
 update vacuum repeated    | same 1466 read listed
 update vacuum int         | same 55 read listed
 update vacuum absent      | same 0 counted
 update vacuum too common  | same 9898 read
 update vacuum two         | same 0 read listed
 update vacuum overlap     | same 120 read listed
 update vacuum not         | same 28521 counted
 update vacuum with filter | same 69 read listed
(10 rows)

SELECT sift.col_merge('ac_col', true) > 0 AS merged;
 merged 
--------
 t
(1 row)

SELECT * FROM shapes('merge');
       shape       |        result        
-------------------+----------------------
 merge text        | same 69 counted
 merge text any    | same 51 counted
 merge repeated    | same 1466 counted
 merge int         | same 55 counted
 merge absent      | same 0 counted
 merge too common  | same 9898 read
 merge two         | same 0 counted
 merge overlap     | same 120 read listed
 merge not         | same 28521 counted
 merge with filter | same 69 read listed
(10 rows)

REINDEX INDEX ac_col;
VACUUM (FREEZE) ac;
SELECT * FROM shapes('reindex');
        shape        |        result        
---------------------+----------------------
 reindex text        | same 69 counted
 reindex text any    | same 51 counted
 reindex repeated    | same 1466 counted
 reindex int         | same 55 counted
 reindex absent      | same 0 counted
 reindex too common  | same 9898 read
 reindex two         | same 0 counted
 reindex overlap     | same 120 read listed
 reindex not         | same 28521 counted
 reindex with filter | same 69 read listed
(10 rows)

-- a page that is no longer all-visible while the index is clean: a row of the segment updated in a way that changes no element
UPDATE ac SET id = id WHERE id = 1000;
SELECT * FROM shapes('page');
      shape       |        result         
------------------+-----------------------
 page text        | same 69 read listed
 page text any    | same 51 read listed
 page repeated    | same 1466 read listed
 page int         | same 55 read listed
 page absent      | same 0 counted
 page too common  | same 9898 read
 page two         | same 0 counted
 page overlap     | same 120 read listed
 page not         | same 28521 counted
 page with filter | same 69 read listed
(10 rows)

DROP TABLE ac;
DROP FUNCTION gen(int, int), cnt(text, boolean), shapes(text, boolean);
DROP EXTENSION pg_sift;