-- A worker of a parallel aggregate sends its table to the leader as blobs of about blob_bytes (256 MB): a blob is one buffer of at most 1 GB, which a table
-- with 134 million pairs of a count(DISTINCT) passed. The limit is lowered so that a few thousand groups take several blobs: the rows are those of the normal
-- plan, and a group that alone takes more than four times the limit ends the scan with an error that names what to set.
CREATE EXTENSION pg_sift;
CREATE EXTENSION pg_sift_helpers;
SET client_min_messages = warning;
SET sift.agg_preagg = off;
CREATE TABLE facts (id int PRIMARY KEY, grp int, kind text, amount numeric(10,2), big bigint) WITH (autovacuum_enabled = false);
INSERT INTO facts SELECT i, i % 2000, (ARRAY['cat_a', 'cat_b', 'cat_c', 'cat_d'])[1 + i % 4],
       CASE WHEN i % 29 = 0 THEN NULL ELSE ((hashint4(i) & 2147483647) % 3000)::numeric / 10 END, (hashint4(i + 3) & 2147483647) % 50000 * 100003::bigint
FROM generate_series(1, 20000) i;
CREATE INDEX facts_col ON facts USING sift_col (id, grp, kind, amount, big);
VACUUM ANALYZE facts;
SET parallel_setup_cost = 0;
SET parallel_tuple_cost = 0;
SET min_parallel_table_scan_size = 0;
SET min_parallel_index_scan_size = 0;
SET seq_page_cost = 10000;
CREATE TABLE shapes (n serial, name text, q text);
INSERT INTO shapes (name, q) VALUES
	('count by group', 'SELECT grp, count(DISTINCT big), count(*) FROM facts GROUP BY grp'),
	('count and sum',  'SELECT grp, count(DISTINCT amount), sum(DISTINCT amount) FROM facts GROUP BY grp'),
	('few groups',     'SELECT kind, count(DISTINCT big), count(DISTINCT grp) FROM facts GROUP BY kind');
CREATE FUNCTION rows_of(q text, workers int) RETURNS text LANGUAGE sql AS $$ SELECT sift_helpers.rows_agg_with_workers_and_leader(q, true, workers, false) $$;
CREATE FUNCTION oracle(q text) RETURNS text LANGUAGE sql AS $$ SELECT sift_helpers.rows_agg_with_workers_and_leader(q, false, 0, true) $$;
-- the groups of the table go in blobs of about 4 kB (a group takes 100 bytes of state and 8 for each pair)
SET sift.test_limits = 'blob_bytes=4096';
SELECT s.name, sift_helpers.plan_mode_workers(s.q, 2) AS mode, rows_of(s.q, 2) = oracle(s.q) AS same_w2, rows_of(s.q, 3) = oracle(s.q) AS same_w3
FROM shapes s WHERE s.name <> 'few groups' ORDER BY s.n;
      name      |   mode   | same_w2 | same_w3 
----------------+----------+---------+---------
 count by group | parallel | t       | t
 count and sum  | parallel | t       | t
(2 rows)

-- the groups of 'few groups' have thousands of pairs each: past four times the limit
DO $$
DECLARE msg text; hint text;
BEGIN
	PERFORM rows_of((SELECT q FROM shapes WHERE name = 'few groups'), 2);
	RAISE NOTICE 'no error';
EXCEPTION WHEN program_limit_exceeded THEN
	GET STACKED DIAGNOSTICS msg = MESSAGE_TEXT, hint = PG_EXCEPTION_HINT;
	RAISE WARNING 'SQLSTATE % names the group: % hint: %', SQLSTATE, msg ~ 'partial result of a group of a parallel aggregate takes about', hint;
END $$;
WARNING:  SQLSTATE 54000 names the group: t hint: Set max_parallel_workers_per_gather to 0 for this query, or sift.enable_agg to off.
RESET sift.test_limits;
DROP FUNCTION rows_of(text, int), oracle(text);
DROP TABLE facts, shapes;
DROP EXTENSION pg_sift_helpers;
DROP EXTENSION pg_sift;