-- The global order grouped by a column: the arrays of order_array_columns are sorted by the group and then the key, so a scan that pins the group
-- (WHERE g = x ORDER BY k LIMIT n) reads the entries of that group only; the rows with no group are not in them, a query that does not pin the group reads the units. The answers are
-- the heap's through NULL keys, NULL groups, ties, both directions, ranges and pages after a key, DML, a promotion, a merge and a REINDEX.
CREATE EXTENSION pg_sift_test CASCADE;
CREATE EXTENSION pg_sift_helpers;
SET sift.async_maintenance = off;
SET sift.test_col_legacy_options = off;
SET max_parallel_maintenance_workers = 0;
SET max_parallel_workers_per_gather = 0;
SET client_min_messages = notice;
\set VERBOSITY terse

CREATE TABLE gg (id int, g int, k int, t timestamp, pad text DEFAULT repeat('x', 200));
ALTER TABLE gg ALTER COLUMN pad SET STORAGE PLAIN;
INSERT INTO gg (id, g, k, t)
SELECT i, CASE WHEN i % 53 = 0 THEN NULL ELSE (i * 7) % 9 END, CASE WHEN i % 31 = 0 THEN NULL ELSE (i * 7919) % 500 END,
	   CASE WHEN i % 43 = 0 THEN NULL ELSE TIMESTAMP '2000-01-01' + ((i * 104729) % 90000) * interval '1 second' END
FROM generate_series(1, 12000) i;
-- a group that is only NULL keys, and a group of one row
INSERT INTO gg (id, g, k, t) SELECT 20000 + i, 100, NULL, NULL FROM generate_series(1, 30) i;
INSERT INTO gg (id, g, k, t) VALUES (20100, 101, 5, TIMESTAMP '2001-01-01');
CREATE INDEX gg_i ON gg USING sift_col (id sift.col_int4_ops, g sift.col_int4_ops, k sift.col_int4_ops, t sift.col_timestamp_ops)
	WITH (order_array_columns = 'k, t', order_array_group = 'g');
SELECT sift.col_verify('gg_i');
ANALYZE gg;

-- an answer is right when the rows are the heap's, in order; 'global order' says the scan read the array of the group
CREATE FUNCTION gg_chk(q text) RETURNS text LANGUAGE sql AS $$ SELECT sift_helpers.topn_check_ordered_rows(q) $$;

CREATE FUNCTION gg_all() RETURNS TABLE (result text, n bigint) LANGUAGE sql AS
$$ SELECT gg_chk(format('SELECT id, %1$s FROM gg WHERE %2$s ORDER BY %1$s %3$s %4$s, id LIMIT %5$s', c, w, d, nl, l)), 1::bigint
   FROM (VALUES ('k'), ('t')) q(c),
		unnest(ARRAY['g = 3', 'g = 0', 'g = 8 AND id > 0', 'g = 100', 'g = 101', 'g = 500', 'g = 4 AND id > 6000', 'g = 2 AND id <= 3000',
									 'g IN (3, 4, 5)', 'g IN (0, 8, 8, 3)', 'g IN (100, 101, 500)']) w,
		unnest(ARRAY['ASC', 'DESC']) d, unnest(ARRAY['NULLS FIRST', 'NULLS LAST']) nl, unnest(ARRAY[1, 10, 100, 2000]) l $$;
SELECT result, count(*) FROM gg_all() GROUP BY 1 ORDER BY 1;

-- a range of the key and the page after a key inside the group
SELECT result, count(*) FROM (
	SELECT gg_chk(format('SELECT id, k FROM gg WHERE g = 5 AND %s ORDER BY k %s, id LIMIT %s', w, d, l)) AS result
	FROM unnest(ARRAY['k > 250', 'k <= 40', 'k BETWEEN 100 AND 120', 'k = 77', '(k, id) > (250, 5000)', 'k > 1000', 'k < 10 OR k > 490']) w, unnest(ARRAY['ASC', 'DESC']) d, unnest(ARRAY[1, 10, 100]) l) r
GROUP BY 1 ORDER BY 1;

-- a list of groups is read one group after the other and merged by the key; more values than the scan pins or a NULL in the list read the plain order or the units, and another condition on the group is tested on the heap row of each entry
SELECT result, count(*) FROM (
	SELECT gg_chk(format('SELECT id, k FROM gg WHERE %s ORDER BY k %s, id LIMIT %s', w, d, l)) AS result
	FROM unnest(ARRAY['g IN (1, 2)', 'g IN (3, 3)', 'g = ANY (ARRAY[6, 7, 8])', 'g IN (2, 4, 6) AND k > 250', 'g IN (2, 4, 6) AND (k, id) > (250, 5000)', 'g IN (100, 101)',
					  'g IN (500, 501)', 'g IN (0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15)', 'g IN (0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16)',
					  'g IN (3, NULL)', 'g IN (3, 4) OR g = 5', 'g NOT IN (3, 4)']) w, unnest(ARRAY['ASC', 'DESC']) d, unnest(ARRAY[1, 10, 100]) l) r
GROUP BY 1 ORDER BY 1;

-- no pin on the group, or a condition that leaves no bound on the key: the units (two groups written as OR are the list)
SELECT gg_chk('SELECT id, k FROM gg ORDER BY k, id LIMIT 10') AS no_group, gg_chk('SELECT id, k FROM gg WHERE g > 3 ORDER BY k, id LIMIT 10') AS group_range,
	   gg_chk('SELECT id, k FROM gg WHERE g = 3 OR g = 4 ORDER BY k, id LIMIT 10') AS two_groups, gg_chk('SELECT id, k FROM gg WHERE g = 3 AND t > TIMESTAMP ''2000-01-02'' ORDER BY k, id LIMIT 10') AS other_column;

-- the planner prices the read of the array as what it is: with the units dear to read (sift.plan_scale), a shape the array answers costs about the start of the scan, any other the
-- units it reads; 'array' is a priced read that the scan did, 'units' a scan priced as one, 'MISPRICED' a read priced as the array that the scan did not do
CREATE FUNCTION gg_plan(q text) RETURNS text LANGUAGE plpgsql AS $$
DECLARE pl text; cheap bool := false; used bool := false; agg bool := false;
BEGIN
	SET LOCAL sift.enable_col_scan = on; SET LOCAL enable_seqscan = off; SET LOCAL sift.plan_scale = 1000;
	FOR pl IN EXECUTE 'EXPLAIN (ANALYZE, COSTS on, TIMING off, SUMMARY off) ' || q LOOP
		agg := agg OR pl LIKE '%Custom Scan (SiftAgg)%';
		IF pl LIKE '%Custom Scan (SiftAgg)%' THEN cheap := substring(pl from 'cost=([0-9.]+)\.\.')::float8 < 5000; END IF;
		used := used OR pl LIKE '%Global Order%';
	END LOOP;
	RETURN CASE WHEN NOT agg THEN 'native' WHEN used AND cheap THEN 'array' WHEN NOT used AND NOT cheap THEN 'units' ELSE 'MISPRICED' END;
END $$;
SELECT gg_plan(q) AS plan, q FROM unnest(ARRAY[
 'SELECT id, k FROM gg WHERE g = 3 ORDER BY k DESC LIMIT 10',
 'SELECT id, k FROM gg WHERE g = 3 AND k >= 100 ORDER BY k LIMIT 10',
 'SELECT id, k FROM gg WHERE g = 3 AND k BETWEEN 100 AND 200 ORDER BY k DESC LIMIT 10',
 'SELECT id, k FROM gg WHERE g = 3 AND (k, id) > (100, 5) ORDER BY k, id LIMIT 10',
 'SELECT id, k FROM gg WHERE g = 3 AND (k < 20 OR k > 480) ORDER BY k LIMIT 10',
 'SELECT id, k FROM gg WHERE g = 3 ORDER BY k DESC LIMIT 10 OFFSET 5',
 'SELECT id, k FROM gg WHERE g > 3 ORDER BY k DESC LIMIT 10',
 'SELECT id, k FROM gg ORDER BY k DESC LIMIT 10',
 'SELECT id, k FROM gg WHERE g <> 3 AND k > 10 ORDER BY k LIMIT 10',
 'SELECT id, k FROM gg WHERE g = 3 AND k IS NOT NULL ORDER BY k LIMIT 10'
]) q;
DROP FUNCTION gg_plan(text);

-- min(k) and max(k) of a group: PostgreSQL plans them as the first row of the group in the order of k that is not NULL, and the scan reads the array for it (an empty group, a group of NULL
-- keys, ties, a range, deleted and updated rows and a write buffer give the heap's answer)
CREATE FUNCTION gg_mm(q text) RETURNS text LANGUAGE sql AS $$ SELECT sift_helpers.topn_check_min_max_with_plan_scale(q) $$;
CREATE FUNCTION gg_mm_all() RETURNS TABLE (result text, n bigint) LANGUAGE sql AS
$$ SELECT gg_mm(format('SELECT %s(%s) FROM gg WHERE %s', f, c, w)), count(*)
   FROM unnest(ARRAY['min', 'max']) f, unnest(ARRAY['k', 't']) c,
		unnest(ARRAY['g = 3', 'g = 0', 'g = 100', 'g = 101', 'g = 500', 'g = 8 AND k > 250', 'g = 8 AND k <= 40', 'g = 8 AND k = 77', 'g = 4 AND k > 1000']) w
   GROUP BY 1 ORDER BY 1 $$;
SELECT * FROM gg_mm_all();

-- rows after the index (the write buffer, filtered by the group), deletes, updates of the group and of the key, VACUUM, a promotion and a merge
INSERT INTO gg (id, g, k, t) SELECT 30000 + i, i % 11, (i * 13) % 300 - 20, TIMESTAMP '2003-01-01' + i * interval '1 minute' FROM generate_series(1, 2500) i;
DELETE FROM gg WHERE id % 11 = 0;
UPDATE gg SET g = coalesce(g, 0) + 1 WHERE id % 7 = 0;
UPDATE gg SET k = coalesce(k, 0) + 1000 WHERE id % 19 = 0;
SELECT result, count(*) FROM gg_all() GROUP BY 1 ORDER BY 1;
SELECT * FROM gg_mm_all();
VACUUM gg;
SELECT result, count(*) FROM gg_all() GROUP BY 1 ORDER BY 1;
SELECT * FROM gg_mm_all();
SELECT sift.col_promote('gg_i'::regclass) >= 0 AS promoted;
SELECT sift.col_verify('gg_i');
SELECT result, count(*) FROM gg_all() GROUP BY 1 ORDER BY 1;
SELECT sift.col_merge('gg_i'::regclass, true) >= 0 AS merged;
SELECT sift.col_verify('gg_i');
SELECT result, count(*) FROM gg_all() GROUP BY 1 ORDER BY 1;
-- a write buffer of 5,000 rows is cut into ranges of the key and the few the read touches are sorted: the order answers
INSERT INTO gg (id, g, k, t) SELECT 50000 + i, 3, i, NULL FROM generate_series(1, 5000) i;
SELECT gg_chk('SELECT id, k FROM gg WHERE g = 3 ORDER BY k, id LIMIT 10') AS big_buffer;

-- the group and the bounds of a probe come from the outer query: a LATERAL join and a correlated sub-select take them at each probe (PARAM_EXEC), where the scan once pinned group 0 for all of them and returned no row
SELECT gg_chk('SELECT v.g, x.id, x.k FROM generate_series(0, 9) v(g), LATERAL (SELECT id, k FROM gg WHERE g = v.g ORDER BY k DESC, id LIMIT 1) x ORDER BY v.g') AS lateral_group;
SELECT gg_chk('SELECT v.g, x.id, x.t FROM generate_series(0, 9) v(g), LATERAL (SELECT id, t FROM gg WHERE g = v.g AND t > TIMESTAMP ''2000-01-01'' + v.g * interval ''2 hours'' ORDER BY t, id LIMIT 3) x ORDER BY v.g, x.t, x.id') AS lateral_group_and_bound;
SELECT gg_chk('SELECT v.g, (SELECT k FROM gg WHERE g = v.g ORDER BY k DESC, id LIMIT 1) FROM generate_series(0, 9) v(g) ORDER BY v.g') AS subselect_group;
SELECT gg_chk('SELECT v.g, (SELECT max(k) FROM gg WHERE g = v.g) FROM generate_series(0, 9) v(g) ORDER BY v.g') AS subselect_max;
-- the scan starts over at each probe and keeps what the plan fixes: the same operand twice, NULL, a group with no array entry, a bound that is no pin (the units), a rewind
SELECT gg_chk('SELECT v.i, v.g, x.id, x.k FROM (VALUES (1, 3), (2, 3), (3, NULL), (4, 100), (5, 101), (6, 7), (7, 3), (8, 55)) v(i, g), LATERAL (SELECT id, k FROM gg WHERE g = v.g ORDER BY k DESC, id LIMIT 2) x ORDER BY v.i, x.k DESC, x.id') AS lateral_repeats;
SELECT gg_chk('SELECT v.g, x.id, x.k FROM generate_series(0, 4) v(g), LATERAL (SELECT id, k FROM gg WHERE k > v.g * 90 ORDER BY k, id LIMIT 2) x ORDER BY v.g, x.k, x.id') AS lateral_bound_only;
DO $$
DECLARE c refcursor; r record; fwd text := ''; bwd text := ''; want text;
BEGIN
	SET LOCAL sift.enable_agg = 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;
	OPEN c SCROLL FOR SELECT v.g, x.id, x.k FROM generate_series(0, 9) v(g), LATERAL (SELECT id, k FROM gg WHERE g = v.g ORDER BY k DESC, id LIMIT 1) x;
	LOOP FETCH NEXT FROM c INTO r; EXIT WHEN NOT FOUND; fwd := fwd || r::text; END LOOP;
	LOOP FETCH PRIOR FROM c INTO r; EXIT WHEN NOT FOUND; bwd := r::text || bwd; END LOOP;
	CLOSE c;
	SET LOCAL sift.enable_agg = off; SET LOCAL enable_seqscan = on; SET LOCAL enable_sort = on;
	SELECT string_agg(x::text, '' ORDER BY g) INTO want FROM (SELECT v.g, x.id, x.k FROM generate_series(0, 9) v(g), LATERAL (SELECT id, k FROM gg WHERE g = v.g ORDER BY k DESC, id LIMIT 1) x) x;
	RAISE NOTICE 'cursor: % %', CASE WHEN fwd = want THEN 'forward is the heap''s' ELSE 'DIFFERS forward' END, CASE WHEN bwd = want THEN 'backward is the same' ELSE 'DIFFERS backward' END;
END $$;

-- segments written without the group: a plain array answers with the condition tested on the heap row (the pin does not narrow it: the units)
DROP INDEX gg_i;
CREATE INDEX gg_p ON gg USING sift_col (id sift.col_int4_ops, g sift.col_int4_ops, k sift.col_int4_ops) WITH (order_array_columns = 'k');
SELECT gg_chk('SELECT id, k FROM gg WHERE g = 3 ORDER BY k, id LIMIT 10') AS plain_array;
DROP INDEX gg_p;

-- the option is validated: one column, an integer, date or timestamp, not an array column
CREATE INDEX gg_bad ON gg USING sift_col (id sift.col_int4_ops, k sift.col_int4_ops) WITH (order_array_columns = 'k', order_array_group = 'k');
CREATE INDEX gg_bad ON gg USING sift_col (id sift.col_int4_ops, g sift.col_int4_ops, k sift.col_int4_ops) WITH (order_array_columns = 'k', order_array_group = 'g, id');
CREATE INDEX gg_bad ON gg USING sift_col (id sift.col_int4_ops, pad sift.col_text_ops, k sift.col_int4_ops) WITH (order_array_columns = 'k', order_array_group = 'pad');

DROP TABLE gg CASCADE;
DROP FUNCTION gg_chk(text), gg_all(), gg_mm(text), gg_mm_all();
DROP EXTENSION pg_sift_helpers;
DROP EXTENSION pg_sift_test, pg_sift CASCADE;