-- IN lists / ScalarArrayOpExpr (DESIGN.md section 15).
--
-- `col = ANY (array)` reaches the access method as one SK_SEARCHARRAY scan
-- key (amsearcharray) and the count pushdown as one clause whose posting set
-- is the union of the listed values'.  Every query below is answered twice,
-- through the index and through a sequential scan, and the two results must
-- be equal as multisets.
\set VERBOSITY terse
SET client_min_messages = warning;
LOAD 'pg_lion';
/* VACUUM can only set all-visible for commits that reached disk (citext.sql). */
SET synchronous_commit = on;
SET default_statistics_target = 1000;
CREATE EXTENSION IF NOT EXISTS pg_lion;
/*
 * lion_incmp() runs one query through the index (count pushdown enabled) and
 * through a plain sequential scan, and reports whether the custom node was
 * used and how many rows came out, after proving the two results match.
 */
CREATE FUNCTION lion_incmp(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);
	PERFORM set_config('enable_seqscan', 'off', 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_i_idx AS %s', q);

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

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

	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 'index scan' END,
				  nrows);
END $$;
/* The plan of one query with the index preferred. */
CREATE FUNCTION lion_inplan(q text) RETURNS SETOF text
LANGUAGE plpgsql AS $$
DECLARE
	ln text;
BEGIN
	PERFORM set_config('enable_seqscan', 'off', true);
	FOR ln IN EXECUTE 'EXPLAIN (COSTS OFF) ' || q LOOP
		-- 18 marks the disabled scan; 16 and 17 do not
		CONTINUE WHEN ln ~ '^\s*Disabled: true$';
		RETURN NEXT ln;
	END LOOP;
	PERFORM set_config('enable_seqscan', 'on', true);
END $$;
/*
 * The literal text of an array of values lo .. hi taken modulo m, so that the
 * queries below hold a Const array - which is what the pushdown requires -
 * without spelling two hundred numbers out in the test file.
 */
CREATE FUNCTION lion_inlist(lo int, hi int, m int) RETURNS text
LANGUAGE sql AS $$
	SELECT '{' || string_agg((g % m)::text, ',' ORDER BY g) || '}'
	  FROM generate_series(lo, hi) g
$$;
CREATE TABLE lion_in (
	id	int		NOT NULL,
	a	int		NOT NULL,		-- 500 distinct values
	b	int		NOT NULL,		-- 7 distinct values
	c	text	NOT NULL,		-- 4 distinct values
	n	int						-- nullable, 5 values
);
INSERT INTO lion_in
SELECT i, i % 500, 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_in_a ON lion_in USING lion (a);
CREATE INDEX lion_in_b ON lion_in USING lion (b);
CREATE INDEX lion_in_c ON lion_in USING lion (c);
CREATE INDEX lion_in_n ON lion_in USING lion (n);
VACUUM (FREEZE, ANALYZE) lion_in;
SELECT lion_index_verify('lion_in_a', true);
 lion_index_verify 
-------------------
 
(1 row)

-- ---- plans -------------------------------------------------------------
SELECT lion_inplan('SELECT id FROM lion_in WHERE a IN (1, 2, 3)');
                  lion_inplan                   
------------------------------------------------
 Index Scan using lion_in_a on lion_in
   Index Cond: (a = ANY ('{1,2,3}'::integer[]))
(2 rows)

SELECT lion_inplan('SELECT count(*) FROM lion_in WHERE a IN (1, 2, 3)');
                        lion_inplan                         
------------------------------------------------------------
 Custom Scan (LionCount)
   Lion Indexes: lion_in_a (a = ANY ('{1,2,3}'::integer[]))
(2 rows)

SELECT lion_inplan($$SELECT count(*) FROM lion_in WHERE a IN (1, 2) AND c = 'c1'$$);
                                     lion_inplan                                      
--------------------------------------------------------------------------------------
 Custom Scan (LionCount)
   Lion Indexes: lion_in_a (a = ANY ('{1,2}'::integer[])), lion_in_c (c = 'c1'::text)
(2 rows)

SELECT lion_inplan('SELECT b, count(*) FROM lion_in WHERE b IN (1, 2) GROUP BY b');
                               lion_inplan                               
-------------------------------------------------------------------------
 Custom Scan (LionCount)
   Lion Indexes: lion_in_b (b), lion_in_b (b = ANY ('{1,2}'::integer[]))
   Group Key: b
(3 rows)

-- = ALL is not a union of keys and must not be pushed down
SELECT lion_inplan('SELECT count(*) FROM lion_in WHERE a = ALL (ARRAY[1, 2])');
                  lion_inplan                   
------------------------------------------------
 Aggregate
   ->  Seq Scan on lion_in
         Filter: (a = ALL ('{1,2}'::integer[]))
(3 rows)

-- ---- bitmap scans (amsearcharray) --------------------------------------
SET pg_lion.enable_count_pushdown = off;
SELECT lion_incmp('SELECT id FROM lion_in WHERE a IN (7)');
      lion_incmp      
----------------------
 index scan, 200 rows
(1 row)

SELECT lion_incmp('SELECT id FROM lion_in WHERE a IN (7, 8, 9)');
      lion_incmp      
----------------------
 index scan, 600 rows
(1 row)

SELECT lion_incmp('SELECT id FROM lion_in WHERE a IN (7, 7, 8, 8, 8)');
      lion_incmp      
----------------------
 index scan, 400 rows
(1 row)

SELECT lion_incmp('SELECT id FROM lion_in WHERE a IN (7, NULL, 8)');
      lion_incmp      
----------------------
 index scan, 400 rows
(1 row)

SELECT lion_incmp('SELECT id FROM lion_in WHERE a IN (NULL, NULL)');
     lion_incmp     
--------------------
 index scan, 0 rows
(1 row)

SELECT lion_incmp('SELECT id FROM lion_in WHERE a = ANY (''{}''::int[])');
     lion_incmp     
--------------------
 index scan, 0 rows
(1 row)

SELECT lion_incmp('SELECT id FROM lion_in WHERE a IN (900, 901)');
     lion_incmp     
--------------------
 index scan, 0 rows
(1 row)

SELECT lion_incmp('SELECT id FROM lion_in WHERE a IN (7, 900)');
      lion_incmp      
----------------------
 index scan, 200 rows
(1 row)

-- 200 values, and the same 200 with every one of them repeated
SELECT lion_incmp(format('SELECT id FROM lion_in WHERE a = ANY (''%s''::int[])', lion_inlist(0, 199, 200)));
       lion_incmp       
------------------------
 index scan, 40000 rows
(1 row)

SELECT lion_incmp(format('SELECT id FROM lion_in WHERE a = ANY (''%s''::int[])', lion_inlist(0, 399, 200)));
       lion_incmp       
------------------------
 index scan, 40000 rows
(1 row)

-- cross-type arrays
SELECT lion_incmp('SELECT id FROM lion_in WHERE a = ANY (''{7,8}''::int8[])');
      lion_incmp      
----------------------
 index scan, 400 rows
(1 row)

SELECT lion_incmp('SELECT id FROM lion_in WHERE a = ANY (''{7,8}''::int2[])');
      lion_incmp      
----------------------
 index scan, 400 rows
(1 row)

-- text, and a list combined with other quals
SELECT lion_incmp($$SELECT id FROM lion_in WHERE c IN ('c1', 'c3')$$);
       lion_incmp       
------------------------
 index scan, 50000 rows
(1 row)

SELECT lion_incmp($$SELECT id FROM lion_in WHERE a IN (7, 8) AND c IN ('c1', 'c3')$$);
      lion_incmp      
----------------------
 index scan, 200 rows
(1 row)

SELECT lion_incmp($$SELECT id FROM lion_in WHERE a IN (7, 8) AND c = 'c3'$$);
      lion_incmp      
----------------------
 index scan, 200 rows
(1 row)

-- a nullable column: NULL is never = ANY(...)
SELECT lion_incmp('SELECT id FROM lion_in WHERE n IN (1, 2)');
       lion_incmp       
------------------------
 index scan, 39604 rows
(1 row)

SELECT lion_incmp('SELECT id FROM lion_in WHERE n IN (1, 2) OR n IS NULL');
       lion_incmp       
------------------------
 index scan, 40594 rows
(1 row)

/*
 * Two quals on the one key column reach the access method as two scan keys.
 * It answers both lists and ANDs them (DESIGN.md §5 SCAN step 5), and an
 * `IS NOT NULL` beside a list or an equality is implied by it.
 */
SELECT lion_incmp('SELECT id FROM lion_in WHERE b IN (1, 2) AND b IN (2, 3)');
       lion_incmp       
------------------------
 index scan, 14286 rows
(1 row)

SELECT lion_incmp('SELECT id FROM lion_in WHERE b IN (1, 2) AND b = 2');
       lion_incmp       
------------------------
 index scan, 14286 rows
(1 row)

SELECT lion_incmp('SELECT id FROM lion_in WHERE n IN (1, 2) AND n IS NOT NULL');
       lion_incmp       
------------------------
 index scan, 39604 rows
(1 row)

SELECT lion_incmp('SELECT id FROM lion_in WHERE n IS NOT NULL AND n = 3');
       lion_incmp       
------------------------
 index scan, 19802 rows
(1 row)

SET pg_lion.enable_count_pushdown = on;
-- ---- the count pushdown ------------------------------------------------
SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE a IN (7)');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE a IN (7, 8, 9)');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE a IN (7, 7, 8, 8, 8)');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE a IN (7, NULL, 8)');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE a IN (NULL, NULL)');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE a = ANY (''{}''::int[])');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE a IN (900, 901)');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE a IN (7, 900)');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp(format('SELECT count(*) FROM lion_in WHERE a = ANY (''%s''::int[])', lion_inlist(0, 199, 200)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp(format('SELECT count(*) FROM lion_in WHERE a = ANY (''%s''::int[])', lion_inlist(0, 399, 200)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE a = ANY (''{7,8}''::int8[])');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE a = ANY (''{7,8}''::int2[])');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp($$SELECT count(*) FROM lion_in WHERE c IN ('c1', 'c3')$$);
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp($$SELECT count(*) FROM lion_in WHERE b IN (1, 2) AND c IN ('c1', 'c3')$$);
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp($$SELECT count(*) FROM lion_in WHERE b IN (1, 2) AND c = 'c3'$$);
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE n IN (1, 2)');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

-- an IN list and a null test on the same column
SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE n IN (1, 2) AND n IS NOT NULL');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

-- the same list twice is a duplicate clause; two different ones are two
-- sources, ANDed like lists on two columns (samecolumn.sql)
SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE b IN (1, 2) AND b IN (1, 2)');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE b IN (1, 2) AND b IN (2, 3)');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

-- = ALL is answered by the ordinary plan
SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE a = ALL (ARRAY[1, 2])');
     lion_incmp     
--------------------
 index scan, 1 rows
(1 row)

SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE a = ALL (ARRAY[1])');
     lion_incmp     
--------------------
 index scan, 1 rows
(1 row)

-- ---- GROUP BY over the listed values -----------------------------------
SELECT lion_incmp('SELECT b, count(*) FROM lion_in WHERE b IN (1, 2) GROUP BY b');
     lion_incmp      
---------------------
 pushed down, 2 rows
(1 row)

SELECT lion_incmp($$SELECT b, count(*) FROM lion_in WHERE b IN (1, 2) AND c IN ('c1', 'c3') GROUP BY b$$);
     lion_incmp      
---------------------
 pushed down, 2 rows
(1 row)

SELECT lion_incmp('SELECT b, count(*) FROM lion_in WHERE b IN (900, 901) GROUP BY b');
     lion_incmp      
---------------------
 pushed down, 0 rows
(1 row)

SELECT lion_incmp('SELECT b, count(*) FROM lion_in WHERE b = ANY (''{}''::int[]) GROUP BY b');
     lion_incmp      
---------------------
 pushed down, 0 rows
(1 row)

SELECT lion_incmp('SELECT n, count(*) FROM lion_in WHERE n IN (1, 2) GROUP BY n');
     lion_incmp      
---------------------
 pushed down, 2 rows
(1 row)

SELECT lion_incmp($$SELECT c, count(*) FROM lion_in WHERE b IN (1, 2) GROUP BY c$$);
     lion_incmp      
---------------------
 pushed down, 4 rows
(1 row)

SELECT lion_incmp('SELECT count(b) FROM lion_in WHERE b IN (1, 2)');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

-- the results themselves
SELECT b, count(*) FROM lion_in WHERE b IN (1, 2) GROUP BY b ORDER BY b;
 b | count 
---+-------
 1 | 14286
 2 | 14286
(2 rows)

SELECT count(*) FROM lion_in WHERE a IN (7, 8, 9);
 count 
-------
   600
(1 row)

SELECT count(*) FROM lion_in WHERE a IN (7, NULL, 8);
 count 
-------
   400
(1 row)

SELECT count(*) FROM lion_in WHERE a = ANY ('{}'::int[]);
 count 
-------
     0
(1 row)

-- ---- a dirty heap (the recheck path) -----------------------------------
DELETE FROM lion_in WHERE id % 3 = 0;
SELECT lion_index_verify('lion_in_a', true);
 lion_index_verify 
-------------------
 
(1 row)

SET pg_lion.enable_count_pushdown = off;
SELECT lion_incmp('SELECT id FROM lion_in WHERE a IN (7, 8, 9)');
      lion_incmp      
----------------------
 index scan, 400 rows
(1 row)

SET pg_lion.enable_count_pushdown = on;
SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE a IN (7, 8, 9)');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE b IN (1, 2)');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp('SELECT b, count(*) FROM lion_in WHERE b IN (1, 2) GROUP BY b');
     lion_incmp      
---------------------
 pushed down, 2 rows
(1 row)

-- ---- and after VACUUM (the visibility-map path) ------------------------
VACUUM (FREEZE) lion_in;
SELECT lion_index_verify('lion_in_a', true);
 lion_index_verify 
-------------------
 
(1 row)

SET pg_lion.enable_count_pushdown = off;
SELECT lion_incmp('SELECT id FROM lion_in WHERE a IN (7, 8, 9)');
      lion_incmp      
----------------------
 index scan, 400 rows
(1 row)

SET pg_lion.enable_count_pushdown = on;
SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE a IN (7, 8, 9)');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp('SELECT count(*) FROM lion_in WHERE b IN (1, 2)');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp('SELECT b, count(*) FROM lion_in WHERE b IN (1, 2) GROUP BY b');
     lion_incmp      
---------------------
 pushed down, 2 rows
(1 row)

SELECT lion_incmp($$SELECT count(*) FROM lion_in WHERE b IN (1, 2) AND c IN ('c1', 'c3')$$);
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

-- ---- long literal lists -----------------------------------------------
/*
 * A literal list of up to LION_MAX_LIST_ELEMS (100,000) values is pushed down
 * (DESIGN.md §15, "Long lists"): one longer than a batch is located a batch at
 * a time, as a parameter's array always was.  1,200 values was past the old
 * cap of 1,000.  Past 100,000 the pushdown declines and the bitmap scan
 * answers.  Both must agree with the heap.
 */
SELECT lion_incmp(format('SELECT count(*) FROM lion_in WHERE a = ANY (''%s''::int[])', lion_inlist(0, 1200, 100000)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT count(*) FROM lion_in WHERE a < 1201;
 count 
-------
 66667
(1 row)

SELECT lion_incmp(format('SELECT a, count(*) FROM lion_in WHERE a = ANY (''%s''::int[]) GROUP BY a', lion_inlist(0, 1200, 100000)));
      lion_incmp       
-----------------------
 pushed down, 500 rows
(1 row)

SELECT lion_incmp(format('SELECT count(*) FROM lion_in WHERE a = ANY (''%s''::int[]) AND b = 3', lion_inlist(0, 1200, 100000)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

-- ... in batches: at 64kB of work_mem a batch is 1,000 values
SET work_mem = '64kB';
SELECT lion_incmp(format('SELECT count(*) FROM lion_in WHERE a = ANY (''%s''::int[])', lion_inlist(0, 2500, 1700)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp(format('SELECT a, count(*) FROM lion_in WHERE a = ANY (''%s''::int[]) GROUP BY a', lion_inlist(0, 2500, 1700)));
      lion_incmp       
-----------------------
 pushed down, 500 rows
(1 row)

RESET work_mem;
-- the longest list pushed down, 100,000 values, and one more declined
SELECT lion_incmp(format('SELECT count(*) FROM lion_in WHERE a = ANY (''%s''::int[])', lion_inlist(0, 99999, 1000000)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp(format('SELECT count(*) FROM lion_in WHERE a = ANY (''%s''::int[])', lion_inlist(0, 100000, 1000000)));
     lion_incmp     
--------------------
 index scan, 1 rows
(1 row)

/*
 * The union itself: lion_index_count_any() (DESIGN.md section 15).
 *
 * The pushdown's IN list and this function share the whole evaluation - the
 * values are located by lion_posting_set_lookup_many(), which hashes them all
 * and then visits the bucket pages in (bucket, hash) order, and the union is
 * the k-way merge over the leaf cursors, ORed through a bitset image once
 * more than a handful of them stand at one container key.  Going through SQL
 * reaches the shapes the pushdown declines: a list longer than
 * LION_MAX_ARRAY_ELEMS, and lists whose duplicates and NULLs are not constant
 * folded away.
 */
CREATE FUNCTION lion_anycmp(idx text, tbl text, col text, arr text)
RETURNS text LANGUAGE plpgsql AS $$
DECLARE
	a bigint;
	b bigint;
BEGIN
	EXECUTE format('SELECT lion_index_count_any(%L::regclass, %s)', idx, arr)
		INTO a;
	EXECUTE format('SELECT count(*) FROM %s WHERE %I = ANY (%s)', tbl, col, arr)
		INTO b;
	IF a IS DISTINCT FROM b THEN
		RETURN format('MISMATCH roaring=%s select=%s', a, b);
	END IF;
	RETURN format('ok %s', a);
END $$;
SELECT lion_anycmp('lion_in_a', 'lion_in', 'a', '''{7}''::int[]');
 lion_anycmp 
-------------
 ok 133
(1 row)

SELECT lion_anycmp('lion_in_a', 'lion_in', 'a', '''{7,8,9}''::int[]');
 lion_anycmp 
-------------
 ok 400
(1 row)

-- duplicates: a union of a set with itself is that set
SELECT lion_anycmp('lion_in_a', 'lion_in', 'a', '''{7,7,7,8,8,9}''::int[]');
 lion_anycmp 
-------------
 ok 400
(1 row)

-- NULLs are not = ANY(...), and an all-NULL or empty list selects nothing
SELECT lion_anycmp('lion_in_a', 'lion_in', 'a', '''{7,NULL,8}''::int[]');
 lion_anycmp 
-------------
 ok 267
(1 row)

SELECT lion_anycmp('lion_in_a', 'lion_in', 'a', '''{NULL,NULL}''::int[]');
 lion_anycmp 
-------------
 ok 0
(1 row)

SELECT lion_anycmp('lion_in_a', 'lion_in', 'a', '''{}''::int[]');
 lion_anycmp 
-------------
 ok 0
(1 row)

-- values with no entry at all, alone and mixed in
SELECT lion_anycmp('lion_in_a', 'lion_in', 'a', '''{900,901}''::int[]');
 lion_anycmp 
-------------
 ok 0
(1 row)

SELECT lion_anycmp('lion_in_a', 'lion_in', 'a', '''{7,900,8,901}''::int[]');
 lion_anycmp 
-------------
 ok 267
(1 row)

-- every key of the index, which is every row: the union is the whole table
SELECT lion_anycmp('lion_in_a', 'lion_in', 'a',
				  format('%L::int[]', lion_inlist(0, 499, 500)));
 lion_anycmp 
-------------
 ok 66667
(1 row)

-- ... and the same list with every value repeated four times, plus absent ones
SELECT lion_anycmp('lion_in_a', 'lion_in', 'a',
				  format('%L::int[]', lion_inlist(0, 2499, 625)));
 lion_anycmp 
-------------
 ok 66667
(1 row)

-- a 3000-value list: longer than the pushdown accepts, and exact here
SELECT lion_anycmp('lion_in_a', 'lion_in', 'a',
				  format('%L::int[]', lion_inlist(0, 2999, 100000)));
 lion_anycmp 
-------------
 ok 66667
(1 row)

-- by-reference keys, a two-value list, and cross-type element types
SELECT lion_anycmp('lion_in_c', 'lion_in', 'c', '''{c1,c3}''::text[]');
 lion_anycmp 
-------------
 ok 33333
(1 row)

SELECT lion_anycmp('lion_in_b', 'lion_in', 'b', '''{1,2}''::int8[]');
 lion_anycmp 
-------------
 ok 19048
(1 row)

SELECT lion_anycmp('lion_in_b', 'lion_in', 'b', '''{1,2}''::int2[]');
 lion_anycmp 
-------------
 ok 19048
(1 row)

-- a nullable column: the reserved NULL entry is not a listed value
SELECT lion_anycmp('lion_in_n', 'lion_in', 'n', '''{1,2}''::int[]');
 lion_anycmp 
-------------
 ok 26403
(1 row)

SELECT lion_anycmp('lion_in_n', 'lion_in', 'n', '''{1,2,3,4,5,6}''::int[]');
 lion_anycmp 
-------------
 ok 52805
(1 row)

-- the same errors the single-key count raises
SELECT lion_index_count_any('lion_in_a', '{1}'::text[]);
ERROR:  type text cannot be compared with index "lion_in_a"
SELECT lion_index_count_any('lion_in', '{1}'::int[]);
ERROR:  "lion_in" is not an index
SELECT lion_index_count_any('lion_in_a', NULL::int[]) AS null_array;
 null_array 
------------
          0
(1 row)

-- ---- an empty table ----------------------------------------------------
CREATE TABLE lion_in_empty (k int NOT NULL);
CREATE INDEX lion_in_empty_k ON lion_in_empty USING lion (k);
VACUUM (FREEZE, ANALYZE) lion_in_empty;
SELECT lion_incmp('SELECT count(*) FROM lion_in_empty WHERE k IN (1, 2)');
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_incmp('SELECT k, count(*) FROM lion_in_empty WHERE k IN (1, 2) GROUP BY k');
     lion_incmp      
---------------------
 pushed down, 0 rows
(1 row)

-- ---- long lists: the disjoint-sum short-circuit (DESIGN.md §15) ---------
/*
 * The posting sets of two different entries of a SCALAR index are disjoint -
 * a row has one value in the column, so its TID is under exactly one entry -
 * so the count of an IN list's union is the SUM of the per-entry counts and
 * no union has to be built at all.  It applies when the list is the only
 * positive source, and when the list is on the very column a GROUP BY drives;
 * it must NOT apply when the list is intersected with something else (the
 * union has to be materialised per container key to be intersected), nor for
 * a multi-key opclass (DESIGN.md §17), where one row is under many entries
 * and the sum over them is not a row count.
 *
 * Every query below is still answered twice and compared as a multiset, which
 * is what proves the sum and the merge agree; lion_insum() then reports which
 * of the two the node took, from the EXPLAIN counter.
 */
CREATE FUNCTION lion_insum(q text) RETURNS text
LANGUAGE plpgsql AS $$
DECLARE
	ln text;
	nm text;
	pushed boolean := false;
	summed bigint := 0;
BEGIN
	PERFORM set_config('enable_seqscan', 'off', 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;
		nm := btrim(split_part(ln, ':', 1));
		IF nm = 'Posting Sets Summed' THEN
			summed := btrim(split_part(ln, ':', 2))::bigint;
		END IF;
	END LOOP;
	PERFORM set_config('enable_seqscan', 'on', true);
	IF NOT pushed THEN
		RETURN 'NOT PUSHED DOWN';
	END IF;
	RETURN format('posting sets summed %s',
				  CASE WHEN summed = 0 THEN '= 0' ELSE '> 0' END);
END $$;
/* Does the planner pick the node on cost, with nothing disabled? */
CREATE FUNCTION lion_inpick(q text) RETURNS text
LANGUAGE plpgsql AS $$
DECLARE
	ln text;
BEGIN
	FOR ln IN EXECUTE 'EXPLAIN (COSTS OFF) ' || q LOOP
		IF ln LIKE '%Custom Scan (LionCount)%' THEN
			RETURN 'pushed down';
		END IF;
	END LOOP;
	RETURN 'not pushed down';
END $$;
CREATE TABLE lion_inbig (
	id		int			NOT NULL,
	k		int			NOT NULL,	-- 2000 distinct values, high cardinality
	t		text		NOT NULL,	-- the same values as text
	lo		int			NOT NULL,	-- 4 distinct values, low cardinality
	md		int			NOT NULL,	-- 50 distinct values: DENSE, and enough of
									-- them for the merge's bitset image
	n		int,					-- nullable, 2000 values and NULL
	tags	text[]					-- multi-key (DESIGN.md §17)
);
INSERT INTO lion_inbig
SELECT i, i % 2000, 't' || (i % 2000), i % 4, i % 50,
	   CASE WHEN i % 7 = 0 THEN NULL ELSE i % 2000 END,
	   ARRAY['g' || (i % 50), 'g' || (i % 97)]
  FROM generate_series(1, 200000) i;
CREATE INDEX lion_inbig_k ON lion_inbig USING lion (k);
CREATE INDEX lion_inbig_t ON lion_inbig USING lion (t);
CREATE INDEX lion_inbig_lo ON lion_inbig USING lion (lo);
CREATE INDEX lion_inbig_md ON lion_inbig USING lion (md);
CREATE INDEX lion_inbig_n ON lion_inbig USING lion (n);
CREATE INDEX lion_inbig_tags ON lion_inbig USING lion (tags);
VACUUM (FREEZE, ANALYZE) lion_inbig;
SELECT lion_index_verify('lion_inbig_k', true);
 lion_index_verify 
-------------------
 
(1 row)

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

-- a thousand values on an int column, and the same list on a text column
SELECT lion_incmp(format('SELECT count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[])',
						lion_inlist(0, 999, 100000)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_insum(format('SELECT count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[])',
						lion_inlist(0, 999, 100000)));
       lion_insum        
-------------------------
 posting sets summed > 0
(1 row)

SELECT lion_incmp(format($$SELECT count(*) FROM lion_inbig WHERE t = ANY ('%s'::text[])$$,
						(SELECT '{' || string_agg('t' || g, ',') || '}'
						   FROM generate_series(0, 999) g)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_insum(format($$SELECT count(*) FROM lion_inbig WHERE t = ANY ('%s'::text[])$$,
						(SELECT '{' || string_agg('t' || g, ',') || '}'
						   FROM generate_series(0, 999) g)));
       lion_insum        
-------------------------
 posting sets summed > 0
(1 row)

-- duplicates (every value four times) and values with no entry at all
SELECT lion_incmp(format('SELECT count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[])',
						lion_inlist(0, 999, 250)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_insum(format('SELECT count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[])',
						lion_inlist(0, 999, 250)));
       lion_insum        
-------------------------
 posting sets summed > 0
(1 row)

SELECT lion_incmp(format('SELECT count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[])',
						lion_inlist(3000, 3999, 100000)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

-- NULLs in the list, and a nullable column (the NULL entry is not a value)
SELECT lion_incmp(format('SELECT count(*) FROM lion_inbig WHERE n = ANY (''{NULL,%s}''::int[])',
						(SELECT string_agg(g::text, ',') FROM generate_series(0, 998) g)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_insum(format('SELECT count(*) FROM lion_inbig WHERE n = ANY (''{NULL,%s}''::int[])',
						(SELECT string_agg(g::text, ',') FROM generate_series(0, 998) g)));
       lion_insum        
-------------------------
 posting sets summed > 0
(1 row)

-- a GROUP BY on the very column the list constrains: the groups are the values
SELECT lion_incmp(format('SELECT k, count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[]) GROUP BY k',
						lion_inlist(0, 999, 100000)));
       lion_incmp       
------------------------
 pushed down, 1000 rows
(1 row)

SELECT lion_insum(format('SELECT k, count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[]) GROUP BY k',
						lion_inlist(0, 999, 100000)));
       lion_insum        
-------------------------
 posting sets summed = 0
(1 row)

-- ... including with a second clause, which the groups are intersected with
-- ... and a GROUP BY on a DIFFERENT column, which the entry scan still drives.
-- (Both forced: the node takes 9.4 and 9.2 ms of CPU, the bitmap heap scan and
-- hash aggregate 15.6 and 20, but core prices hashing a row at a fraction of
-- what it costs and the model chooses the aggregate - DESIGN.md §10.)
SET enable_bitmapscan = off;
SET pg_lion.enable_plain_scan = off;
SELECT lion_incmp(format('SELECT k, count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[]) AND lo = 1 GROUP BY k',
						lion_inlist(0, 999, 100000)));
      lion_incmp       
-----------------------
 pushed down, 250 rows
(1 row)

SELECT lion_incmp(format('SELECT lo, count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[]) GROUP BY lo',
						lion_inlist(0, 999, 100000)));
     lion_incmp      
---------------------
 pushed down, 4 rows
(1 row)

RESET enable_bitmapscan;
RESET pg_lion.enable_plain_scan;
-- ANDed with another clause: the merge, not the sum
SELECT lion_incmp(format('SELECT count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[]) AND lo = 1',
						lion_inlist(0, 999, 100000)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_insum(format('SELECT count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[]) AND lo = 1',
						lion_inlist(0, 999, 100000)));
       lion_insum        
-------------------------
 posting sets summed = 0
(1 row)

-- an IN list under an OR is one leaf of a union, and never summed
SELECT lion_incmp(format('SELECT count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[]) OR lo = 1',
						lion_inlist(0, 99, 100000)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_insum(format('SELECT count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[]) OR lo = 1',
						lion_inlist(0, 99, 100000)));
       lion_insum        
-------------------------
 posting sets summed = 0
(1 row)

-- a multi-key column: one row is under many entries, so the sum would be wrong
SELECT lion_incmp($$SELECT count(*) FROM lion_inbig WHERE tags && ARRAY['g1','g2','g3']$$);
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_insum($$SELECT count(*) FROM lion_inbig WHERE tags && ARRAY['g1','g2','g3']$$);
       lion_insum        
-------------------------
 posting sets summed = 0
(1 row)

SELECT lion_insum($$SELECT count(*) FROM lion_inbig WHERE tags @> ARRAY['g1']$$);
       lion_insum        
-------------------------
 posting sets summed = 0
(1 row)

/*
 * The sum is also declined when it would be SLOWER, which is a question about
 * the data and not about correctness (lion_sum_is_cheaper()): entries dense
 * enough that the merge's union amortizes the visibility-map check over all of
 * them, and enough entries for the merge to reach its bitset image.  `md` has
 * fifty values of four thousand rows each, so a list of all fifty is both;
 * ten of them is not, because the merge would fold them pairwise.
 */
SELECT lion_incmp(format('SELECT count(*) FROM lion_inbig WHERE md = ANY (''%s''::int[])',
						lion_inlist(0, 49, 100000)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_insum(format('SELECT count(*) FROM lion_inbig WHERE md = ANY (''%s''::int[])',
						lion_inlist(0, 49, 100000)));
       lion_insum        
-------------------------
 posting sets summed = 0
(1 row)

SELECT lion_incmp(format('SELECT count(*) FROM lion_inbig WHERE md = ANY (''%s''::int[])',
						lion_inlist(0, 9, 100000)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_insum(format('SELECT count(*) FROM lion_inbig WHERE md = ANY (''%s''::int[])',
						lion_inlist(0, 9, 100000)));
       lion_insum        
-------------------------
 posting sets summed > 0
(1 row)

-- ---- a dirty heap, then an all-visible one -----------------------------
DELETE FROM lion_inbig WHERE id % 5 = 0;
SELECT lion_incmp(format('SELECT count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[])',
						lion_inlist(0, 999, 100000)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_insum(format('SELECT count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[])',
						lion_inlist(0, 999, 100000)));
       lion_insum        
-------------------------
 posting sets summed > 0
(1 row)

SELECT lion_incmp(format('SELECT k, count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[]) GROUP BY k',
						lion_inlist(0, 999, 100000)));
      lion_incmp       
-----------------------
 pushed down, 800 rows
(1 row)

VACUUM (FREEZE) lion_inbig;
SELECT lion_index_verify('lion_inbig_k', true);
 lion_index_verify 
-------------------
 
(1 row)

SELECT lion_incmp(format('SELECT count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[])',
						lion_inlist(0, 999, 100000)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_insum(format('SELECT count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[])',
						lion_inlist(0, 999, 100000)));
       lion_insum        
-------------------------
 posting sets summed > 0
(1 row)

-- ---- a partitioned table (DESIGN.md §16) -------------------------------
CREATE TABLE lion_inpart (id int NOT NULL, k int NOT NULL) PARTITION BY RANGE (id);
CREATE TABLE lion_inpart1 PARTITION OF lion_inpart FOR VALUES FROM (1) TO (50000);
CREATE TABLE lion_inpart2 PARTITION OF lion_inpart FOR VALUES FROM (50000) TO (200000);
/* 20000 distinct keys, so that each partition's entries are sparse enough for
 * the sum: a partition is a relation of its own and the choice is made per
 * partition, over that partition's heap. */
INSERT INTO lion_inpart SELECT i, i % 20000 FROM generate_series(1, 199999) i;
CREATE INDEX lion_inpart1_k ON lion_inpart1 USING lion (k);
CREATE INDEX lion_inpart2_k ON lion_inpart2 USING lion (k);
VACUUM (FREEZE, ANALYZE) lion_inpart;
SELECT lion_incmp(format('SELECT count(*) FROM lion_inpart WHERE k = ANY (''%s''::int[])',
						lion_inlist(0, 999, 100000)));
     lion_incmp      
---------------------
 pushed down, 1 rows
(1 row)

SELECT lion_insum(format('SELECT count(*) FROM lion_inpart WHERE k = ANY (''%s''::int[])',
						lion_inlist(0, 999, 100000)));
       lion_insum        
-------------------------
 posting sets summed > 0
(1 row)

SELECT lion_incmp(format('SELECT k, count(*) FROM lion_inpart WHERE k = ANY (''%s''::int[]) GROUP BY k',
						lion_inlist(0, 999, 100000)));
       lion_incmp       
------------------------
 pushed down, 1000 rows
(1 row)

-- ---- plan choice, with nothing disabled --------------------------------
/*
 * A list over MANY distinct keys used to lose to the B-tree above a handful of
 * values, because every element was another sub-cursor in the union; with the
 * sum there is no union and the node wins at every length (DESIGN.md §15).  A
 * list over FEW distinct keys wins by a wide margin either way, because the
 * B-tree has to read an index tuple per row.
 */
CREATE INDEX lion_inbig_kb ON lion_inbig (k);
CREATE INDEX lion_inbig_lob ON lion_inbig (lo);
VACUUM (FREEZE, ANALYZE) lion_inbig;
SELECT n, lion_inpick(format('SELECT count(*) FROM lion_inbig WHERE k = ANY (''%s''::int[])',
							lion_inlist(0, n - 1, 100000)))
  FROM (VALUES (3), (10), (100), (1000)) v(n);
  n   | lion_inpick 
------+-------------
    3 | pushed down
   10 | pushed down
  100 | pushed down
 1000 | pushed down
(4 rows)

SELECT n, lion_inpick(format('SELECT count(*) FROM lion_inbig WHERE lo = ANY (''%s''::int[])',
							lion_inlist(0, n - 1, 4)))
  FROM (VALUES (3), (10), (100), (1000)) v(n);
  n   | lion_inpick 
------+-------------
    3 | pushed down
   10 | pushed down
  100 | pushed down
 1000 | pushed down
(4 rows)

DROP TABLE lion_in, lion_in_empty, lion_inbig, lion_inpart;
DROP FUNCTION lion_incmp(text);
DROP FUNCTION lion_insum(text);
DROP FUNCTION lion_inpick(text);
DROP FUNCTION lion_anycmp(text, text, text, text);
DROP FUNCTION lion_inlist(int, int, int);
DROP FUNCTION lion_inplan(text);