-- count(*) of a range of an integer, date or timestamp leaf with a global order: a segment whose units are all exact (every page
-- all-visible, no dead row) is counted from the entries between the bounds of the range, with no unit read ("Rows Scanned: 0"); the others are read. The counts must be the heap's
-- through DML, VACUUM, the write buffer, promotion and a merge, for every type, and a second test, another aggregate or an OR must not be taken for a range.
CREATE EXTENSION pg_sift_test CASCADE;
NOTICE:  installing required extension "pg_sift"
SET client_min_messages = warning;
SET sift.async_maintenance = off;
SET max_parallel_maintenance_workers = 0;
CREATE TABLE rc (id int, a int2, b int4, c int8, d date, e timestamp, f timestamptz, pad text DEFAULT repeat('x', 200));
ALTER TABLE rc ALTER COLUMN pad SET STORAGE PLAIN;
INSERT INTO rc (id, a, b, c, d, e, f)
SELECT i, CASE WHEN i % 41 = 0 THEN NULL ELSE (i * 37) % 3001 - 1500 END, CASE WHEN i % 53 = 0 THEN NULL ELSE (i * 7919) % 100003 END, (i * 104729::bigint) % 1000003 - 500000,
	   date '2020-01-01' + (i * 13) % 2000, timestamp '2020-01-01' + ((i * 7919) % 34560000) * interval '1 second', timestamptz '2020-01-01 00:00:00+00' + ((i * 104729::bigint) % 34560000) * interval '1 second'
FROM generate_series(1, 60000) i;
CREATE INDEX rc_i ON rc USING sift_col (a sift.col_int2_ops, b sift.col_int4_ops, c sift.col_int8_ops, d sift.col_date_ops, e sift.col_timestamp_ops, f sift.col_timestamptz_ops, id sift.col_int4_ops)
	WITH (write_buffer_limit = 256, order_columns = 'a, b, c, d, e, f', order_array_columns = 'a, b, c, d, e, f');
VACUUM (ANALYZE) rc;
-- the rows a statement read: 0 when the segments counted
CREATE FUNCTION rc_scanned(q text) RETURNS int LANGUAGE plpgsql SET sift.enable_agg = on SET max_parallel_workers_per_gather = 0 SET enable_indexscan = off SET enable_indexonlyscan = off SET enable_bitmapscan = off SET enable_seqscan = off AS $$
DECLARE l text;
BEGIN
	FOR l IN EXECUTE 'EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF) ' || q LOOP
		IF l ~ 'Rows: (\d+) scanned' THEN RETURN substring(l from 'Rows: (\d+) scanned')::int; END IF;
	END LOOP;
	RETURN -1;
END $$;
CREATE FUNCTION rc_agg(w text) RETURNS bigint LANGUAGE plpgsql SET sift.enable_agg = on SET max_parallel_workers_per_gather = 0 SET enable_indexscan = off SET enable_indexonlyscan = off SET enable_bitmapscan = off SET enable_seqscan = off AS $$
DECLARE r bigint;
BEGIN EXECUTE 'SELECT count(*) FROM rc WHERE ' || w INTO r; RETURN r; END $$;
CREATE FUNCTION rc_heap(w text) RETURNS bigint LANGUAGE plpgsql SET sift.enable_agg = off SET sift.enable_col_scan = off SET enable_indexscan = off SET enable_bitmapscan = off SET enable_indexonlyscan = off AS $$
DECLARE r bigint;
BEGIN EXECUTE 'SELECT count(*) FROM rc WHERE ' || w INTO r; RETURN r; END $$;
CREATE TABLE rc_w (n int, w text);
INSERT INTO rc_w VALUES
 (1, 'b BETWEEN 1000 AND 1200'), (2, 'b >= 99000'), (3, 'b < 150'), (4, 'b > 500 AND b <= 5000'), (5, 'b = 4321'), (6, 'b BETWEEN 90 AND 10'),
 (7, 'b BETWEEN 0 AND 100002'), (8, 'b > 1000000'), (9, 'a BETWEEN -50 AND 50'), (10, 'a >= 1499'), (11, 'c BETWEEN -1000 AND 1000'), (12, 'c < -499900'),
 (13, 'd BETWEEN date ''2021-03-01'' AND date ''2021-03-15'''), (14, 'd = date ''2020-05-05'''), (15, 'e >= timestamp ''2020-12-30'' AND e < timestamp ''2021-01-02'''),
 (16, 'f BETWEEN timestamptz ''2020-06-01 00:00:00+00'' AND timestamptz ''2020-06-02 00:00:00+00'''), (17, 'e < timestamp ''2019-01-01'''),
 (18, 'b BETWEEN 1000 AND 1200 AND b <> 1100'), (19, 'b BETWEEN 1000 AND 1200 AND a > 0'), (20, 'b < 100 OR b > 99900'), (21, 'b IS NOT NULL AND b < 100');
SELECT n, left(w, 60) AS w, rc_agg(w) = rc_heap(w) AS agree, rc_heap(w) AS rows, rc_scanned('SELECT count(*) FROM rc WHERE ' || w) = 0 AS counted FROM rc_w ORDER BY n;
 n  |                              w                               | agree | rows  | counted 
----+--------------------------------------------------------------+-------+-------+---------
  1 | b BETWEEN 1000 AND 1200                                      | t     |   117 | t
  2 | b >= 99000                                                   | t     |   592 | t
  3 | b < 150                                                      | t     |    88 | t
  4 | b > 500 AND b <= 5000                                        | t     |  2649 | t
  5 | b = 4321                                                     | t     |     1 | t
  6 | b BETWEEN 90 AND 10                                          | t     |     0 | t
  7 | b BETWEEN 0 AND 100002                                       | t     | 58868 | t
  8 | b > 1000000                                                  | t     |     0 | t
  9 | a BETWEEN -50 AND 50                                         | t     |  1970 | t
 10 | a >= 1499                                                    | t     |    40 | t
 11 | c BETWEEN -1000 AND 1000                                     | t     |   120 | t
 12 | c < -499900                                                  | t     |     8 | t
 13 | d BETWEEN date '2021-03-01' AND date '2021-03-15'            | t     |   450 | t
 14 | d = date '2020-05-05'                                        | t     |    30 | t
 15 | e >= timestamp '2020-12-30' AND e < timestamp '2021-01-02'   | t     |   427 | t
 16 | f BETWEEN timestamptz '2020-06-01 00:00:00+00' AND timestamp | t     |   149 | t
 17 | e < timestamp '2019-01-01'                                   | t     |     0 | t
 18 | b BETWEEN 1000 AND 1200 AND b <> 1100                        | t     |   116 | f
 19 | b BETWEEN 1000 AND 1200 AND a > 0                            | t     |    57 | f
 20 | b < 100 OR b > 99900                                         | t     |   119 | f
 21 | b IS NOT NULL AND b < 100                                    | t     |    58 | f
(21 rows)

-- another aggregate, a group, a second test: the units are read
SELECT rc_scanned('SELECT count(*), sum(id) FROM rc WHERE b BETWEEN 1000 AND 1200') > 0 AS with_a_sum_scans,
       rc_scanned('SELECT b % 2, count(*) FROM rc WHERE b BETWEEN 1000 AND 1200 GROUP BY 1') > 0 AS with_a_group_scans,
       rc_scanned('SELECT count(*) FROM rc WHERE b BETWEEN 1000 AND 1200 AND id > 5') > 0 AS with_another_test_scans;
 with_a_sum_scans | with_a_group_scans | with_another_test_scans 
------------------+--------------------+-------------------------
 t                | t                  | t
(1 row)

-- rows in the write buffer, dead rows, rows not yet visible to all: counts stay the heap's, and the segments that cannot be counted are read
INSERT INTO rc (id, a, b, c, d, e, f) SELECT 60000 + i, i % 100, 1000 + i % 300, i, date '2021-03-01' + i % 20, timestamp '2021-01-01', timestamptz '2021-01-01+00' FROM generate_series(1, 150) i;
SELECT n, rc_agg(w) = rc_heap(w) AS agree FROM rc_w WHERE n IN (1, 2, 4, 9, 13) ORDER BY n;
 n  | agree 
----+-------
  1 | t
  2 | t
  4 | t
  9 | t
 13 | t
(5 rows)

SELECT sift.col_promote('rc_i') >= 0 AS promoted;
 promoted 
----------
 t
(1 row)

SELECT n, rc_agg(w) = rc_heap(w) AS agree FROM rc_w WHERE n IN (1, 2, 4, 9, 13) ORDER BY n;
 n  | agree 
----+-------
  1 | t
  2 | t
  4 | t
  9 | t
 13 | t
(5 rows)

VACUUM rc;
SELECT rc_scanned('SELECT count(*) FROM rc WHERE b BETWEEN 1000 AND 1200') = 0 AS counted_after_vacuum;
 counted_after_vacuum 
----------------------
 t
(1 row)

DELETE FROM rc WHERE id % 1000 = 7;
SELECT n, rc_agg(w) = rc_heap(w) AS agree FROM rc_w ORDER BY n;
 n  | agree 
----+-------
  1 | t
  2 | t
  3 | t
  4 | t
  5 | t
  6 | t
  7 | t
  8 | t
  9 | t
 10 | t
 11 | t
 12 | t
 13 | t
 14 | t
 15 | t
 16 | t
 17 | t
 18 | t
 19 | t
 20 | t
 21 | t
(21 rows)

SELECT rc_scanned('SELECT count(*) FROM rc WHERE b BETWEEN 1000 AND 1200') > 0 AS dead_rows_are_read;
 dead_rows_are_read 
--------------------
 t
(1 row)

VACUUM rc;
UPDATE rc SET b = b + 1 WHERE id % 97 = 0;
SELECT n, rc_agg(w) = rc_heap(w) AS agree FROM rc_w ORDER BY n;
 n  | agree 
----+-------
  1 | t
  2 | t
  3 | t
  4 | t
  5 | t
  6 | t
  7 | t
  8 | t
  9 | t
 10 | t
 11 | t
 12 | t
 13 | t
 14 | t
 15 | t
 16 | t
 17 | t
 18 | t
 19 | t
 20 | t
 21 | t
(21 rows)

VACUUM rc;
SELECT sift.col_merge('rc_i', true) >= 0 AS merged;
 merged 
--------
 t
(1 row)

SELECT n, rc_agg(w) = rc_heap(w) AS agree FROM rc_w ORDER BY n;
 n  | agree 
----+-------
  1 | t
  2 | t
  3 | t
  4 | t
  5 | t
  6 | t
  7 | t
  8 | t
  9 | t
 10 | t
 11 | t
 12 | t
 13 | t
 14 | t
 15 | t
 16 | t
 17 | t
 18 | t
 19 | t
 20 | t
 21 | t
(21 rows)

SELECT rc_scanned('SELECT count(*) FROM rc WHERE b BETWEEN 1000 AND 1200') = 0 AS counted_after_merge;
 counted_after_merge 
---------------------
 t
(1 row)

SELECT sift.col_verify('rc_i', true);
 col_verify 
------------
 
(1 row)

-- a segment without the array is read
ALTER INDEX rc_i RESET (order_array_columns);
REINDEX INDEX rc_i;
SELECT rc_scanned('SELECT count(*) FROM rc WHERE b BETWEEN 1000 AND 1200') > 0 AS no_array_is_read, rc_agg('b BETWEEN 1000 AND 1200') = rc_heap('b BETWEEN 1000 AND 1200') AS agree;
 no_array_is_read | agree 
------------------+-------
 t                | t
(1 row)

DROP FUNCTION rc_scanned(text), rc_agg(text), rc_heap(text);
DROP TABLE rc_w, rc;
DROP EXTENSION pg_sift_test, pg_sift CASCADE;