-- A CTE that several aggregates read is planned inline (agg_cte_inline_nested), which copies its body into every reader. A sub-select of that body that
-- computes something (the keys with the most rows: two aggregates of the table) would then run once per copy: it becomes a CTE of the same WITH,
-- which PostgreSQL computes once. Only an IN sub-select that reads nothing of the query around it moves; the CTEs it reads (periods) stay the ones
-- of the WITH. The rows are the normal plan's.
CREATE EXTENSION pg_sift;
CREATE EXTENSION pg_sift_helpers;
SET max_parallel_workers_per_gather = 0;
SET client_min_messages = warning;
CREATE TABLE events (id int PRIMARY KEY, parent_id int NOT NULL, ts date NOT NULL, payload jsonb) WITH (autovacuum_enabled = false);
INSERT INTO events
SELECT g, 1 + g % 50, date '2025-01-01' + (g * 7919 % 360),
       CASE WHEN g % 13 = 0 THEN NULL
            ELSE jsonb_build_object('parts', (SELECT jsonb_agg(jsonb_build_object('cat', 'cat_' || chr(97 + (g + j) % 4), 'sub', 'catg_' || chr(97 + (g * j) % 3), 'score', (g * j) % 10))
                                              FROM generate_series(1, 1 + g % 3) j)) END
FROM generate_series(1, 30000) g;
CREATE INDEX events_col ON events USING sift_col (id, parent_id, ts, payload) WITH (paths = 'payload.parts[]{cat text, sub text, score float8}');
VACUUM (FREEZE, ANALYZE) events;

-- part_rows is read by the two arms of per_cat; filter is the sub-select of its WHERE (the keys of the request)
CREATE FUNCTION series_query(filter text) RETURNS text LANGUAGE sql AS $$
	SELECT format($q$WITH periods AS (SELECT * FROM (VALUES ('cur', daterange(DATE '2025-03-01', DATE '2025-12-31', '[]')),
	                                                        ('prev', daterange(DATE '2024-03-01', DATE '2024-12-31', '[]'))) p(period, range)),
	  part_rows AS (SELECT e.id, periods.period AS bucket_period, o.cat, o.sub, o.score
	                FROM events e JOIN periods ON e.ts <@ periods.range
	                CROSS JOIN LATERAL jsonb_to_recordset(e.payload['parts']) o(cat text, sub text, score float8)
	                WHERE o.score IS NOT NULL AND e.parent_id IN (%s)),
	  per_cat AS (
	    SELECT bucket_period, cat, sub, avg(score) AS score, count(*) AS n FROM part_rows GROUP BY bucket_period, cat, sub
	    UNION ALL
	    SELECT bucket_period, NULL, NULL, avg(score), count(*) FROM part_rows GROUP BY bucket_period)
	SELECT p.period, coalesce(x.cat, 'mhead') AS cat, x.sub, round(x.score::numeric, 6) AS value, x.n
	FROM periods p LEFT JOIN per_cat x ON x.bucket_period = p.period$q$, filter)
$$;

CREATE FUNCTION plan_nodes(q text) RETURNS TABLE (scan_nodes int, hoisted_ctes int, hoisted_reads int) LANGUAGE plpgsql AS $$
DECLARE l text;
BEGIN
	scan_nodes := 0; hoisted_ctes := 0; hoisted_reads := 0;
	FOR l IN EXECUTE 'EXPLAIN (COSTS OFF) ' || q LOOP
		IF l ~ 'Custom Scan \(SiftAgg\)' THEN scan_nodes := scan_nodes + 1; END IF;
		IF l ~ '^ *CTE sift_hoisted' THEN hoisted_ctes := hoisted_ctes + 1; END IF;
		IF l ~ 'CTE Scan on sift_hoisted' THEN hoisted_reads := hoisted_reads + 1; END IF;
	END LOOP;
	RETURN NEXT;
END $$;
CREATE FUNCTION differences(q text) RETURNS bigint LANGUAGE sql AS $$ SELECT sift_helpers.agg_differences_min_10_rows(q) $$;

-- an aggregating sub-select moves (the second reads a CTE of the WITH from inside it); a lookup of the table, a correlated one and NOT IN stay
SELECT 'aggregate' AS filter, * FROM plan_nodes(series_query('SELECT parent_id FROM events GROUP BY parent_id HAVING min(id) < 25'));
SELECT 'reads periods' AS filter, * FROM plan_nodes(series_query($f$SELECT x.parent_id FROM events x JOIN periods p ON x.ts <@ p.range GROUP BY x.parent_id HAVING min(x.id) < 35 AND count(*) > 100$f$));
SELECT 'lookup' AS filter, * FROM plan_nodes(series_query('SELECT parent_id FROM events WHERE id < 5000'));
SELECT 'correlated' AS filter, hoisted_ctes, hoisted_reads FROM plan_nodes(series_query('SELECT x.parent_id FROM events x WHERE x.id = e.id GROUP BY x.parent_id'));
SELECT 'not in' AS filter, hoisted_ctes, hoisted_reads FROM plan_nodes(replace(series_query('SELECT parent_id FROM events GROUP BY parent_id HAVING min(id) < 25'), 'e.parent_id IN', 'e.parent_id NOT IN'));
SELECT differences(series_query('SELECT parent_id FROM events GROUP BY parent_id HAVING min(id) < 25')) AS aggregate_differences,
       differences(series_query($f$SELECT x.parent_id FROM events x JOIN periods p ON x.ts <@ p.range GROUP BY x.parent_id HAVING min(x.id) < 35 AND count(*) > 100$f$)) AS periods_differences;
DROP TABLE events;
DROP FUNCTION differences(text), plan_nodes(text), series_query(text);
DROP EXTENSION pg_sift_helpers;
DROP EXTENSION pg_sift;