-- ORDER BY key, id LIMIT n OFFSET m (the key breaks ties by a second key) with an OFFSET of thousands: the global order takes the front of the OFFSET off its sources in one cut (a bucket
-- of a number, a string of a text key) and the walk queues the rest, whole groups of ties and the group that holds the first row to keep. The rows are the plan's without the extension,
-- and those of the walk of every entry (test_col_gorder_noskip), through ties, NULLs, ASC and DESC, NULLS FIRST and LAST, deleted, updated and vacuumed rows, new rows in the write
-- buffer, and a second pass of the same scan (a rewind).
CREATE EXTENSION pg_sift_test CASCADE;
NOTICE:  installing required extension "pg_sift"
CREATE EXTENSION pg_sift_helpers;
SET sift.test_col_legacy_options = on;
SET sift.async_maintenance = off;
SET max_parallel_maintenance_workers = 0;
SET max_parallel_workers_per_gather = 0;
\set VERBOSITY terse
CREATE TABLE ob (id int PRIMARY KEY, n int, t text, dt date, ts timestamp, pad text DEFAULT repeat('x', 100));
ALTER TABLE ob ALTER COLUMN pad SET STORAGE PLAIN;
INSERT INTO ob (id, n, t, dt, ts)
SELECT i, CASE WHEN i % 41 = 0 THEN NULL ELSE (i * 7919) % 700 END, CASE WHEN i % 43 = 0 THEN NULL ELSE 't' || lpad(((i::bigint * 7919) % 1500)::text, 5, '0') END,
	   CASE WHEN i % 37 = 0 THEN NULL ELSE DATE '2020-01-01' + (i * 13) % 400 END, CASE WHEN i % 29 = 0 THEN NULL ELSE TIMESTAMP '2020-01-01' + ((i * 7919) % 100000) * interval '1 second' END
FROM generate_series(1, 20000) i;
CREATE INDEX ob_i ON ob USING sift_col (id sift.col_int4_ops, n sift.col_int4_ops, t sift.col_text_ops, dt sift.col_date_ops, ts sift.col_timestamp_ops) WITH (order_columns = 'id, n, dt, ts', order_array_columns = 'id, n, t');
NOTICE:  sift_col index "ob_i": 2 automatic choices (order_array_columns, track_changes)
NOTICE:  sift_col index "ob_i" is column format minor 44, which a library before 4.3 cannot read
VACUUM (FREEZE, ANALYZE, DISABLE_PAGE_SKIPPING) ob;
SET sift.test_topn_heads = off;
-- the answer says DIFFERS with the query when the rows of the scan are not those of the plan without the extension or of the walk of every entry; 'bulk' when the scan used the global order
CREATE FUNCTION ob_chk(q text) RETURNS text LANGUAGE plpgsql AS $$
DECLARE pl text; used bool := false; ka text; kb text; kc text;
	rows text := 'SELECT string_agg(id || '':'' || coalesce(k::text, ''N''), '','' ORDER BY ord) FROM (SELECT row_number() OVER () AS ord, * FROM (%s) r) x';
BEGIN
	SET LOCAL sift.enable_agg = on; SET LOCAL sift.enable_col_scan = on; SET LOCAL enable_indexscan = off; SET LOCAL enable_bitmapscan = off; SET LOCAL enable_indexonlyscan = off;
	SET LOCAL sift.agg_force = on; SET LOCAL enable_seqscan = off; SET LOCAL enable_sort = off;
	FOR pl IN EXECUTE 'EXPLAIN (ANALYZE, COSTS off, TIMING off, SUMMARY off) ' || q LOOP
		used := used OR pl LIKE '%Global Order%';
	END LOOP;
	EXECUTE format(rows, q) INTO ka;
	SET LOCAL sift.test_col_gorder_noskip = on;
	EXECUTE format(rows, q) INTO kc;
	SET LOCAL sift.test_col_gorder_noskip = off;
	SET LOCAL sift.enable_agg = off; SET LOCAL enable_seqscan = on; SET LOCAL enable_sort = on;
	EXECUTE format(rows, q) INTO kb;
	RETURN CASE WHEN ka IS DISTINCT FROM kb OR kc IS DISTINCT FROM kb THEN 'DIFFERS ' || q WHEN NOT used THEN 'units' ELSE 'bulk' END;
END $$;
-- the cut is taken: the entries it took off the sources (sift._test_col_gdead_stats) are most of the OFFSET, and none with the walk of every entry
CREATE FUNCTION ob_cut(q text, noskip bool DEFAULT false) RETURNS int8 LANGUAGE plpgsql AS $$
DECLARE c0 int8; c1 int8; r int8;
BEGIN
	SET LOCAL sift.enable_agg = on; SET LOCAL sift.enable_col_scan = on; SET LOCAL sift.agg_force = on; SET LOCAL enable_seqscan = off; SET LOCAL enable_sort = off;
	EXECUTE format('SET LOCAL sift.test_col_gorder_noskip = %s', noskip);
	SELECT cut INTO c0 FROM sift._test_col_gdead_stats();
	EXECUTE format('SELECT count(*) FROM (%s) x', q) INTO r;
	SELECT cut INTO c1 FROM sift._test_col_gdead_stats();
	RETURN c1 - c0;
END $$;
SELECT c, ob_cut(format('SELECT id, %s FROM ob ORDER BY %s, id LIMIT 20 OFFSET 7000', c, c)) BETWEEN 5200 AND 7000 AS cut, ob_cut(format('SELECT id, %s FROM ob ORDER BY %s, id LIMIT 20 OFFSET 7000', c, c), true) AS walked FROM unnest(ARRAY['n', 't']) c;
 c | cut | walked 
---+-----+--------
 n | t   |      0
 t | t   |      0
(2 rows)

-- the same scan read twice by a join (the inner side is rewound, not materialized)
CREATE FUNCTION ob_twice(q text) RETURNS text LANGUAGE sql AS $$ SELECT sift_helpers.topn_offset_rows_twice_same(q) $$;
CREATE FUNCTION ob_all() RETURNS TABLE (result text, n bigint) LANGUAGE sql AS
$$ SELECT ob_chk(format('SELECT id, %s AS k FROM ob ORDER BY %s %s NULLS %s, %s, id LIMIT 20 OFFSET %s', c, c, d, nl, k2, o)) AS result, count(*)
   FROM unnest(ARRAY['n', 't']) c, unnest(ARRAY['ASC', 'DESC']) d, unnest(ARRAY['FIRST', 'LAST']) nl, unnest(ARRAY['id', 'id DESC', 'dt', 'dt DESC NULLS LAST', 'ts DESC', 'ts NULLS FIRST']) k2,
	    unnest(ARRAY[2100, 7000]) o GROUP BY 1 ORDER BY 1 $$;
-- a condition on a column other than the key (IS NULL, IS NOT NULL, integer and date comparisons, AND and OR) is decided from the units for the entries that are exact, the heap says for the others
CREATE FUNCTION ob_filt() RETURNS TABLE (result text, n bigint) LANGUAGE sql AS
$$ SELECT ob_chk(format('SELECT id, %s AS k FROM ob WHERE %s ORDER BY %s %s, id LIMIT 20 OFFSET %s', c, w, c, d, o)) AS result, count(*)
   FROM unnest(ARRAY['n']) c, unnest(ARRAY['ASC', 'DESC']) d,
	    unnest(ARRAY['ts IS NOT NULL', 'ts IS NULL OR dt > DATE ''2020-02-01''', 'dt < DATE ''2020-11-01''', 'n >= 100 AND ts IS NOT NULL', 'n < 400 OR ts > TIMESTAMP ''2020-01-01 10:00''', 'dt >= DATE ''2020-03-01'' AND ts IS NOT NULL']) w,
	    unnest(ARRAY[2100, 4000]) o GROUP BY 1 ORDER BY 1 $$;
SELECT * FROM ob_all();
 result | n  
--------+----
 bulk   | 96
(1 row)

SELECT * FROM ob_filt();
 result | n  
--------+----
 bulk   | 24
(1 row)

SELECT ob_twice('SELECT id, t AS k FROM ob ORDER BY t, id LIMIT 20 OFFSET 7000') AS t_rewind, ob_twice('SELECT id, n AS k FROM ob ORDER BY n, id LIMIT 20 OFFSET 7000') AS n_rewind;
 t_rewind | n_rewind 
----------+----------
 same     | same
(1 row)

-- deleted and updated rows, a VACUUM, new rows in the write buffer
DELETE FROM ob WHERE id % 53 = 0;
UPDATE ob SET n = n + 1, t = 'u' || t WHERE id % 97 = 0;
SELECT * FROM ob_all();
 result | n  
--------+----
 bulk   | 96
(1 row)

VACUUM ob;
SELECT * FROM ob_all();
 result | n  
--------+----
 bulk   | 96
(1 row)

SELECT * FROM ob_filt();
 result | n  
--------+----
 bulk   | 24
(1 row)

-- the dead rows of a text key leave the cut short of the OFFSET in rows: it is taken further, past them (each costs a string of the dictionary to read in the walk)
SELECT ob_cut('SELECT id, t FROM ob ORDER BY t, id LIMIT 20 OFFSET 7000') > 7000 AS text_cut_past_dead_rows;
 text_cut_past_dead_rows 
-------------------------
 t
(1 row)

INSERT INTO ob (id, n, t) SELECT 200000 + i, (i * 31) % 700, 't' || lpad(((i * 17) % 1500)::text, 5, '0') FROM generate_series(1, 300) i;
SELECT * FROM ob_all();
 result | n  
--------+----
 bulk   | 96
(1 row)

SELECT * FROM ob_filt();
 result | n  
--------+----
 bulk   | 24
(1 row)

SELECT ob_twice('SELECT id, t AS k FROM ob ORDER BY t, id LIMIT 20 OFFSET 7000') AS t_rewind, ob_twice('SELECT id, n AS k FROM ob ORDER BY n, id LIMIT 20 OFFSET 7000') AS n_rewind;
 t_rewind | n_rewind 
----------+----------
 same     | same
(1 row)

-- the group of ties that holds the first row to keep has the keys after the first read from the units, not fetched, when its entries are exact (10 values of g, 2,000 rows each); when the heap must confirm them,
-- it is fetched once, and the rows of it that the Limit throws away are not fetched again to go out
CREATE TABLE og (id int PRIMARY KEY, g int, w int, pad text DEFAULT repeat('x', 100));
ALTER TABLE og ALTER COLUMN pad SET STORAGE PLAIN;
INSERT INTO og (id, g, w) SELECT i, (i * 7919) % 10, CASE WHEN i % 50 = 0 THEN NULL ELSE i % 1000 END FROM generate_series(1, 20000) i;
CREATE INDEX og_i ON og USING sift_col (id sift.col_int4_ops, g sift.col_int4_ops, w sift.col_int4_ops) WITH (order_columns = 'id, g, w', order_array_columns = 'id, g');
NOTICE:  sift_col index "og_i": 2 automatic choices (order_array_columns, track_changes)
NOTICE:  sift_col index "og_i" is column format minor 44, which a library before 4.3 cannot read
VACUUM (FREEZE, ANALYZE, DISABLE_PAGE_SKIPPING) og;
CREATE FUNCTION og_fetches(q text) RETURNS int LANGUAGE plpgsql AS $$
DECLARE pl text; n int := -1; gord bool := false;
BEGIN
	SET LOCAL sift.enable_agg = on; SET LOCAL sift.enable_col_scan = on; SET LOCAL sift.agg_force = on; SET LOCAL enable_seqscan = off; SET LOCAL enable_sort = off; SET LOCAL enable_indexscan = off; SET LOCAL enable_bitmapscan = off;
	FOR pl IN EXECUTE 'EXPLAIN (ANALYZE, COSTS off, TIMING off, SUMMARY off) ' || q LOOP
		gord := gord OR pl LIKE '%Global Order%';
		IF pl LIKE '%heap fetches%' THEN n := substring(pl from '([0-9]+) heap fetches')::int; END IF;
	END LOOP;
	RETURN CASE WHEN gord THEN n ELSE -1 END;
END $$;
SELECT o, og_fetches(format('SELECT id, g FROM og ORDER BY g, id LIMIT 20 OFFSET %s', o)) BETWEEN 0 AND 100 AS read_not_fetched FROM unnest(ARRAY[5000, 9000, 15000]) o;
   o   | read_not_fetched 
-------+------------------
  5000 | t
  9000 | t
 15000 | t
(3 rows)

-- the same under a condition on another column: decided from the units, not by the heap rows of the entries before the OFFSET
SELECT o, og_fetches(format('SELECT id, g FROM og WHERE %s ORDER BY g, id LIMIT 20 OFFSET %s', w, o)) BETWEEN 0 AND 100 AS cond_read FROM unnest(ARRAY['w IS NOT NULL', 'w < 900', 'w >= 0 AND w IS NOT NULL']) w, unnest(ARRAY[3000, 5000]) o;
  o   | cond_read 
------+-----------
 3000 | t
 3000 | t
 3000 | t
 5000 | t
 5000 | t
 5000 | t
(6 rows)

-- the pages of a table that a third of the rows were updated in are not all-visible: every entry is confirmed by the heap once (the cut's, the group's) and the 20 rows go out
UPDATE og SET pad = repeat('y', 100) WHERE id % 3 = 0;
SELECT o, og_fetches(format('SELECT id, g FROM og ORDER BY g, id LIMIT 20 OFFSET %s', o)) = (o / 2000) * 2000 + 2000 + 20 AS one_fetch FROM unnest(ARRAY[5000, 9000, 15000]) o;
   o   | one_fetch 
-------+-----------
  5000 | t
  9000 | t
 15000 | t
(3 rows)

DROP FUNCTION og_fetches(text);
DROP TABLE og;
-- random strings: the dictionary of a text key is in compressed blocks, a string read decompresses one; the bisections of the cut read a block again within a few reads, which is kept (168 blocks decompressed without, 110 with; they are
-- sift._test_col_gdead_stats().blocks)
CREATE TABLE orr (id int PRIMARY KEY, r text, pad text DEFAULT repeat('x', 100));
ALTER TABLE orr ALTER COLUMN pad SET STORAGE PLAIN;
INSERT INTO orr (id, r) SELECT i, left(md5(i::text), 10) FROM generate_series(1, 20000) i;
CREATE INDEX orr_i ON orr USING sift_col (id sift.col_int4_ops, r sift.col_text_ops) WITH (order_columns = 'id', order_array_columns = 'id, r');
NOTICE:  sift_col index "orr_i": 2 automatic choices (order_array_columns, track_changes)
NOTICE:  sift_col index "orr_i" is column format minor 44, which a library before 4.3 cannot read
VACUUM (FREEZE, ANALYZE, DISABLE_PAGE_SKIPPING) orr;
DELETE FROM orr WHERE id % 53 = 0;
UPDATE orr SET r = 'u' || r WHERE id % 97 = 0;
VACUUM orr;
CREATE FUNCTION orr_blocks(q text) RETURNS int8 LANGUAGE plpgsql AS $$
DECLARE c0 int8; c1 int8; r int8;
BEGIN
	SET LOCAL sift.enable_agg = on; SET LOCAL sift.enable_col_scan = on; SET LOCAL sift.agg_force = on; SET LOCAL enable_seqscan = off; SET LOCAL enable_sort = off;
	SELECT blocks INTO c0 FROM sift._test_col_gdead_stats();
	EXECUTE format('SELECT count(*) FROM (%s) x', q) INTO r;
	SELECT blocks INTO c1 FROM sift._test_col_gdead_stats();
	RETURN c1 - c0;
END $$;
SELECT orr_blocks('SELECT id, r FROM orr ORDER BY r, id LIMIT 20 OFFSET 7000') < 140 AS blocks_kept;
 blocks_kept 
-------------
 t
(1 row)

SELECT ob_chk(format('SELECT id, r AS k FROM orr ORDER BY r %s, id LIMIT 20 OFFSET %s', d, o)) AS result, count(*) FROM unnest(ARRAY['ASC', 'DESC']) d, unnest(ARRAY[2100, 7000]) o GROUP BY 1;
 result | count 
--------+-------
 bulk   |     4
(1 row)

DROP FUNCTION orr_blocks(text);
DROP TABLE orr;
DROP FUNCTION ob_all(), ob_filt(), ob_twice(text), ob_chk(text), ob_cut(text, bool);
DROP TABLE ob;
DROP EXTENSION pg_sift_test;
DROP EXTENSION pg_sift_helpers;
DROP EXTENSION pg_sift;