-- Lion indexes under debug_discard_caches (2026-09-24): every catalog read may
-- flush any cache entry, this index's relcache entry and its rd_amcache
-- included.  The recorded order's comparison check (DESIGN.md §21) used to
-- read the identity out of a state that such a flush had just freed, and every
-- lion index then refused to open with a spurious REINDEX error.
--
-- The setting exists only in builds with assertions (DISCARD_CACHES_ENABLED);
-- elsewhere lion_ddc() runs the same queries without it, so the expected
-- output is the same either way.  The tables are small: every query here reads
-- the catalog afresh, many times over.
\set VERBOSITY terse
SET client_min_messages = warning;
CREATE EXTENSION IF NOT EXISTS pg_lion;
CREATE EXTENSION IF NOT EXISTS citext;
CREATE EXTENSION IF NOT EXISTS pg_lion_citext;
RESET client_min_messages;
SET synchronous_commit = on;
CREATE FUNCTION lion_ddc(q text) RETURNS text
LANGUAGE plpgsql AS $$
DECLARE
	r text;
BEGIN
	BEGIN
		PERFORM set_config('debug_discard_caches', '1', true);
	EXCEPTION WHEN OTHERS THEN
		NULL;					-- not an assert build: run it without
	END;
	EXECUTE format('SELECT string_agg(s::text, '' '' ORDER BY s::text) FROM (%s) s', q)
		INTO r;
	PERFORM set_config('debug_discard_caches', '0', true);
	RETURN r;
END $$;
CREATE FUNCTION lion_ddc_do(q text) RETURNS void
LANGUAGE plpgsql AS $$
BEGIN
	BEGIN
		PERFORM set_config('debug_discard_caches', '1', true);
	EXCEPTION WHEN OTHERS THEN
		NULL;
	END;
	EXECUTE q;
	PERFORM set_config('debug_discard_caches', '0', true);
END $$;
-- a custom ordered opclass: a string-body comparison a btree class sorts with
CREATE FUNCTION lion_ddc_cmp(int4, int4) RETURNS int4 LANGUAGE sql IMMUTABLE STRICT
	AS 'SELECT btint4cmp($1, $2)';
CREATE FUNCTION lion_ddc_lt(int4, int4) RETURNS bool LANGUAGE sql IMMUTABLE STRICT AS 'SELECT $1 < $2';
CREATE FUNCTION lion_ddc_le(int4, int4) RETURNS bool LANGUAGE sql IMMUTABLE STRICT AS 'SELECT $1 <= $2';
CREATE FUNCTION lion_ddc_eq(int4, int4) RETURNS bool LANGUAGE sql IMMUTABLE STRICT AS 'SELECT $1 = $2';
CREATE FUNCTION lion_ddc_ge(int4, int4) RETURNS bool LANGUAGE sql IMMUTABLE STRICT AS 'SELECT $1 >= $2';
CREATE FUNCTION lion_ddc_gt(int4, int4) RETURNS bool LANGUAGE sql IMMUTABLE STRICT AS 'SELECT $1 > $2';
CREATE OPERATOR <<~ (LEFTARG = int4, RIGHTARG = int4, FUNCTION = lion_ddc_lt);
CREATE OPERATOR <=~ (LEFTARG = int4, RIGHTARG = int4, FUNCTION = lion_ddc_le);
CREATE OPERATOR ==~ (LEFTARG = int4, RIGHTARG = int4, FUNCTION = lion_ddc_eq);
CREATE OPERATOR >=~ (LEFTARG = int4, RIGHTARG = int4, FUNCTION = lion_ddc_ge);
CREATE OPERATOR >>~ (LEFTARG = int4, RIGHTARG = int4, FUNCTION = lion_ddc_gt);
CREATE OPERATOR CLASS lion_ddc_bt FOR TYPE int4 USING btree AS
	OPERATOR 1 <<~, OPERATOR 2 <=~, OPERATOR 3 ==~, OPERATOR 4 >=~, OPERATOR 5 >>~,
	FUNCTION 1 lion_ddc_cmp(int4, int4);
CREATE OPERATOR FAMILY lion_ddc_fam USING lion;
CREATE OPERATOR CLASS lion_ddc_ops FOR TYPE int4 USING lion FAMILY lion_ddc_fam AS
	OPERATOR 1 = (int4, int4),
	OPERATOR 6 < (int4, int4),
	OPERATOR 9 > (int4, int4),
	FUNCTION 1 hashint4(int4),
	FUNCTION 4 lion_ddc_cmp(int4, int4);
CREATE TABLE lion_ddct (id int NOT NULL, t text, k int, c citext, u int);
INSERT INTO lion_ddct SELECT i, 'v' || (i % 40), i % 40, chr(65 + i % 26) || (i % 3), i % 50
  FROM generate_series(1, 600) i;
CREATE INDEX lion_ddct_t ON lion_ddct USING lion (t);
CREATE INDEX lion_ddct_k ON lion_ddct USING lion (k);
CREATE INDEX lion_ddct_c ON lion_ddct USING lion (c);
CREATE INDEX lion_ddct_u ON lion_ddct USING lion (u lion_ddc_ops);
VACUUM ANALYZE lion_ddct;
SELECT attno, ordered FROM lion_index_stats('lion_ddct_u');
 attno | ordered 
-------+---------
     1 | t
(1 row)

SET enable_seqscan = off;
-- the count pushdown
SELECT lion_ddc('SELECT count(*) FROM lion_ddct WHERE t = ''v7''');
 lion_ddc 
----------
 (15)
(1 row)

SELECT lion_ddc('SELECT count(*) FROM lion_ddct WHERE t < ''v2''');
 lion_ddc 
----------
 (180)
(1 row)

SELECT lion_ddc('SELECT k, count(*) FROM lion_ddct WHERE k BETWEEN 3 AND 5 GROUP BY k');
       lion_ddc       
----------------------
 (3,15) (4,15) (5,15)
(1 row)

SELECT lion_ddc('SELECT count(*) FROM lion_ddct WHERE c = ''b1''');
 lion_ddc 
----------
 (8)
(1 row)

SELECT lion_ddc('SELECT count(*) FROM lion_ddct WHERE c >= ''y''');
 lion_ddc 
----------
 (46)
(1 row)

SELECT lion_ddc('SELECT count(*) FROM lion_ddct WHERE u = 7');
 lion_ddc 
----------
 (12)
(1 row)

SELECT lion_ddc('SELECT count(*) FROM lion_ddct WHERE u > 45');
 lion_ddc 
----------
 (48)
(1 row)

SELECT lion_ddc('SELECT count(DISTINCT k) FROM lion_ddct WHERE k < 10');
 lion_ddc 
----------
 (10)
(1 row)

-- the bitmap scan
SET pg_lion.enable_count_pushdown = off;
SELECT lion_ddc('SELECT count(*) FROM lion_ddct WHERE t BETWEEN ''v1'' AND ''v3''');
 lion_ddc 
----------
 (345)
(1 row)

SELECT lion_ddc('SELECT count(*) FROM lion_ddct WHERE k = 11 AND u < 20');
 lion_ddc 
----------
 (6)
(1 row)

SELECT lion_ddc('SELECT count(*) FROM lion_ddct WHERE c < ''C''');
 lion_ddc 
----------
 (47)
(1 row)

SELECT lion_ddc('SELECT count(*) FROM lion_ddct WHERE u = 7');
 lion_ddc 
----------
 (12)
(1 row)

-- the plain index scan
SET enable_bitmapscan = off;
SELECT lion_ddc('SELECT id FROM lion_ddct WHERE t = ''v7'' ORDER BY id LIMIT 3');
   lion_ddc    
---------------
 (47) (7) (87)
(1 row)

SELECT lion_ddc('SELECT id FROM lion_ddct WHERE k = 11 ORDER BY id LIMIT 3');
    lion_ddc    
----------------
 (11) (51) (91)
(1 row)

-- and a write under it
SELECT lion_ddc_do('INSERT INTO lion_ddct VALUES (601, ''v7'', 7, ''b1'', 7)');
 lion_ddc_do 
-------------
 
(1 row)

RESET enable_bitmapscan;
SELECT lion_ddc('SELECT count(*) FROM lion_ddct WHERE t = ''v7'' AND u = 7');
 lion_ddc 
----------
 (4)
(1 row)

RESET pg_lion.enable_count_pushdown;
RESET enable_seqscan;
SELECT lion_index_verify('lion_ddct_u', true);
 lion_index_verify 
-------------------
 
(1 row)

DROP TABLE lion_ddct;
-- LionOrdered (DESIGN.md §30) under the same flushes: its set, its btree walk
-- and the fetch-and-sort switch, each compared in order with the ordinary
-- plan.  c is correlated with k, so `c = 39` lies at the far end of the walk
-- and the switch fires once 10000 entries have been walked.
CREATE TABLE lion_ddco (id int NOT NULL, k int NOT NULL, c int NOT NULL, g int NOT NULL);
INSERT INTO lion_ddco SELECT i, i, (i - 1) / 300, i % 7 FROM generate_series(1, 12000) i;
CREATE INDEX lion_ddco_k ON lion_ddco (k, id);
CREATE INDEX lion_ddco_c ON lion_ddco USING lion (c);
CREATE INDEX lion_ddco_g ON lion_ddco USING lion (g);
VACUUM (FREEZE, ANALYZE) lion_ddco;
-- q through LionOrdered under debug_discard_caches (every other scan off),
-- then without the node and without the flushes: whether the node ran, and
-- whether the two agree row for row, in order.
CREATE FUNCTION lion_ddc_ord(q text, want text DEFAULT NULL) RETURNS text
LANGUAGE plpgsql AS $$
DECLARE
	ln text;
	r record;
	used boolean := false;
	switched boolean := false;
	a text[] := '{}';
	b text[] := '{}';
BEGIN
	PERFORM set_config('enable_seqscan', 'off', true);
	PERFORM set_config('enable_bitmapscan', 'off', true);
	PERFORM set_config('pg_lion.enable_plain_scan', 'off', true);
	PERFORM set_config('pg_lion.pushdown_margin', '1000', true);
	PERFORM set_config('enable_indexonlyscan', 'off', true);
	BEGIN
		PERFORM set_config('debug_discard_caches', '1', true);
	EXCEPTION WHEN OTHERS THEN
		NULL;
	END;
	FOR ln IN EXECUTE 'EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF) ' || q LOOP
		used := used OR ln ~ 'Custom Scan \(LionOrdered\)';
		switched := switched OR ln ~ 'Switched to Fetch and Sort: [1-9]';
	END LOOP;
	FOR r IN EXECUTE q LOOP
		a := a || r::text;
	END LOOP;
	PERFORM set_config('debug_discard_caches', '0', true);
	PERFORM set_config('enable_seqscan', 'on', true);
	PERFORM set_config('enable_bitmapscan', 'on', true);
	PERFORM set_config('pg_lion.enable_plain_scan', 'on', true);
	PERFORM set_config('pg_lion.pushdown_margin', '1', true);
	PERFORM set_config('enable_indexonlyscan', 'on', true);
	PERFORM set_config('pg_lion.enable_ordered_scan', 'off', true);
	FOR r IN EXECUTE q LOOP
		b := b || r::text;
	END LOOP;
	PERFORM set_config('pg_lion.enable_ordered_scan', 'on', true);
	RETURN CASE WHEN used THEN 'LionOrdered' ELSE 'no LionOrdered' END ||
		CASE WHEN switched THEN ', switched' ELSE '' END ||
		CASE WHEN a = b THEN ', ' || cardinality(a) || ' rows' ELSE ', MISMATCH' END;
END $$;
SELECT lion_ddc_ord('SELECT id, k FROM lion_ddco WHERE c = 7 AND g = 3 ORDER BY k, id LIMIT 10');
     lion_ddc_ord     
----------------------
 LionOrdered, 10 rows
(1 row)

SELECT lion_ddc_ord('SELECT id, k FROM lion_ddco WHERE c = 7 ORDER BY k DESC, id DESC LIMIT 10');
     lion_ddc_ord     
----------------------
 LionOrdered, 10 rows
(1 row)

SELECT lion_ddc_ord('SELECT id, k FROM lion_ddco WHERE c = 39 ORDER BY k, id LIMIT 5');
         lion_ddc_ord          
-------------------------------
 LionOrdered, switched, 5 rows
(1 row)

SELECT lion_ddc_ord('SELECT o.x, s.id FROM generate_series(36, 39) o(x),
	LATERAL (SELECT id FROM lion_ddco WHERE c = o.x ORDER BY k, id LIMIT 3) s');
          lion_ddc_ord          
--------------------------------
 LionOrdered, switched, 12 rows
(1 row)

-- a cursor paused between fetches while rows the snapshot cannot see arrive;
-- the flushes run through all of it
CREATE FUNCTION lion_ddc_cursor() RETURNS text
LANGUAGE plpgsql AS $$
DECLARE
	cur refcursor;
	r record;
	ln text;
	used boolean := false;
	a text[] := '{}';
	b text[] := '{}';
	q text := 'SELECT id, k FROM lion_ddco WHERE c IN (5, 38) ORDER BY k, id';
BEGIN
	PERFORM set_config('enable_seqscan', 'off', true);
	PERFORM set_config('enable_bitmapscan', 'off', true);
	PERFORM set_config('pg_lion.enable_plain_scan', 'off', true);
	PERFORM set_config('pg_lion.pushdown_margin', '1000', true);
	PERFORM set_config('enable_indexonlyscan', 'off', true);
	BEGIN
		PERFORM set_config('debug_discard_caches', '1', true);
	EXCEPTION WHEN OTHERS THEN
		NULL;
	END;
	FOR ln IN EXECUTE 'EXPLAIN (COSTS OFF) ' || q LOOP
		used := used OR ln ~ 'Custom Scan \(LionOrdered\)';
	END LOOP;
	OPEN cur FOR EXECUTE q;
	FOR i IN 1 .. 3 LOOP
		FETCH cur INTO r;
		a := a || r::text;
	END LOOP;
	INSERT INTO lion_ddco VALUES (12001, 0, 5, 0), (12002, 11999, 38, 0);
	LOOP
		FETCH cur INTO r;
		EXIT WHEN NOT FOUND;
		a := a || r::text;
	END LOOP;
	CLOSE cur;
	PERFORM set_config('debug_discard_caches', '0', true);
	PERFORM set_config('pg_lion.enable_ordered_scan', 'off', true);
	FOR r IN EXECUTE q || ' ' LOOP
		CONTINUE WHEN r.id > 12000;			-- inserted after the cursor opened
		b := b || r::text;
	END LOOP;
	RETURN CASE WHEN used THEN 'LionOrdered' ELSE 'no LionOrdered' END ||
		CASE WHEN a = b THEN ', ' || cardinality(a) || ' rows, same' ELSE ', MISMATCH' END;
END $$;
SELECT lion_ddc_cursor();
       lion_ddc_cursor       
-----------------------------
 LionOrdered, 600 rows, same
(1 row)

SELECT lion_index_verify('lion_ddco_c', true);
 lion_index_verify 
-------------------
 
(1 row)

DROP TABLE lion_ddco;
DROP FUNCTION lion_ddc_ord(text, text), lion_ddc_cursor();
DROP OPERATOR FAMILY lion_ddc_fam USING lion;
DROP OPERATOR CLASS lion_ddc_bt USING btree;
DROP OPERATOR <<~ (int4, int4);
DROP OPERATOR <=~ (int4, int4);
DROP OPERATOR ==~ (int4, int4);
DROP OPERATOR >=~ (int4, int4);
DROP OPERATOR >>~ (int4, int4);
DROP FUNCTION lion_ddc_lt(int4, int4), lion_ddc_le(int4, int4), lion_ddc_eq(int4, int4),
	lion_ddc_ge(int4, int4), lion_ddc_gt(int4, int4), lion_ddc_cmp(int4, int4);
DROP FUNCTION lion_ddc(text), lion_ddc_do(text);