-- The LionCount CustomScan (DESIGN.md section 10).
--
-- Two things are checked for every query: that the plan is (or deliberately
-- is not) the custom node, and that the rows it produces are exactly the rows
-- the ordinary plan produces.  The second check is done inside the test with
-- EXCEPT ALL in both directions, so a wrong answer fails even if it happens
-- to be stable.
\set VERBOSITY terse
SET client_min_messages = warning;
LOAD 'pg_lion';

/*
 * The visibility map is what makes this feature worth having, and a heap page
 * only becomes all-visible once the inserting transaction's commit record has
 * reached disk.  The dev cluster runs with synchronous_commit = off, so ask
 * for flushed commits here; otherwise whether VACUUM can set the bits depends
 * on how fast the WAL writer happens to be.
 */
SET synchronous_commit = on;

/*
 * Plans are compared literally below, so the statistics must not depend on
 * ANALYZE's random sample: a target of 1000 makes it read every row of the
 * 100k-row table used here.
 */
SET default_statistics_target = 1000;

CREATE EXTENSION IF NOT EXISTS pg_lion;

/*
 * lion_pd() runs one query twice, once with the pushdown enabled and once
 * without, and reports whether the custom node was used and how many rows
 * came out - after proving the two result sets are equal as multisets.
 */
CREATE OR REPLACE FUNCTION lion_pd(q text) RETURNS text
LANGUAGE plpgsql AS $$
DECLARE
	ln text;
	pushed boolean := false;
	nrows bigint;
	ndiff bigint;
BEGIN
	PERFORM set_config('pg_lion.enable_count_pushdown', 'on', true);
	FOR ln IN EXECUTE 'EXPLAIN (COSTS OFF) ' || q LOOP
		IF ln LIKE '%Custom Scan (LionCount)%' THEN
			pushed := true;
		END IF;
	END LOOP;
	EXECUTE format('CREATE TEMP TABLE lion_pd_on AS %s', q);

	PERFORM set_config('pg_lion.enable_count_pushdown', 'off', true);
	EXECUTE format('CREATE TEMP TABLE lion_pd_off AS %s', q);
	PERFORM set_config('pg_lion.enable_count_pushdown', 'on', true);

	EXECUTE 'SELECT count(*) FROM lion_pd_on' INTO nrows;
	EXECUTE 'SELECT (SELECT count(*) FROM (SELECT * FROM lion_pd_on EXCEPT ALL SELECT * FROM lion_pd_off) a)'
			' + (SELECT count(*) FROM (SELECT * FROM lion_pd_off EXCEPT ALL SELECT * FROM lion_pd_on) b)'
		INTO ndiff;
	EXECUTE 'DROP TABLE lion_pd_on, lion_pd_off';

	IF ndiff <> 0 THEN
		RETURN format('MISMATCH: %s rows differ', ndiff);
	END IF;
	RETURN format('%s, %s rows',
				  CASE WHEN pushed THEN 'pushed down' ELSE 'not pushed down' END,
				  nrows);
END $$;

/* Show both plans for one query. */
CREATE OR REPLACE FUNCTION lion_plans(q text) RETURNS SETOF text
LANGUAGE plpgsql AS $$
DECLARE
	ln text;
BEGIN
	PERFORM set_config('pg_lion.enable_count_pushdown', 'on', true);
	RETURN NEXT '-- on:';
	FOR ln IN EXECUTE 'EXPLAIN (COSTS OFF) ' || q LOOP
		-- 18 marks a disabled node; 16 and 17 do not
		CONTINUE WHEN ln ~ '^\s*Disabled: true$';
		RETURN NEXT ln;
	END LOOP;
	PERFORM set_config('pg_lion.enable_count_pushdown', 'off', true);
	RETURN NEXT '-- off:';
	FOR ln IN EXECUTE 'EXPLAIN (COSTS OFF) ' || q LOOP
		CONTINUE WHEN ln ~ '^\s*Disabled: true$';
		RETURN NEXT ln;
	END LOOP;
	PERFORM set_config('pg_lion.enable_count_pushdown', 'on', true);
END $$;

/*
 * What the model charges for the node itself, with every other plan
 * discouraged so that the node is the plan whatever the estimate says.  The
 * number is never printed - it moves whenever the model is recalibrated -
 * only ever compared with another one of these, which is how a statement
 * ABOUT the shape of the estimate can be pinned without pinning the estimate.
 */
CREATE OR REPLACE FUNCTION lion_pd_cost(q text) RETURNS float8
LANGUAGE plpgsql AS $$
DECLARE
	p json;
BEGIN
	PERFORM set_config('enable_bitmapscan', 'off', true);
	PERFORM set_config('enable_seqscan', 'off', true);
	PERFORM set_config('pg_lion.enable_plain_scan', 'off', true);
	PERFORM set_config('pg_lion.enable_count_pushdown', 'on', true);
	EXECUTE 'EXPLAIN (FORMAT JSON) ' || q INTO p;
	IF p -> 0 -> 'Plan' ->> 'Custom Plan Provider' IS DISTINCT FROM 'LionCount' THEN
		RAISE EXCEPTION 'not the node: %', p -> 0 -> 'Plan' ->> 'Node Type';
	END IF;
	RETURN (p -> 0 -> 'Plan' ->> 'Total Cost')::float8;
END $$;

/*
 * The instrumentation EXPLAIN ANALYZE prints for the custom node, reduced to
 * zero / non-zero: the exact counts depend on how the rows fall on heap
 * pages, but whether the heap had to be visited at all does not.
 */
CREATE OR REPLACE FUNCTION lion_pd_counters(q text) RETURNS SETOF text
LANGUAGE plpgsql AS $$
DECLARE
	ln text;
	nm text;
	val bigint;
	pushed boolean := false;
BEGIN
	FOR ln IN EXECUTE 'EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF, BUFFERS OFF) ' || q LOOP
		IF ln LIKE '%Custom Scan (LionCount)%' THEN
			pushed := true;
		END IF;
		nm := btrim(split_part(ln, ':', 1));
		IF nm IN ('Heap Blocks Skipped via VM', 'Heap TIDs Rechecked',
				  'Heap Blocks Rechecked', 'Containers Visited',
				  'Heap Blocks From Cache',
				  'Heap Blocks Past Cache Budget') THEN
			val := btrim(split_part(ln, ':', 2))::bigint;
			RETURN NEXT format('%s %s', nm,
							   CASE WHEN val = 0 THEN '= 0' ELSE '> 0' END);
		END IF;
	END LOOP;
	IF NOT pushed THEN
		RETURN NEXT 'NOT PUSHED DOWN';
	END IF;
END $$;

CREATE TABLE lion_pdt (
	id	int		NOT NULL,
	a	int		NOT NULL,
	b	int		NOT NULL,
	c	text	NOT NULL,
	n	int
);
INSERT INTO lion_pdt
SELECT i,
	   i % 10,
	   i % 7,
	   'c' || (i % 4),
	   CASE WHEN i % 101 = 0 THEN NULL ELSE i % 5 END
  FROM generate_series(1, 100000) i;

CREATE INDEX lion_pdt_a ON lion_pdt USING lion (a);
CREATE INDEX lion_pdt_b ON lion_pdt USING lion (b);
CREATE INDEX lion_pdt_c ON lion_pdt USING lion (c);
CREATE INDEX lion_pdt_n ON lion_pdt USING lion (n);
-- The pushdown only wins when most of the heap is all-visible, which is the
-- case it exists for; VACUUM makes that true and deterministic here.
VACUUM (FREEZE, ANALYZE) lion_pdt;

-- ---- plans -------------------------------------------------------------
SELECT lion_plans('SELECT count(*) FROM lion_pdt WHERE a = 3');
SELECT lion_plans('SELECT count(*) FROM lion_pdt WHERE a = 3 AND b = 2');
SELECT lion_plans($$SELECT count(*) FROM lion_pdt WHERE a = 3 AND b = 2 AND c = 'c1'$$);
SELECT lion_plans('SELECT a, count(*) FROM lion_pdt GROUP BY a');
SELECT lion_plans('SELECT a, count(*) FROM lion_pdt WHERE b = 2 GROUP BY a');
SELECT lion_plans('SELECT count(*) FROM lion_pdt GROUP BY a');
SELECT lion_plans('SELECT count(a) FROM lion_pdt WHERE a = 3');
SELECT lion_plans('SELECT a, count(*) FROM lion_pdt WHERE a = 3 GROUP BY a');
SELECT lion_plans('SELECT n, count(*) FROM lion_pdt GROUP BY n');
SELECT lion_plans('SELECT count(*) FROM lion_pdt WHERE n IS NULL');

-- ---- shapes that must push down ----------------------------------------
SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE a = 3');
SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE a = 3 AND b = 2');
SELECT lion_pd($$SELECT count(*) FROM lion_pdt WHERE a = 3 AND b = 2 AND c = 'c1'$$);
SELECT lion_pd($$SELECT count(*) FROM lion_pdt WHERE c = 'c2'$$);
-- the constant on the left, and a cross-type constant
SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE 3 = a');
SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE a = 3::int8');
-- a constant that matches no key at all
SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE a = 999');
-- the same column constrained twice with the same constant
SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE a = 3 AND a = 3');
-- GROUP BY, with and without a WHERE clause
SELECT lion_pd('SELECT a, count(*) FROM lion_pdt GROUP BY a');
SELECT lion_pd('SELECT a, count(*) FROM lion_pdt WHERE b = 2 GROUP BY a');
SELECT lion_pd($$SELECT a, count(*) FROM lion_pdt WHERE b = 2 AND c = 'c1' GROUP BY a$$);
SELECT lion_pd('SELECT count(*) FROM lion_pdt GROUP BY a');
SELECT lion_pd('SELECT b, count(*) FROM lion_pdt WHERE b = 2 GROUP BY b');
/*
 * GROUP BY a column the WHERE clause pins to a single value.  The planner
 * folds such a column out of the group clause, so the node emits at most one
 * row and reports the key the index stored for it - which is what makes a
 * cross-type constant safe here.
 */
SELECT lion_pd('SELECT a, count(*) FROM lion_pdt WHERE a = 3 GROUP BY a');
SELECT lion_pd('SELECT a, count(*) FROM lion_pdt WHERE a = 3::int8 GROUP BY a');
SELECT lion_pd('SELECT a, count(*) FROM lion_pdt WHERE a = 999 GROUP BY a');
SELECT lion_pd('SELECT a, b, count(*) FROM lion_pdt WHERE a = 3 GROUP BY a, b');
SELECT lion_pd($$SELECT c, count(*) FROM lion_pdt WHERE c = 'c2' GROUP BY c$$);
SELECT lion_pd('SELECT b, count(*) FROM lion_pdt WHERE b = 2 GROUP BY b');
-- a by-reference group key whose type prints what it stores
SELECT lion_pd('SELECT c, count(*) FROM lion_pdt GROUP BY c');
-- count(col) where col cannot be NULL
SELECT lion_pd('SELECT count(a) FROM lion_pdt WHERE a = 3');
SELECT lion_pd('SELECT a, count(a) FROM lion_pdt GROUP BY a');
SELECT lion_pd('SELECT count(*) AS c1, count(a) AS c2, count(*) AS c3 FROM lion_pdt WHERE a = 3');

/*
 * A nullable group column is fine: the NULL keys have an entry of their own,
 * so the NULL group is produced like any other (DESIGN.md section 14, and
 * test/sql/null.sql for the whole story).
 */
SELECT lion_pd('SELECT n, count(*) FROM lion_pdt GROUP BY n');

-- ---- HAVING on the count (DESIGN.md section 10) --------------------------
-- The node knows a group's count before it emits the group, so a HAVING over
-- the counts and the columns it can print is a filter on its own output.
-- Every group of lion_pdt.a has exactly 10000 rows.
SELECT lion_pd('SELECT a, count(*) FROM lion_pdt GROUP BY a HAVING count(*) >= 10000');
SELECT lion_pd('SELECT a, count(*) FROM lion_pdt GROUP BY a HAVING count(*) > 10000');
-- a count the HAVING alone mentions is a column the projection does not print
SELECT lion_pd('SELECT a FROM lion_pdt GROUP BY a HAVING count(*) >= 10000');
SELECT lion_pd('SELECT a FROM lion_pdt GROUP BY a HAVING count(*) > 10000');
-- the plain aggregate is one row, which a HAVING may drop
SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE a = 3 HAVING count(*) = 10000');
SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE a = 3 HAVING count(*) > 10000');
-- a two-column group, and two conditions on the count
SELECT lion_pd('SELECT a, b, count(*) FROM lion_pdt GROUP BY a, b HAVING count(*) >= 1429 AND count(*) < 2000');
-- count(n) of the group column is 0 in the NULL group (section 14)
SELECT lion_pd('SELECT n, count(*) FROM lion_pdt GROUP BY n HAVING count(n) = 0');
SELECT lion_pd('SELECT n FROM lion_pdt GROUP BY n HAVING count(n) < count(*)');
-- the count compared with a pinned column and with an InitPlan's value
SELECT lion_pd('SELECT b, count(*) FROM lion_pdt WHERE a = 3 GROUP BY a, b HAVING count(*) > a * 400 + (SELECT 228)');
-- EXPLAIN in a form every supported release prints the same way: subplans
-- are named "expr_N" as in PostgreSQL 19 ("N" before), 18's "Disabled: true"
-- lines are dropped, actual row counts are integers (18 adds ".00"), and
-- sorting is off while it plans, so that a query the pushdown must refuse gets
-- the same core plan whether disabled paths are counted (18) or priced (16,
-- 17).  LionCount never sorts, so this cannot hide it.
CREATE OR REPLACE FUNCTION lion_explain_norm(q text, opts text DEFAULT 'COSTS OFF')
RETURNS SETOF text
LANGUAGE plpgsql AS $$
DECLARE
	l text;
BEGIN
	PERFORM set_config('enable_sort', 'off', true);
	FOR l IN EXECUTE 'EXPLAIN (' || opts || ') ' || q LOOP
		CONTINUE WHEN l ~ '^\s*Disabled: true$';
		l := regexp_replace(l, '(InitPlan|SubPlan) (\d+)', '\1 expr_\2', 'g');
		RETURN NEXT regexp_replace(l, 'rows=(\d+)\.00 ', 'rows=\1 ', 'g');
	END LOOP;
	PERFORM set_config('enable_sort', 'on', true);
END
$$;
-- the filter is the plan's qual: EXPLAIN prints it, ANALYZE counts what it removed
EXPLAIN (COSTS OFF) SELECT a FROM lion_pdt GROUP BY a HAVING count(*) > 10000;
SELECT * FROM lion_explain_norm('SELECT a, count(*) FROM lion_pdt WHERE b = 2 GROUP BY a HAVING count(*) > 1428',
								'ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF, BUFFERS OFF') AS p("QUERY PLAN");
-- a parameter in the HAVING, under a generic plan
PREPARE lion_ph(bigint) AS SELECT a FROM lion_pdt GROUP BY a HAVING count(*) >= $1;
SET plan_cache_mode = force_generic_plan;
EXPLAIN (COSTS OFF) EXECUTE lion_ph(10000);
EXECUTE lion_ph(10000);
EXECUTE lion_ph(10001);
RESET plan_cache_mode;
DEALLOCATE lion_ph;

-- ---- shapes that must NOT push down ------------------------------------
-- count() of a column that neither the GROUP BY nor a WHERE clause
-- constrains, under a GROUP BY; without one it is pushed down as count(*)
-- AND n IS NOT NULL (DESIGN.md §10, "Shapes rewritten to the node's counts")
SELECT lion_pd('SELECT a, count(n) FROM lion_pdt WHERE a IN (3, 4) GROUP BY a');
SELECT lion_pd('SELECT count(n) FROM lion_pdt WHERE a = 3');
-- a HAVING that needs a subquery per group (a correlated SubPlan): the
-- grouping stays with core, though the subquery's own count is ours
SELECT * FROM lion_explain_norm('SELECT a, count(*) FROM lion_pdt GROUP BY a HAVING count(*) > (SELECT count(*) FROM lion_pdt x WHERE x.b = lion_pdt.a)') AS p("QUERY PLAN");
-- a HAVING without an aggregate is a WHERE by the time the planner asks us,
-- and since DESIGN.md §28 an inequality on the key bounds the group walk
-- (test/sql/range.sql)
SELECT lion_pd('SELECT a, count(*) FROM lion_pdt GROUP BY a HAVING a > 3');
-- an aggregate we cannot answer
SELECT lion_pd('SELECT sum(id) FROM lion_pdt WHERE a = 3');
-- ... while count(DISTINCT) is answered since DESIGN.md §26 (distinct.sql)
SELECT lion_pd('SELECT count(DISTINCT b) FROM lion_pdt WHERE a = 3');
-- a FILTER the index answers is one more set to count (countfilter.sql); one
-- it cannot answer leaves the query to the ordinary plan
SELECT lion_pd('SELECT count(*) FILTER (WHERE b = 2) FROM lion_pdt WHERE a = 3');
SELECT lion_pd('SELECT count(*) FILTER (WHERE id < 500) FROM lion_pdt WHERE a = 3');
-- a qual that is not an indexed equality to a constant: residual, tested on
-- the heap rows a = 3 selects (DESIGN.md §10, "Residual clauses")
SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE a = 3 AND id < 500');
-- ... which a range on an indexed column is, since DESIGN.md §28
SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE a > 3');
SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE a = 3 AND id = 5');
-- two different constants on one column
SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE a = 3 AND a = 4');
-- no equality key and no GROUP BY at all: the sum over every entry of the
-- lion column with the fewest, which is every row (DESIGN.md §35)
SELECT lion_pd('SELECT count(*) FROM lion_pdt');
-- GROUP BY an unindexed column
SELECT lion_pd('SELECT id, count(*) FROM lion_pdt WHERE a = 3 GROUP BY id');
-- three indexed columns are the decoded walk of DESIGN.md section 34
SELECT lion_pd('SELECT a, b, c, count(*) FROM lion_pdt GROUP BY a, b, c');
-- two indexed columns are the nested loop of DESIGN.md section 20 or the
-- decoded walk, whichever the model finds cheaper, below
SELECT lion_pd('SELECT a, b, count(*) FROM lion_pdt GROUP BY a, b');
-- GROUP BY an expression
SELECT lion_pd('SELECT a + 1, count(*) FROM lion_pdt GROUP BY a + 1');
-- a join
SELECT lion_pd('SELECT count(*) FROM lion_pdt x, lion_pdt y WHERE x.a = 3 AND y.b = 2');

/*
 * Applicable, but the cost model prefers the ordinary plan: a key the
 * statistics say matches nothing makes the bitmap plan look almost free.
 */
SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE a = 3 AND b = 999');
-- ORDER BY on the group column needs no Sort any more: the entries come out
-- in key order and the path says so (DESIGN.md §21).  `a` is NOT NULL, which
-- is what lets the pathkeys be NULLS LAST like the ORDER BY.
SELECT lion_pd('SELECT a, count(*) FROM lion_pdt GROUP BY a ORDER BY a');
SELECT lion_plans('SELECT a, count(*) FROM lion_pdt GROUP BY a ORDER BY a');

-- ---- the results themselves --------------------------------------------
SELECT a, count(*) FROM lion_pdt GROUP BY a ORDER BY a;
SELECT a, count(*) FROM lion_pdt WHERE b = 2 GROUP BY a ORDER BY a;
SELECT count(*) FROM lion_pdt WHERE a = 3 AND b = 2;
SELECT count(*) FROM lion_pdt WHERE a = 3 AND b = 2 AND c = 'c1';
SELECT count(*) FROM lion_pdt WHERE a = 999;
SELECT a, count(*) FROM lion_pdt WHERE a = 3 GROUP BY a;
SELECT a, b, count(*) FROM lion_pdt WHERE a = 3 GROUP BY a, b ORDER BY b;

-- ---- after a DELETE, before VACUUM (the heap recheck path) -------------
DELETE FROM lion_pdt WHERE id % 3 = 0;

SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE a = 3');
SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE a = 3 AND b = 2');
SELECT lion_pd('SELECT a, count(*) FROM lion_pdt GROUP BY a');
SELECT lion_pd('SELECT a, count(*) FROM lion_pdt WHERE b = 2 GROUP BY a');
SELECT a, count(*) FROM lion_pdt GROUP BY a ORDER BY a;

-- ---- and after VACUUM (the visibility-map path) ------------------------
VACUUM (FREEZE) lion_pdt;

SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE a = 3');
SELECT lion_pd('SELECT count(*) FROM lion_pdt WHERE a = 3 AND b = 2');
SELECT lion_pd('SELECT a, count(*) FROM lion_pdt GROUP BY a');
SELECT lion_pd('SELECT a, count(*) FROM lion_pdt WHERE b = 2 GROUP BY a');
SELECT a, count(*) FROM lion_pdt GROUP BY a ORDER BY a;

-- ---- a group whose rows have all been deleted disappears ---------------
DELETE FROM lion_pdt WHERE a = 5;
SELECT lion_pd('SELECT a, count(*) FROM lion_pdt GROUP BY a');
SELECT a, count(*) FROM lion_pdt GROUP BY a ORDER BY a;
VACUUM (FREEZE) lion_pdt;
SELECT lion_pd('SELECT a, count(*) FROM lion_pdt GROUP BY a');
SELECT a, count(*) FROM lion_pdt GROUP BY a ORDER BY a;

-- ---- an empty table ----------------------------------------------------
CREATE TABLE lion_pde (k int NOT NULL);
CREATE INDEX lion_pde_k ON lion_pde USING lion (k);
VACUUM (FREEZE, ANALYZE) lion_pde;
SELECT lion_pd('SELECT count(*) FROM lion_pde WHERE k = 1');
SELECT lion_pd('SELECT k, count(*) FROM lion_pde GROUP BY k');
SELECT count(*) FROM lion_pde WHERE k = 1;

-- ---- the node inside a larger plan --------------------------------------
SELECT lion_pd('SELECT x, c FROM (VALUES (1), (2)) v(x), LATERAL (SELECT count(*) c FROM lion_pdt WHERE a = 3) s');
-- an outer WHERE on the count becomes the subquery's HAVING, and pushes down with it
SELECT lion_pd('SELECT * FROM (SELECT a, count(*) AS c FROM lion_pdt GROUP BY a) s WHERE c > 0');

-- ---- what EXPLAIN ANALYZE reports ---------------------------------------
VACUUM (FREEZE) lion_pdt;
SELECT lion_pd_counters('SELECT count(*) FROM lion_pdt WHERE a = 3');
-- the pages a DELETE dirties are no longer all-visible, so their TIDs are
-- fetched from the heap and counted as rechecked blocks
DELETE FROM lion_pdt WHERE id % 500 = 0;
SELECT lion_pd_counters('SELECT count(*) FROM lion_pdt WHERE a = 3');

-- ---- a grouping over a heap that is not all-visible ---------------------
/*
 * A thousand groups whose rows are spread over every heap page, on a heap the
 * visibility map cannot vouch for.  Every TID has to be resolved against the
 * snapshot - but each dirty BLOCK is fetched only once per query, because the
 * per-query visibility cache answers every later group from memory
 * (DESIGN.md §9), so the node makes one pass over the heap and not one per
 * group.  Before the cache, and before the estimate stopped charging
 * numgroups random reads per dirty page, this was the case that had to lose
 * by far.  On the release build it is a tie - the node 14.8 ms of CPU, the
 * sequential scan and hash aggregate 15.6 - and the model gives it to the
 * hash aggregate: over rows this wide the scan's pages are most of core's
 * plan, and only its hashing is priced below its time (DESIGN.md §39, "A
 * hashed aggregate's own rate").  Once the pages are all-visible again the
 * node wins by more, and is chosen.
 */
CREATE TABLE lion_pdd (g int NOT NULL, pad text NOT NULL);
INSERT INTO lion_pdd
SELECT i % 1000, repeat('x', 200) FROM generate_series(1, 100000) i;
CREATE INDEX lion_pdd_g ON lion_pdd USING lion (g);
ANALYZE lion_pdd;			-- no VACUUM: relallvisible stays 0
SELECT lion_pd('SELECT g, count(*) FROM lion_pdd GROUP BY g');
-- and once it is all-visible there is nothing left to recheck at all
VACUUM (FREEZE, ANALYZE) lion_pdd;
SELECT lion_pd('SELECT g, count(*) FROM lion_pdd GROUP BY g');

-- ---- the cost model still refuses a grouping it cannot win -------------
/*
 * The other end of the same model.  Twenty thousand groups of five rows over
 * a hundred thousand rows: the recheck is not what costs here, the per-group
 * work is - one entry lookup and one container per group - and it is more
 * than the sequential scan plus a HashAggregate does.  The planner must
 * refuse, and measurably should: 28.0 ms for the sequential aggregate
 * against 40.8 ms for the node when sequential scans are discouraged
 * (2026-09-21).  Nothing is disabled here; which plan the cost model picks IS
 * the test.
 */
CREATE TABLE lion_pdh (g int NOT NULL, pad text NOT NULL);
INSERT INTO lion_pdh
SELECT i % 20000, repeat('x', 200) FROM generate_series(1, 100000) i;
CREATE INDEX lion_pdh_g ON lion_pdh USING lion (g);
ANALYZE lion_pdh;
SELECT lion_pd('SELECT g, count(*) FROM lion_pdh GROUP BY g');
SELECT lion_plans('SELECT g, count(*) FROM lion_pdh GROUP BY g');

-- ---- a grouping over a heap a few per cent of which is dirty -----------
/*
 * The case the 2026-09-21 follow-up review measured: a vacuumed table, then
 * five per cent of its rows updated, which leaves about a tenth of its heap
 * pages unable to be vouched for - the ordinary state of a large table that
 * is mostly read and occasionally written.  A GROUP BY returns to those
 * pages for every group, but the per-query visibility cache fetches each of
 * them once (DESIGN.md §9), so the estimate must charge one pass over the
 * dirty working set and not numgroups random reads per page.  Charging the
 * latter asked 1.8M cost units against the sequential aggregate's 175k on
 * five million rows and lost a query the node wins by 7.9x; here the node
 * runs in 2.0 ms against 15.8 ms (2026-09-21).  Nothing is disabled: which
 * plan the cost model picks IS the test.
 */
CREATE TABLE lion_pdg (id int NOT NULL, k int NOT NULL, pad text NOT NULL);
INSERT INTO lion_pdg
SELECT i, i % 20, repeat('x', 200) FROM generate_series(1, 100000) i;
CREATE INDEX lion_pdg_k ON lion_pdg USING lion (k);
VACUUM (FREEZE, ANALYZE) lion_pdg;
UPDATE lion_pdg SET pad = pad || 'y' WHERE id <= 5000;
ANALYZE lion_pdg;			-- no VACUUM: the updated pages stay dirty
SELECT relallvisible > 0 AND relallvisible < relpages AS mostly_all_visible
  FROM pg_class WHERE relname = 'lion_pdg';
SELECT lion_pd('SELECT k, count(*) FROM lion_pdg GROUP BY k');
SELECT lion_plans('SELECT k, count(*) FROM lion_pdg GROUP BY k');
/*
 * ... and this is the cache the estimate is allowed to assume: every group
 * comes back to the same dirty pages, and every visit after the first is
 * answered out of memory instead of pinning the page again.  A single count
 * walks the result in TID order and never returns to a block, so it has no
 * cache hits at all (the calls further up).
 */
SELECT lion_pd_counters('SELECT k, count(*) FROM lion_pdg GROUP BY k');

-- ---- a long IN list, priced per element ---------------------------------
/*
 * Every element of an IN list is a directory lookup and a container chain of
 * its own, and the estimate has to say so: charging a list as one page made
 * a thousand-element list look four times cheaper than it is (the 2026-09-21
 * follow-up review).  What it is NOT charged for any more is a k-way union per
 * element, because since the disjoint-sum short-circuit of DESIGN.md §15 there
 * may be no union at all - and when there is one, it costs a heap sift per
 * CONTAINER and a bitset image per container key, not log2(k) comparisons per
 * row.
 *
 * So all three lengths belong to the node here, and the thousand-value case is
 * the one that says the merge term is no longer wild: 4.3 ms against the
 * B-tree index-only scan's 13.4 on this table (2026-09-21, assert build), where
 * the old estimate refused it.
 */
CREATE TABLE lion_pdi (id int NOT NULL, k int NOT NULL);
INSERT INTO lion_pdi SELECT i, i % 1000 FROM generate_series(1, 200000) i;
CREATE INDEX lion_pdi_r ON lion_pdi USING lion (k);
CREATE INDEX lion_pdi_b ON lion_pdi (k);
VACUUM (FREEZE, ANALYZE) lion_pdi;
/* The list is generated rather than written out, so that the plan text this
 * reports stays short; lion_pd() prints the choice and the row count only. */
CREATE OR REPLACE FUNCTION lion_pd_in(n int) RETURNS text
LANGUAGE plpgsql AS $$
BEGIN
	RETURN lion_pd(format('SELECT count(*) FROM lion_pdi WHERE k IN (%s)',
						 (SELECT string_agg(g::text, ',')
							FROM generate_series(0, n - 1) g)));
END $$;
SELECT lion_pd_in(3);
SELECT lion_pd_in(10);
SELECT lion_pd_in(1000);
DROP FUNCTION lion_pd_in(int);


-- ---- a nearly all-visible heap is what the pushdown is for --------------
/*
 * The other side of the same model: 200k rows vacuumed, then a few thousand
 * more inserted, so that relallvisible is just short of relpages - the
 * ordinary state of a large table that is mostly read.  The rechecks can only
 * touch the pages the visibility map cannot vouch for, and the estimate has
 * to say so; charging random reads across the whole heap made the node lose
 * to plans doing far more work (the 2026-09-20 review, finding 5).  Nothing
 * is disabled here: which plan the cost model picks IS the test.
 */
CREATE TABLE lion_pdw (id int NOT NULL, k int NOT NULL);
INSERT INTO lion_pdw SELECT i, i % 10 FROM generate_series(1, 200000) i;
CREATE INDEX lion_pdw_k ON lion_pdw USING lion (k);
VACUUM (FREEZE, ANALYZE) lion_pdw;
INSERT INTO lion_pdw SELECT 200000 + i, i % 10 FROM generate_series(1, 4000) i;
ANALYZE lion_pdw;
SELECT relallvisible > 0 AND relallvisible < relpages AS nearly_all_visible
  FROM pg_class WHERE relname = 'lion_pdw';
SELECT lion_pd('SELECT count(*) FROM lion_pdw WHERE k = 3');
SELECT lion_plans('SELECT count(*) FROM lion_pdw WHERE k = 3');
SELECT lion_pd('SELECT k, count(*) FROM lion_pdw GROUP BY k');
SELECT lion_plans('SELECT k, count(*) FROM lion_pdw GROUP BY k');

-- ---- an opclass whose equality is not the grouping equality -------------
/*
 * An operator class may define a coarser equality than the type's own, and
 * this one does: it calls two strings equal when they differ only in case.
 * It is a valid roaring opclass - its hash agrees with its equality - but its
 * entries are already-merged groups, and no GROUP BY may be answered from
 * them, not even a count-only one whose key never reaches the output: the
 * counts themselves would belong to the wrong groups (the 2026-09-20 review,
 * finding 3).
 */
CREATE FUNCTION lion_lower_eq(text, text) RETURNS boolean
	LANGUAGE sql IMMUTABLE STRICT PARALLEL SAFE
	AS $$ SELECT lower($1) = lower($2) $$;
CREATE OPERATOR === (LEFTARG = text, RIGHTARG = text, FUNCTION = lion_lower_eq,
					 COMMUTATOR = ===);
CREATE FUNCTION lion_lower_hash(text) RETURNS integer
	LANGUAGE sql IMMUTABLE STRICT PARALLEL SAFE
	AS $$ SELECT hashtext(lower($1)) $$;
CREATE OPERATOR CLASS lion_lower_ops FOR TYPE text USING lion AS
	OPERATOR 1 === (text, text),
	FUNCTION 1 lion_lower_hash(text);

CREATE TABLE lion_pdc (v text NOT NULL);
INSERT INTO lion_pdc SELECT 'A' FROM generate_series(1, 50000);
INSERT INTO lion_pdc SELECT 'a' FROM generate_series(1, 50000);
CREATE INDEX lion_pdc_v ON lion_pdc USING lion (v lion_lower_ops);
VACUUM (FREEZE, ANALYZE) lion_pdc;
-- one entry for the two spellings, which is what the opclass says
SELECT entries FROM lion_index_stats('lion_pdc_v');
SELECT lion_pd('SELECT v, count(*) FROM lion_pdc GROUP BY v');
SELECT lion_plans('SELECT v, count(*) FROM lion_pdc GROUP BY v');
SELECT lion_pd('SELECT count(*) FROM lion_pdc GROUP BY v');
SELECT v, count(*) FROM lion_pdc GROUP BY v ORDER BY v;
/*
 * The very same data under the default text opclass does drive the node, so
 * what the queries above turned down is the opclass and not the table or the
 * cost model.
 */
CREATE TABLE lion_pdv (v text NOT NULL);
INSERT INTO lion_pdv SELECT 'A' FROM generate_series(1, 50000);
INSERT INTO lion_pdv SELECT 'a' FROM generate_series(1, 50000);
CREATE INDEX lion_pdv_v ON lion_pdv USING lion (v);
VACUUM (FREEZE, ANALYZE) lion_pdv;
SELECT entries FROM lion_index_stats('lion_pdv_v');
SELECT lion_pd('SELECT v, count(*) FROM lion_pdv GROUP BY v');
SELECT lion_plans('SELECT v, count(*) FROM lion_pdv GROUP BY v');
SELECT v, count(*) FROM lion_pdv GROUP BY v ORDER BY v;

-- ---- a type whose equality does not preserve the representation ---------
/*
 * numeric 1.0 and 1.00 are equal and share one posting-set entry, but they
 * are different strings, and the entry keeps whichever of them was indexed
 * first.  Printing that key could produce a value no visible row holds, so
 * every value-producing shape steps aside; counting the class is exact
 * either way and is still pushed down (the 2026-09-20 review, finding 4).
 */
CREATE TABLE lion_pdn (v numeric NOT NULL);
INSERT INTO lion_pdn SELECT 1.0 FROM generate_series(1, 20000);
INSERT INTO lion_pdn SELECT 1.00 FROM generate_series(1, 20000);
INSERT INTO lion_pdn SELECT 2.5 FROM generate_series(1, 20000);
CREATE INDEX lion_pdn_v ON lion_pdn USING lion (v);
VACUUM (FREEZE, ANALYZE) lion_pdn;
SELECT entries FROM lion_index_stats('lion_pdn_v');
SELECT lion_pd('SELECT v, count(*) FROM lion_pdn GROUP BY v');
SELECT lion_pd('SELECT v, count(*) FROM lion_pdn WHERE v = 1.000 GROUP BY v');
/*
 * With every other plan disabled the node would be chosen if the planner had
 * built it at all, which makes the two answers here the plan-time rule and
 * not the cost model: no node for the grouping, and the count as before.
 */
SET enable_seqscan = off;
SET enable_bitmapscan = off;
SET pg_lion.enable_plain_scan = off;
-- (and no Sort, so that core's own plan for the refused grouping is the same
-- whether disabled paths are counted, as in 18, or priced, as in 16 and 17)
SET enable_sort = off;
SELECT lion_plans('SELECT v, count(*) FROM lion_pdn GROUP BY v');
RESET enable_sort;
SELECT lion_plans('SELECT count(*) FROM lion_pdn WHERE v = 1.000');
SELECT lion_pd('SELECT count(*) FROM lion_pdn WHERE v = 1.000');
RESET enable_seqscan;
RESET enable_bitmapscan;
RESET pg_lion.enable_plain_scan;
SELECT v, count(*) FROM lion_pdn GROUP BY v ORDER BY v;

-- ---- bpchar: an equalimage function that is not enough -----------------
/*
 * bpchar registers btvarstrequalimage, which says yes under a deterministic
 * collation, yet 'a   ' = 'a' and the two are different values: unconstrained
 * bpchar keeps the blanks it was given, and bpchar's output prints them.  The
 * entry kept the spelling of a row that has since been deleted, and printing
 * it leaked that row's value (the 2026-09-23 review).  lion_pd() cannot see
 * the difference - EXCEPT ALL compares with bpchareq - so the value is printed
 * through format() under both settings.  Counting is still pushed down.
 */
CREATE TABLE lion_pdb (c bpchar, k int);
INSERT INTO lion_pdb VALUES ('a   ', 1);
CREATE INDEX lion_pdb_c ON lion_pdb USING lion (c);
CREATE INDEX lion_pdb_k ON lion_pdb USING lion (k);
DELETE FROM lion_pdb;
INSERT INTO lion_pdb VALUES ('a', 1);
ANALYZE lion_pdb;
SET enable_seqscan = off;
SET enable_bitmapscan = off;
SET pg_lion.enable_plain_scan = off;
SELECT lion_pd('SELECT c, count(*) FROM lion_pdb GROUP BY c');
SELECT lion_pd('SELECT c, count(*) FROM lion_pdb WHERE c = ''a'' GROUP BY c');
SELECT lion_pd('SELECT k, c, count(*) FROM lion_pdb WHERE c = ''a'' GROUP BY k, c');
SELECT lion_pd('SELECT count(*) FROM lion_pdb WHERE c = ''a''');
SELECT lion_pd('SELECT k, count(*) FROM lion_pdb WHERE c = ''a'' GROUP BY k');
SET pg_lion.enable_count_pushdown = on;
SELECT format('[%s]', c) AS c, n FROM (SELECT c, count(*) AS n FROM lion_pdb GROUP BY c) s;
SELECT format('[%s]', c) AS c, n FROM (SELECT c, count(*) AS n FROM lion_pdb WHERE c = 'a' GROUP BY c) s;
SELECT k, format('[%s]', c) AS c, n FROM (SELECT k, c, count(*) AS n FROM lion_pdb WHERE c = 'a' GROUP BY k, c) s;
SET pg_lion.enable_count_pushdown = off;
SELECT format('[%s]', c) AS c, n FROM (SELECT c, count(*) AS n FROM lion_pdb GROUP BY c) s;
RESET pg_lion.enable_count_pushdown;
/*
 * The other core types whose equality is coarser than their bytes - float8
 * (-0 and 0), interval ('1 day' and '24 hours'), numeric above - register no
 * equalimage function at all and were refused already; they stay refused.
 */
CREATE TABLE lion_pdf (f float8, i interval);
INSERT INTO lion_pdf VALUES ('-0', '1 day');
CREATE INDEX lion_pdf_f ON lion_pdf USING lion (f);
CREATE INDEX lion_pdf_i ON lion_pdf USING lion (i);
DELETE FROM lion_pdf;
INSERT INTO lion_pdf VALUES ('0', '24 hours');
ANALYZE lion_pdf;
SELECT lion_pd('SELECT f, count(*) FROM lion_pdf GROUP BY f');
SELECT lion_pd('SELECT i, count(*) FROM lion_pdf GROUP BY i');
SELECT f, i FROM (SELECT f, count(*) FROM lion_pdf GROUP BY f) a,
	(SELECT i, count(*) FROM lion_pdf GROUP BY i) b;
RESET enable_seqscan;
RESET enable_bitmapscan;
RESET pg_lion.enable_plain_scan;

-- ---- an unpopulated materialized view --------------------------------
/*
 * Core refuses to scan a materialized view created WITH NO DATA
 * (ExecOpenScanRelation()), and the node has to as well rather than count
 * the empty heap: every shape errors the same way, EXPLAIN without ANALYZE
 * and CREATE TABLE AS ... WITH NO DATA do not, and a REFRESH brings the
 * pushdown back.
 */
CREATE MATERIALIZED VIEW lion_pdm AS
	SELECT g % 3 AS k, g % 5 AS j FROM generate_series(1, 10) g WITH NO DATA;
CREATE INDEX lion_pdm_k ON lion_pdm USING lion (k);
CREATE INDEX lion_pdm_j ON lion_pdm USING lion (j);
SET enable_seqscan = off;
SET enable_bitmapscan = off;
SET pg_lion.enable_plain_scan = off;
EXPLAIN (COSTS OFF) SELECT k, count(*) FROM lion_pdm GROUP BY k;
SELECT k, count(*) FROM lion_pdm GROUP BY k;
SELECT count(*) FROM lion_pdm WHERE k = 1;
SELECT count(*) FROM lion_pdm WHERE k IS NOT NULL;
SELECT count(DISTINCT k) FROM lion_pdm;
SELECT j, count(DISTINCT k) FROM lion_pdm GROUP BY j;
SET pg_lion.enable_count_pushdown = off;
SELECT k, count(*) FROM lion_pdm GROUP BY k;
RESET pg_lion.enable_count_pushdown;
CREATE TEMP TABLE lion_pdm_nd AS SELECT k, count(*) FROM lion_pdm GROUP BY k WITH NO DATA;
DROP TABLE lion_pdm_nd;
REFRESH MATERIALIZED VIEW lion_pdm;
SELECT lion_pd('SELECT k, count(*) FROM lion_pdm GROUP BY k');
SELECT lion_pd('SELECT count(*) FROM lion_pdm WHERE k = 1');
SELECT lion_pd('SELECT j, count(DISTINCT k) FROM lion_pdm GROUP BY j');
RESET enable_seqscan;
RESET enable_bitmapscan;
RESET pg_lion.enable_plain_scan;
DROP MATERIALIZED VIEW lion_pdm;
DROP TABLE lion_pdf;
DROP TABLE lion_pdb;

-- ---- prepared statements: a parameter where a literal may stand -------
/*
 * A GENERIC plan keeps `a = $1` as a Param, and the pushdown used to accept
 * nothing but a literal: the count of a dense key then fell back to the
 * ordinary plan and lost the whole point of the node (a measured 5.6 ms
 * sequential scan against 0.025 ms, and 207 ms against 3.1 ms on five
 * million rows - the 2026-09-21 follow-up review).  A Param is now accepted
 * wherever a Const is, for equality and for the array of an IN list, and the
 * node evaluates it through its own ExprContext at the start of every scan
 * (DESIGN.md §10).  EXPLAIN prints it as `$1`.
 *
 * lion_pd_prep() prepares one query, forces a generic plan so that every
 * parameter really stays a Param, and proves the answer is the one the same
 * query gives with the pushdown switched off - as a multiset, both ways
 * round.
 */
CREATE OR REPLACE FUNCTION lion_pd_prep(q text, args text) RETURNS text
LANGUAGE plpgsql AS $$
DECLARE
	ln text;
	rec record;
	pushed boolean := false;
	onrows text[] := '{}';
	offrows text[] := '{}';
	ndiff bigint;
BEGIN
	PERFORM set_config('plan_cache_mode', 'force_generic_plan', true);

	PERFORM set_config('pg_lion.enable_count_pushdown', 'on', true);
	EXECUTE 'PREPARE lion_pp_on AS ' || q;
	FOR ln IN EXECUTE 'EXPLAIN (COSTS OFF) EXECUTE lion_pp_on(' || args || ')' LOOP
		IF ln LIKE '%Custom Scan (LionCount)%' THEN
			pushed := true;
		END IF;
	END LOOP;
	FOR rec IN EXECUTE 'EXECUTE lion_pp_on(' || args || ')' LOOP
		onrows := onrows || rec::text;
	END LOOP;

	PERFORM set_config('pg_lion.enable_count_pushdown', 'off', true);
	EXECUTE 'PREPARE lion_pp_off AS ' || q;
	FOR rec IN EXECUTE 'EXECUTE lion_pp_off(' || args || ')' LOOP
		offrows := offrows || rec::text;
	END LOOP;

	PERFORM set_config('pg_lion.enable_count_pushdown', 'on', true);
	EXECUTE 'DEALLOCATE lion_pp_on';
	EXECUTE 'DEALLOCATE lion_pp_off';

	SELECT (SELECT count(*) FROM (SELECT unnest(onrows)
								  EXCEPT ALL SELECT unnest(offrows)) a)
		 + (SELECT count(*) FROM (SELECT unnest(offrows)
								  EXCEPT ALL SELECT unnest(onrows)) b)
	  INTO ndiff;
	IF ndiff <> 0 THEN
		RETURN format('MISMATCH: %s rows differ', ndiff);
	END IF;

	RETURN format('%s, %s rows',
				  CASE WHEN pushed THEN 'pushed down' ELSE 'not pushed down' END,
				  coalesce(array_length(onrows, 1), 0));
END $$;

SELECT lion_pd_prep('SELECT count(*) FROM lion_pdt WHERE a = $1', '3');
SELECT lion_pd_prep('SELECT count(*) FROM lion_pdt WHERE a = $1', '-1');
-- a NULL parameter means zero rows, which is not the same as one group of 0
SELECT lion_pd_prep('SELECT count(*) FROM lion_pdt WHERE a = $1', 'NULL::int');
SELECT lion_pd_prep('SELECT count(*) FROM lion_pdt WHERE a = ANY ($1)',
				   'ARRAY[1,3]');
SELECT lion_pd_prep('SELECT count(*) FROM lion_pdt WHERE a = ANY ($1)',
				   'NULL::int[]');
-- `a IN ($1, $2)` keeps an ARRAY[] of Params in a generic plan
SELECT lion_pd_prep('SELECT count(*) FROM lion_pdt WHERE a IN ($1, $2)', '1, 3');
SELECT lion_pd_prep('SELECT b, count(*) FROM lion_pdt WHERE a = $1'
				   ' GROUP BY b ORDER BY b', '3');
-- the pinned column is printed from the entry's stored key, parameter or not
SELECT lion_pd_prep('SELECT a, count(*) FROM lion_pdt WHERE a = $1 GROUP BY a',
				   '3');
SELECT lion_pd_prep('SELECT count(*) FROM lion_pdt WHERE a = $1 AND n IS NULL',
				   '3');

SET plan_cache_mode = force_generic_plan;
PREPARE lion_pp(int) AS SELECT count(*) FROM lion_pdt WHERE a = $1;
EXPLAIN (COSTS OFF) EXECUTE lion_pp(3);
EXECUTE lion_pp(3);
EXECUTE lion_pp(4);
DEALLOCATE lion_pp;
RESET plan_cache_mode;

/*
 * An exec Param, which a nested loop changes between rescans: the node has to
 * re-evaluate it every time rather than count the first outer row's key
 * again.
 */
EXPLAIN (COSTS OFF)
SELECT v.k, s.c FROM (VALUES (1), (3), (-1)) v(k),
	 LATERAL (SELECT count(*) AS c FROM lion_pdt WHERE a = v.k) s;
SELECT v.k, s.c FROM (VALUES (1), (3), (-1)) v(k),
	 LATERAL (SELECT count(*) AS c FROM lion_pdt WHERE a = v.k) s ORDER BY v.k;
SET pg_lion.enable_count_pushdown = off;
SELECT v.k, s.c FROM (VALUES (1), (3), (-1)) v(k),
	 LATERAL (SELECT count(*) AS c FROM lion_pdt WHERE a = v.k) s ORDER BY v.k;
RESET pg_lion.enable_count_pushdown;

-- ---- OR across columns (DESIGN.md section 19) ---------------------------
/*
 * `count(*) WHERE a = 17 OR b = 3` used to be declined and run as BitmapOr
 * over a Bitmap Heap Scan, which at a few per cent selectivity touches nearly
 * every heap page.  Now the whole restriction is ONE source - the union of
 * its arms - which is ANDed with the other clauses and with a GROUP BY driver
 * like any other.  Measured on a million rows of (c200, c20) on this
 * assert-enabled build: 0.6 ms for the node against 18-30 ms for BitmapOr +
 * heap (2026-09-21).
 *
 * Every arm has to be a positive clause the posting sets can answer, or an
 * AND of such clauses, each on a column of the same relation with an index of
 * its own; a negated arm is declined, because the complement of a posting set
 * is not a posting set and under a union there is nothing to subtract it
 * from.
 */
CREATE TABLE lion_pdo (id int NOT NULL, a int NOT NULL, b int NOT NULL,
					   c text NOT NULL, n int);
INSERT INTO lion_pdo
SELECT i, i % 200, i % 20, 'c' || (i % 4),
	   CASE WHEN i % 101 = 0 THEN NULL ELSE i % 5 END
  FROM generate_series(1, 200000) i;
CREATE INDEX lion_pdo_a ON lion_pdo USING lion (a);
CREATE INDEX lion_pdo_b ON lion_pdo USING lion (b);
CREATE INDEX lion_pdo_c ON lion_pdo USING lion (c);
CREATE INDEX lion_pdo_n ON lion_pdo USING lion (n);
VACUUM (FREEZE, ANALYZE) lion_pdo;

SELECT lion_plans('SELECT count(*) FROM lion_pdo WHERE a = 17 OR b = 3');
SELECT lion_pd('SELECT count(*) FROM lion_pdo WHERE a = 17 OR b = 3');
-- three arms
SELECT lion_pd($$SELECT count(*) FROM lion_pdo WHERE a = 17 OR b = 3 OR c = 'c1'$$);
-- an arm that is an AND
SELECT lion_pd($$SELECT count(*) FROM lion_pdo WHERE (a = 17 AND b = 3) OR c = 'c1'$$);
SELECT lion_pd($$SELECT count(*) FROM lion_pdo WHERE (a = 17 AND b = 3) OR (c = 'c1' AND n = 2)$$);
SELECT lion_plans($$SELECT count(*) FROM lion_pdo WHERE (a = 17 AND b = 3) OR (c = 'c1' AND n = 2)$$);
-- an OR next to a plain AND clause, and under a GROUP BY
SELECT lion_pd($$SELECT count(*) FROM lion_pdo WHERE (a = 17 OR b = 3) AND c = 'c1'$$);
SELECT lion_plans($$SELECT count(*) FROM lion_pdo WHERE (a = 17 OR b = 3) AND c = 'c1'$$);
SELECT lion_pd('SELECT c, count(*) FROM lion_pdo WHERE a = 17 OR b = 3 GROUP BY c');
SELECT lion_plans('SELECT c, count(*) FROM lion_pdo WHERE a = 17 OR b = 3 GROUP BY c');
-- two OR restrictions in one query are two sources
SELECT lion_pd($$SELECT count(*) FROM lion_pdo WHERE (a = 17 OR b = 3) AND (c = 'c1' OR n = 2)$$);
SELECT lion_plans($$SELECT count(*) FROM lion_pdo WHERE (a = 17 OR b = 3) AND (c = 'c1' OR n = 2)$$);
-- IN and IS NULL arms
SELECT lion_pd('SELECT count(*) FROM lion_pdo WHERE a IN (17, 18) OR n IS NULL');
SELECT lion_pd('SELECT count(*) FROM lion_pdo WHERE a = 17 OR b IN (3, 4) OR n IS NULL');
SELECT lion_plans('SELECT count(*) FROM lion_pdo WHERE a IN (17, 18) OR n IS NULL');
-- an arm whose value has no entry, and an OR none of whose arms has one
SELECT lion_pd('SELECT count(*) FROM lion_pdo WHERE a = 999 OR b = 3');
SELECT lion_pd('SELECT count(*) FROM lion_pdo WHERE a = 999 OR b = 998');
-- the results themselves
SELECT count(*) FROM lion_pdo WHERE a = 17 OR b = 3;
SELECT c, count(*) FROM lion_pdo WHERE a = 17 OR b = 3 GROUP BY c ORDER BY c;

-- ---- ORs the node must decline -----------------------------------------
-- an arm of negated clauses is not declined: it is a complement, counted by
-- De Morgan (DESIGN.md §35, "Complements in an OR")
SELECT lion_pd('SELECT count(*) FROM lion_pdo WHERE a = 17 OR n IS NOT NULL');
SELECT lion_pd('SELECT count(*) FROM lion_pdo WHERE a = 17 OR NOT (b = 3)');
SELECT lion_plans('SELECT count(*) FROM lion_pdo WHERE a = 17 OR n IS NOT NULL');
-- an arm on a column with no lion index
SELECT lion_pd('SELECT count(*) FROM lion_pdo WHERE a = 17 OR id = 5');
-- an arm the posting sets cannot answer without a recheck
SELECT lion_pd('SELECT count(*) FROM lion_pdo WHERE a = 17 OR b > 3');
-- one unusable leaf spoils its AND arm and with it the whole restriction
SELECT lion_pd('SELECT count(*) FROM lion_pdo WHERE (a = 17 AND id = 5) OR b = 3');
/*
 * A column constrained only INSIDE an OR is constrained in no row of the
 * result: the other arm selects rows it says nothing about.  So it makes no
 * count(col) answerable, which is what these two check - the first would come
 * out as count(*) and the second as 0 if an OR leaf were treated like a
 * top-level clause.  Both are answered as count(*) AND n IS NOT NULL
 * (DESIGN.md §10, "Shapes rewritten to the node's counts"), which the
 * comparison with the ordinary plan checks.
 */
SELECT lion_pd('SELECT count(n) FROM lion_pdo WHERE n = 2 OR a = 17');
SELECT lion_pd('SELECT count(n) FROM lion_pdo WHERE n IS NULL OR a = 17');

-- ---- parameters inside the arms ----------------------------------------
/*
 * A Param stands wherever a Const may (DESIGN.md section 10), inside an arm
 * as anywhere else, and an arm whose parameter comes out NULL selects nothing
 * - the OR of the remaining arms is still the answer.
 */
SELECT lion_pd_prep('SELECT count(*) FROM lion_pdo WHERE a = $1 OR b = $2',
				   '17, 3');
SELECT lion_pd_prep('SELECT count(*) FROM lion_pdo WHERE a = $1 OR b = $2',
				   'NULL::int, 3');
SELECT lion_pd_prep('SELECT count(*) FROM lion_pdo WHERE a = $1 OR b = $2',
				   '17, NULL::int');
SELECT lion_pd_prep('SELECT count(*) FROM lion_pdo WHERE a = $1 OR b = $2',
				   'NULL::int, NULL::int');
SELECT lion_pd_prep('SELECT count(*) FROM lion_pdo WHERE a = ANY ($1) OR n IS NULL',
				   'ARRAY[17,18]');
SELECT lion_pd_prep('SELECT c, count(*) FROM lion_pdo WHERE a = $1 OR b = $2'
				   ' GROUP BY c ORDER BY c', '17, 3');

SET plan_cache_mode = force_generic_plan;
PREPARE lion_ppo(int, int) AS
	SELECT count(*) FROM lion_pdo WHERE a = $1 OR b = $2;
EXPLAIN (COSTS OFF) EXECUTE lion_ppo(17, 3);
EXECUTE lion_ppo(17, 3);
EXECUTE lion_ppo(NULL, 3);
EXECUTE lion_ppo(NULL, NULL);
DEALLOCATE lion_ppo;
RESET plan_cache_mode;

-- an exec Param that changes between rescans
EXPLAIN (COSTS OFF)
SELECT v.k, s.c FROM (VALUES (17), (3), (-1)) v(k),
	 LATERAL (SELECT count(*) AS c FROM lion_pdo WHERE a = v.k OR b = v.k) s;
SELECT v.k, s.c FROM (VALUES (17), (3), (-1)) v(k),
	 LATERAL (SELECT count(*) AS c FROM lion_pdo WHERE a = v.k OR b = v.k) s
	 ORDER BY v.k;
SET pg_lion.enable_count_pushdown = off;
SELECT v.k, s.c FROM (VALUES (17), (3), (-1)) v(k),
	 LATERAL (SELECT count(*) AS c FROM lion_pdo WHERE a = v.k OR b = v.k) s
	 ORDER BY v.k;
RESET pg_lion.enable_count_pushdown;

-- ---- a dirty heap, then an all-visible one ------------------------------
/*
 * Under a union a dead TID may be contributed by a single leaf, so every leaf
 * of an OR is read the pinned way until the merged container has been through
 * the visibility map (DESIGN.md section 19) and the TIDs the map cannot vouch
 * for are resolved against the snapshot like any other.
 */
DELETE FROM lion_pdo WHERE id % 7 = 0;
SELECT lion_pd('SELECT count(*) FROM lion_pdo WHERE a = 17 OR b = 3');
SELECT lion_pd('SELECT c, count(*) FROM lion_pdo WHERE a = 17 OR b = 3 GROUP BY c');
SELECT lion_pd_counters('SELECT count(*) FROM lion_pdo WHERE a = 17 OR b = 3');
VACUUM (FREEZE) lion_pdo;
SELECT lion_pd('SELECT count(*) FROM lion_pdo WHERE a = 17 OR b = 3');
SELECT lion_pd('SELECT c, count(*) FROM lion_pdo WHERE a = 17 OR b = 3 GROUP BY c');
SELECT lion_pd_counters('SELECT count(*) FROM lion_pdo WHERE a = 17 OR b = 3');
SELECT c, count(*) FROM lion_pdo WHERE a = 17 OR b = 3 GROUP BY c ORDER BY c;

-- ---- a multi-key arm (DESIGN.md section 17) -----------------------------
CREATE TABLE lion_pdmo (id int NOT NULL, k int NOT NULL, tags text[]);
INSERT INTO lion_pdmo
SELECT i, i % 50, ARRAY['t' || (i % 13), 't' || (i % 29)]
  FROM generate_series(1, 50000) i;
CREATE INDEX lion_pdmo_k ON lion_pdmo USING lion (k);
CREATE INDEX lion_pdmo_tags ON lion_pdmo USING lion (tags);
VACUUM (FREEZE, ANALYZE) lion_pdmo;
SELECT lion_plans($$SELECT count(*) FROM lion_pdmo WHERE tags @> '{t5}' OR k = 7$$);
SELECT lion_pd($$SELECT count(*) FROM lion_pdmo WHERE tags @> '{t5}' OR k = 7$$);
SELECT lion_pd($$SELECT count(*) FROM lion_pdmo WHERE tags @> '{t5,t7}' OR k = 7$$);
SELECT lion_pd($$SELECT count(*) FROM lion_pdmo WHERE tags && '{t5,t7}' OR k = 7$$);
-- an ALL-mode query, and a parameter that has no shape at plan time: declined
SELECT lion_pd($$SELECT count(*) FROM lion_pdmo WHERE tags @> '{}' OR k = 7$$);
SELECT lion_pd_prep($$SELECT count(*) FROM lion_pdmo WHERE tags @> $1 OR k = 7$$,
				   $$'{t5}'::text[]$$);

-- ---- a partitioned table (DESIGN.md section 16) -------------------------
CREATE TABLE lion_pdpo (id int NOT NULL, a int NOT NULL, b int NOT NULL)
	PARTITION BY RANGE (id);
CREATE TABLE lion_pdpo1 PARTITION OF lion_pdpo FOR VALUES FROM (0) TO (50000);
CREATE TABLE lion_pdpo2 PARTITION OF lion_pdpo FOR VALUES FROM (50000) TO (100000);
INSERT INTO lion_pdpo SELECT i, i % 200, i % 20 FROM generate_series(0, 99999) i;
CREATE INDEX ON lion_pdpo1 USING lion (a);
CREATE INDEX ON lion_pdpo1 USING lion (b);
CREATE INDEX ON lion_pdpo2 USING lion (a);
CREATE INDEX ON lion_pdpo2 USING lion (b);
VACUUM (FREEZE, ANALYZE) lion_pdpo;
SELECT lion_plans('SELECT count(*) FROM lion_pdpo WHERE a = 17 OR b = 3');
SELECT lion_pd('SELECT count(*) FROM lion_pdpo WHERE a = 17 OR b = 3');
SELECT lion_plans('SELECT a, count(*) FROM lion_pdpo WHERE a = 17 OR b = 3 GROUP BY a');
SELECT lion_pd('SELECT a, count(*) FROM lion_pdpo WHERE a = 17 OR b = 3 GROUP BY a');
SELECT count(*) FROM lion_pdpo WHERE a = 17 OR b = 3;

-- ---- GROUP BY two indexed columns (DESIGN.md section 20) ----------------
/*
 * A nested loop over the two indexes' entries: the column with FEWER distinct
 * values drives the scan (it is the outer one, and EXPLAIN names it first),
 * the other index's keys are read once per relation, and each (outer, inner)
 * pair is counted as the intersection of the two groups' posting sets with
 * the WHERE sources.  Only a pair with a visible row is a group.
 *
 * The cost of that is the product of the two cardinalities, and the estimate
 * has to say so.  Measured on the 200k-row table further down (uncorrelated
 * columns, 100-byte rows, assert build, 2026-09-21): 20 x 2 groups run in
 * 5.6 ms against the sequential aggregate's 37.3 ms and 200 x 2 in 14.9
 * against 39.3, while 200 x 20 is 60.7 against 36.8 and 20000 x 2 is 169.8
 * against 51.6 - so the middle of that range is where the node stops winning,
 * and the model must stop choosing it there.
 *
 * The decoded walk of section 34 is the other plan for two columns, and is
 * switched off down to the end of the cardinalities below, which are the
 * nested loop's; what it takes of them is shown after.
 */
SET pg_lion.enable_decoded_walk = off;
SELECT lion_pd('SELECT a, b, count(*) FROM lion_pdt GROUP BY a, b');
SELECT lion_plans('SELECT a, b, count(*) FROM lion_pdt GROUP BY a, b');
SELECT lion_pd($$SELECT a, b, count(*) FROM lion_pdt WHERE c = 'c1' GROUP BY a, b$$);
SELECT lion_pd('SELECT a, b, count(*) FROM lion_pdt WHERE b = 2 GROUP BY a, b');
SELECT lion_pd('SELECT b, a, count(*) FROM lion_pdt GROUP BY b, a');
-- NULLs in either column: each one's NULL group is its index's NULL entry
SELECT lion_pd('SELECT a, n, count(*) FROM lion_pdt GROUP BY a, n');
SELECT lion_pd('SELECT n, a, count(*) FROM lion_pdt GROUP BY n, a');
SELECT lion_pd('SELECT n, c, count(*) FROM lion_pdt GROUP BY n, c');
-- count(col) of a group column is 0 in that column's NULL group (section 14)
SELECT lion_pd('SELECT a, n, count(n) FROM lion_pdt GROUP BY a, n');
SELECT lion_pd('SELECT a, n, count(a) FROM lion_pdt GROUP BY a, n');
-- an OR restriction under a two-column grouping
SELECT lion_pd($$SELECT a, b, count(*) FROM lion_pdt WHERE c = 'c1' OR n = 2 GROUP BY a, b$$);
-- the results themselves
SELECT a, b, count(*) FROM lion_pdt GROUP BY a, b ORDER BY a, b LIMIT 8;
SELECT a, n, count(*) FROM lion_pdt GROUP BY a, n ORDER BY a, n NULLS LAST LIMIT 8;

-- a dirty heap, then a clean one
DELETE FROM lion_pdt WHERE id % 11 = 0;
SELECT lion_pd('SELECT a, b, count(*) FROM lion_pdt GROUP BY a, b');
SELECT lion_pd('SELECT a, n, count(*) FROM lion_pdt GROUP BY a, n');
SELECT lion_pd_counters('SELECT a, b, count(*) FROM lion_pdt GROUP BY a, b');
VACUUM (FREEZE) lion_pdt;
SELECT lion_pd('SELECT a, b, count(*) FROM lion_pdt GROUP BY a, b');
SELECT lion_pd_counters('SELECT a, b, count(*) FROM lion_pdt GROUP BY a, b');
SELECT a, b, count(*) FROM lion_pdt GROUP BY a, b ORDER BY a, b LIMIT 8;

-- a partitioned table: one partial aggregate per pair per partition (§16)
CREATE TABLE lion_pdq (id int NOT NULL, a int NOT NULL, b int NOT NULL)
	PARTITION BY RANGE (id);
CREATE TABLE lion_pdq1 PARTITION OF lion_pdq FOR VALUES FROM (0) TO (50000);
CREATE TABLE lion_pdq2 PARTITION OF lion_pdq FOR VALUES FROM (50000) TO (100000);
INSERT INTO lion_pdq SELECT i, i % 10, i % 4 FROM generate_series(0, 99999) i;
CREATE INDEX ON lion_pdq1 USING lion (a);
CREATE INDEX ON lion_pdq1 USING lion (b);
CREATE INDEX ON lion_pdq2 USING lion (a);
CREATE INDEX ON lion_pdq2 USING lion (b);
VACUUM (FREEZE, ANALYZE) lion_pdq;
SELECT lion_plans('SELECT a, b, count(*) FROM lion_pdq GROUP BY a, b');
SELECT lion_pd('SELECT a, b, count(*) FROM lion_pdq GROUP BY a, b');
SELECT lion_pd('SELECT a, b, count(*) FROM lion_pdq WHERE b = 2 GROUP BY a, b');
SELECT a, b, count(*) FROM lion_pdq GROUP BY a, b ORDER BY a, b LIMIT 6;

-- ---- which cardinalities the cost model accepts -------------------------
/*
 * Nothing is disabled here: which plan the model picks IS the test.  The
 * intersection of a pair scans the members of the smaller of the two groups,
 * so summed over every pair that is Min(entries, entries) x rows - and it is
 * that term, not the pair count alone, that has to refuse the wide cases.
 */
CREATE TABLE lion_pd2 (id int NOT NULL, c2 int NOT NULL, c20 int NOT NULL,
					   c200 int NOT NULL, c20k int NOT NULL, pad text NOT NULL);
INSERT INTO lion_pd2
SELECT i, i % 2, i % 20, i % 200, i % 20000, repeat('x', 200)
  FROM generate_series(1, 100000) i;
CREATE INDEX lion_pd2_c2 ON lion_pd2 USING lion (c2);
CREATE INDEX lion_pd2_c20 ON lion_pd2 USING lion (c20);
CREATE INDEX lion_pd2_c200 ON lion_pd2 USING lion (c200);
CREATE INDEX lion_pd2_c20k ON lion_pd2 USING lion (c20k);
VACUUM (FREEZE, ANALYZE) lion_pd2;
-- 20 x 2 and 200 x 2 are worth it
SELECT lion_pd('SELECT c20, c2, count(*) FROM lion_pd2 GROUP BY c20, c2');
SELECT lion_plans('SELECT c20, c2, count(*) FROM lion_pd2 GROUP BY c20, c2');
SELECT lion_pd('SELECT c200, c2, count(*) FROM lion_pd2 GROUP BY c200, c2');
-- 200 x 20 and 20000 x 200 are not, and are refused
SELECT lion_pd('SELECT c200, c20, count(*) FROM lion_pd2 GROUP BY c200, c20');
SELECT lion_plans('SELECT c200, c20, count(*) FROM lion_pd2 GROUP BY c200, c20');
SELECT lion_pd('SELECT c20k, c200, count(*) FROM lion_pd2 GROUP BY c20k, c200');
SELECT lion_pd('SELECT c20k, c2, count(*) FROM lion_pd2 GROUP BY c20k, c2');
RESET pg_lion.enable_decoded_walk;

-- ---- two columns by the decoded walk (DESIGN.md section 34) ------------
/*
 * Two passes over each container key's rows whatever the number of pairs, so
 * the wide pairs the nested loop is refused are the walk's, and so is a WHERE
 * the nested loop is priced too dear under.  The narrow pairs of bitmaps are
 * the nested loop's, a pass over the words of two NARROWs a pair at each key
 * (20 x 2 in 0.47 ms against the walk's 0.98, release build); a partitioned
 * table stays the nested loop's.
 */
SELECT lion_plans('SELECT a, b, count(*) FROM lion_pdt GROUP BY a, b');
SELECT lion_pd($$SELECT a, b, count(*) FROM lion_pdt WHERE c = 'c1' GROUP BY a, b$$);
SELECT lion_pd($$SELECT a, b, count(*) FROM lion_pdt WHERE c = 'c1' OR n = 2 GROUP BY a, b$$);
SELECT lion_pd('SELECT a, n, count(n) FROM lion_pdt GROUP BY a, n');
SELECT lion_plans('SELECT c20, c2, count(*) FROM lion_pd2 GROUP BY c20, c2');
SELECT lion_pd('SELECT c20, c2, count(*) FROM lion_pd2 GROUP BY c20, c2');
SELECT lion_pd('SELECT c200, c2, count(*) FROM lion_pd2 GROUP BY c200, c2');
SELECT lion_plans('SELECT c200, c20, count(*) FROM lion_pd2 GROUP BY c200, c20');
SELECT lion_pd('SELECT c200, c20, count(*) FROM lion_pd2 GROUP BY c200, c20');
SELECT lion_pd('SELECT c20k, c200, count(*) FROM lion_pd2 GROUP BY c20k, c200');
SELECT lion_pd('SELECT c20k, c2, count(*) FROM lion_pd2 GROUP BY c20k, c2');

-- ---- an AND with a selective source is priced as probes (§22) ----------
/*
 * The leapfrog join walks the source with the FEWEST members and seeks the
 * others to the container keys it produces, so a dense set that is ANDed with
 * a selective one is no longer read end to end - it is read at the places the
 * selective one has a container, one descent each.  Two orderings pin that
 * without pinning any one number, and both were FALSE while every source was
 * charged its whole share of the index whatever drove the merge:
 *
 *	- `c2 = 1` must add less to `id = 77`'s count than it costs alone.  `id`
 *	  is unique, so it has a container at exactly one key and half the
 *	  table's posting set is read at exactly that one place - against the
 *	  four pages of bitmaps a walk of it takes here (9.9 against 5.2 + 7.0
 *	  cost units);
 *	- and adding that clause to a count that already intersects two sets must
 *	  add less than the clause costs alone: the selective source takes over
 *	  as the driver and the other two stop being walked (14.5 against 13.3 +
 *	  5.2).  Each lookup's directory leaf is a random page, which here costs
 *	  more than the walks it saves: the counts take 7 and 8 us against 11 and
 *	  16, which the model does not see.
 *
 * What the bound is worth where it matters needs a table far too big for this
 * suite to build.  At one million rows of the benchmark's `fact`,
 * `c20k = 77 AND c200 = 17 AND c2 = 1` goes from 168.8 cost units to 117.6,
 * of which 100 is the fifty descents into `c2` that the count really makes -
 * measured as 118 buffer accesses against the 154 a walk of that set takes
 * (2026-09-22, release build).  It is still not the plan the model picks
 * there: the BitmapAnd over the same two indexes is priced at 62.1 although it
 * runs in 0.88 ms against the node's 0.25, because what it is charged for is
 * CPU per TID while the node is charged for resident index pages at
 * seq_page_cost.  DESIGN.md §22 records that as the open item.
 */
CREATE INDEX lion_pd2_id ON lion_pd2 USING lion (id);
VACUUM (FREEZE, ANALYZE) lion_pd2;
SELECT lion_pd_cost('SELECT count(*) FROM lion_pd2 WHERE id = 77 AND c2 = 1')
	 - lion_pd_cost('SELECT count(*) FROM lion_pd2 WHERE id = 77')
	 < lion_pd_cost('SELECT count(*) FROM lion_pd2 WHERE c2 = 1')
	AS probing_beats_walking;
SELECT lion_pd_cost($$SELECT count(*) FROM lion_pd2
					  WHERE id = 77 AND c200 = 17 AND c2 = 1$$)
	 - lion_pd_cost('SELECT count(*) FROM lion_pd2 WHERE c200 = 17 AND c2 = 1')
	 < lion_pd_cost('SELECT count(*) FROM lion_pd2 WHERE id = 77')
	AS selective_clause_makes_it_cheaper;
-- the answers themselves, from the node and from the ordinary plan
SELECT lion_pd('SELECT count(*) FROM lion_pd2 WHERE id = 77 AND c200 = 17 AND c2 = 1');
SELECT lion_pd('SELECT count(*) FROM lion_pd2 WHERE c20k = 77 AND c200 = 17 AND c2 = 1');

-- ---- the early exit in the AND merge (§25) ------------------------------
/*
 * The merge probes the other sources in ascending selectivity and abandons a
 * container key as soon as the running intersection is empty, so a dense
 * source is not sought at a key the selective ones have already ruled out.
 * The node is forced here, because the model does not pick it for these
 * shapes (the open item of DESIGN.md §22), and the counter is reduced to
 * zero / non-zero for the reason lion_pd_counters() gives: how many container
 * keys there are depends on how the rows fall on heap pages, but whether any
 * probe was avoided at all does not.  The answer is taken from the forced
 * node and compared with the ordinary plan's, so the early exit cannot buy
 * its pages by dropping a row.
 */
CREATE OR REPLACE FUNCTION lion_pd_probes(q text) RETURNS text
LANGUAGE plpgsql AS $$
DECLARE
	ln text;
	avoided bigint := -1;
	pushed boolean := false;
	non bigint;
	noff bigint;
BEGIN
	PERFORM set_config('enable_bitmapscan', 'off', true);
	PERFORM set_config('enable_seqscan', 'off', true);
	PERFORM set_config('pg_lion.enable_plain_scan', 'off', true);
	PERFORM set_config('pg_lion.enable_count_pushdown', 'on', true);
	FOR ln IN EXECUTE 'EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF, BUFFERS OFF) ' || q LOOP
		IF ln LIKE '%Custom Scan (LionCount)%' THEN
			pushed := true;
		END IF;
		IF btrim(split_part(ln, ':', 1)) = 'Probes Avoided' THEN
			avoided := btrim(split_part(ln, ':', 2))::bigint;
		END IF;
	END LOOP;
	IF NOT pushed THEN
		RETURN 'NOT PUSHED DOWN';
	END IF;

	EXECUTE q INTO non;
	PERFORM set_config('pg_lion.enable_count_pushdown', 'off', true);
	PERFORM set_config('enable_bitmapscan', 'on', true);
	PERFORM set_config('enable_seqscan', 'on', true);
	PERFORM set_config('pg_lion.enable_plain_scan', 'on', true);
	EXECUTE q INTO noff;
	IF non IS DISTINCT FROM noff THEN
		RETURN format('MISMATCH: node %s, ordinary plan %s', non, noff);
	END IF;

	RETURN format('Probes Avoided %s, answer %s',
				  CASE WHEN avoided < 0 THEN 'NOT REPORTED'
					   WHEN avoided = 0 THEN '= 0' ELSE '> 0' END, non);
END $$;
/*
 * `c20k = 77` and `c200 = 17` select DISJOINT sets here: 20000 is a multiple
 * of 200, so a row with i % 20000 = 77 has i % 200 = 77 and never 17.  Every
 * container key the selective `c20k` produces therefore dies on the second
 * source, and `c2` - half the table, and the one that costs pages to read -
 * is never sought at any of them.
 */
SELECT lion_pd_probes($$SELECT count(*) FROM lion_pd2
						WHERE c20k = 77 AND c200 = 17 AND c2 = 1$$);
/*
 * The control, where there is nothing to avoid: `c200 = 3` is a SUBSET of
 * `c20 = 3` (20 divides 200), so the intersection survives at every container
 * key the driver produces and all three sources are consulted at every one of
 * them.  A count whose sources really do overlap everywhere must read exactly
 * what it read before §25.
 */
SELECT lion_pd_probes($$SELECT count(*) FROM lion_pd2
						WHERE c20 = 3 AND c200 = 3 AND c2 = 1$$);
/*
 * A source that is not sought is one whose pin stays where it stood, which is
 * a pin too many and never a missing answer (DESIGN.md §9, §25): the same
 * query against a heap that is NOT all-visible has to go through the heap
 * recheck and come out with the same number.
 */
UPDATE lion_pd2 SET pad = pad WHERE id % 1000 = 0;
SELECT lion_pd_probes($$SELECT count(*) FROM lion_pd2
						WHERE c20 = 3 AND c200 = 3 AND c2 = 1$$);
DROP FUNCTION lion_pd_probes(text);
DROP TABLE lion_pd2;
DROP TABLE lion_pdq;

-- ---- two IN lists, one of them on the group column ----------------------
/*
 * An IN list on the grouping column drives the groups (DESIGN.md §15): the
 * node walks the list's located posting sets instead of every entry of the
 * index, and emits the groups in the order those were located in.  That is key
 * order for every opfamily this extension ships, and hash order for this one,
 * whose cross-type `= (int4, int8)` has no ordering an int8 list can be sorted
 * with.  The planner decided whether the list drives by a rule of its own -
 * it gave up at the second list anywhere in the WHERE - so with a list on `h`
 * as well it priced a walk of all 200 entries of `k` and promised the
 * ORDER BY a sorted output, while the executor drove the groups from the list
 * on `k` all the same and emitted 99, 150, 42, 1, ... under `ORDER BY k` (the
 * 2026-09-25 review).  Both now take the FIRST list on the driving column as
 * the driver, whatever else the WHERE holds, and a list-driven grouping claims
 * no order, so a Sort goes on top.  The other plans are disabled so that the
 * node is the plan the old estimate would have lost with.
 */
CREATE OPERATOR FAMILY lion_pdk_fam USING lion;
CREATE OPERATOR CLASS lion_pdk_ops FOR TYPE int4 USING lion
	FAMILY lion_pdk_fam AS
	OPERATOR	1	= (int4, int4),
	FUNCTION	1	hashint4(int4),
	FUNCTION	4	btint4cmp(int4, int4);
ALTER OPERATOR FAMILY lion_pdk_fam USING lion ADD
	OPERATOR	1	= (int4, int8),
	FUNCTION	1	(int8, int8) hashint8(int8);
CREATE TABLE lion_pdk (k int NOT NULL, h int NOT NULL);
INSERT INTO lion_pdk SELECT i % 200, i % 7 FROM generate_series(1, 100000) i;
CREATE INDEX lion_pdk_k ON lion_pdk USING lion (k lion_pdk_ops);
CREATE INDEX lion_pdk_h ON lion_pdk USING lion (h);
VACUUM (FREEZE, ANALYZE) lion_pdk;
SET enable_seqscan = off;
SET enable_bitmapscan = off;
SET pg_lion.enable_plain_scan = off;
EXPLAIN (COSTS OFF)
SELECT k, count(*) FROM lion_pdk
 WHERE k = ANY ('{5,17,3,150,42,99,1,120}'::int8[]) AND h IN (1, 2)
 GROUP BY k ORDER BY k;
SELECT k, count(*) FROM lion_pdk
 WHERE k = ANY ('{5,17,3,150,42,99,1,120}'::int8[]) AND h IN (1, 2)
 GROUP BY k ORDER BY k;
-- the list on `h` first: it is the first list on `k` that drives, not the first list
SELECT k, count(*) FROM lion_pdk
 WHERE h IN (1, 2) AND k = ANY ('{5,17,3,150,42,99,1,120}'::int8[])
 GROUP BY k ORDER BY k;
-- one list, as before: the list drives, and no order is claimed either
SELECT k, count(*) FROM lion_pdk
 WHERE k = ANY ('{5,17,3,150,42,99,1,120}'::int8[]) GROUP BY k ORDER BY k;
SELECT lion_pd($$SELECT k, count(*) FROM lion_pdk
				WHERE k = ANY ('{5,17,3,150,42,99,1,120}'::int8[]) AND h IN (1, 2)
				GROUP BY k$$);
RESET enable_seqscan;
RESET enable_bitmapscan;
RESET pg_lion.enable_plain_scan;
DROP TABLE lion_pdk;
DROP OPERATOR FAMILY lion_pdk_fam USING lion;
/*
 * ... and the estimate is the executor's too, with nothing disabled: the
 * eight listed entries rather than a walk of all 20000 entries of `k`.
 * Priced as that walk (332 against the bitmap plan's 155) the node lost this
 * query; priced as what it does (95) it is chosen.  The index's own opclass
 * locates the list in key order, but the path claims no order for it either
 * (DESIGN.md §21), so the ORDER BY sorts the eight groups.
 */
CREATE TABLE lion_pdk (k int NOT NULL, h int NOT NULL);
INSERT INTO lion_pdk SELECT i % 20000, i % 7 FROM generate_series(1, 100000) i;
CREATE INDEX lion_pdk_k ON lion_pdk USING lion (k);
CREATE INDEX lion_pdk_h ON lion_pdk USING lion (h);
VACUUM (FREEZE, ANALYZE) lion_pdk;
SELECT lion_pd('SELECT k, count(*) FROM lion_pdk WHERE k IN (5, 17, 3, 150, 42, 99, 1, 120) AND h IN (1, 2) GROUP BY k');
EXPLAIN (COSTS OFF)
SELECT k, count(*) FROM lion_pdk
 WHERE k IN (5, 17, 3, 150, 42, 99, 1, 120) AND h IN (1, 2)
 GROUP BY k ORDER BY k;
SELECT k, count(*) FROM lion_pdk
 WHERE k IN (5, 17, 3, 150, 42, 99, 1, 120) AND h IN (1, 2)
 GROUP BY k ORDER BY k;
DROP TABLE lion_pdk;

DROP TABLE lion_pdpo;
DROP TABLE lion_pdmo;
DROP TABLE lion_pdo;

DROP FUNCTION lion_pd_prep(text, text);

DROP TABLE lion_pdn;
DROP TABLE lion_pdv;
DROP TABLE lion_pdc;
DROP OPERATOR CLASS lion_lower_ops USING lion;
DROP OPERATOR === (text, text);
DROP FUNCTION lion_lower_eq(text, text);
DROP FUNCTION lion_lower_hash(text);
DROP TABLE lion_pdw;
DROP TABLE lion_pdd;
DROP TABLE lion_pdh;
DROP TABLE lion_pdg;
DROP TABLE lion_pdi;
DROP TABLE lion_pde;
DROP TABLE lion_pdt;
DROP FUNCTION lion_pd(text);
DROP FUNCTION lion_pd_counters(text);
DROP FUNCTION lion_pd_cost(text);
DROP FUNCTION lion_plans(text);
-- enable_partitionwise_aggregate must not switch the pushdown off for a plain table
CREATE TABLE lion_pwa (k int NOT NULL);
INSERT INTO lion_pwa SELECT g % 10 FROM generate_series(1, 20000) g;
CREATE INDEX lion_pwa_k ON lion_pwa USING lion (k);
SET synchronous_commit = on;
VACUUM (FREEZE, ANALYZE) lion_pwa;
SET enable_partitionwise_aggregate = on;
EXPLAIN (COSTS OFF) SELECT count(*) FROM lion_pwa WHERE k = 3;
EXPLAIN (COSTS OFF) SELECT k, count(*) FROM lion_pwa GROUP BY k;
RESET enable_partitionwise_aggregate;
DROP TABLE lion_pwa;