-- A preserved side of an outer join (the calendar of `dates LEFT JOIN facts`) with 0, 2 and 4 workers: each participant keeps which rows of the calendar its
-- tuples matched, a worker sends them to the leader with its tables, and the leader adds the rows nobody matched once every worker has told. The rows are
-- those of the normal plan, with float sums (merged in lane order), a FILTER, a condition that matches no row, the leader out of the scan, rows in the
-- write buffer and updated tuples. mode: parallel (workers planned in the SiftAgg node), serial (SiftAgg alone), none (the normal plan).
CREATE EXTENSION pg_sift;
CREATE EXTENSION pg_sift_helpers;
SET client_min_messages = warning;
SET sift.agg_preagg = off;
CREATE TABLE dates (day date PRIMARY KEY, label text);
INSERT INTO dates SELECT d::date, 'd' || (d::date - '2026-01-01') FROM generate_series('2026-01-01'::date, '2026-02-10', '1 day') d;
CREATE TABLE facts (id int PRIMARY KEY, day date, grp int, amount float8, qty int) WITH (autovacuum_enabled = false);
INSERT INTO facts
SELECT i, '2026-01-01'::date + (i % 23), i % 7, CASE WHEN i % 29 = 0 THEN NULL ELSE ((hashint4(i) & 2147483647) % 3000)::float8 / 10 END, (hashint4(i + 2) & 2147483647) % 400
FROM generate_series(1, 30000) i;
CREATE INDEX facts_col ON facts USING sift_col (id, day, grp, amount, qty);
-- many superblocks: each worker scans several lanes (the float sums of the scan are added by lane)
CREATE TABLE laned (id int PRIMARY KEY, qty int) WITH (fillfactor = 10, autovacuum_enabled = false);
INSERT INTO laned SELECT i, (hashint4(i + 2) & 2147483647) % 400 FROM generate_series(1, 30000) i;
CREATE INDEX laned_col ON laned USING sift_col (id, qty);
CREATE TABLE dims (id int PRIMARY KEY, name text, w float8, nxt int);
INSERT INTO dims SELECT i, CASE WHEN i % 5 = 0 THEN NULL ELSE 'n' || (i % 4) END, i, (i * 3) % 8 FROM generate_series(0, 6) i;
CREATE TABLE links (id int PRIMARY KEY, tag text);
INSERT INTO links SELECT i, 't' || (i % 3) FROM generate_series(0, 6) i;
VACUUM ANALYZE facts;
VACUUM ANALYZE laned;
ANALYZE dates;
ANALYZE dims;
ANALYZE links;
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 day',      'SELECT d.day, count(f.id) AS n FROM dates d LEFT JOIN facts f ON f.day = d.day GROUP BY d.day'),
	('float sum by day',  'SELECT d.day, count(f.id), sum(f.amount), avg(f.amount) FROM dates d LEFT JOIN facts f ON f.day = d.day GROUP BY d.day'),
	('by label and grp',  'SELECT d.label, f.grp, count(*), sum(f.qty) FROM dates d LEFT JOIN facts f ON f.day = d.day GROUP BY d.label, f.grp'),
	('filter',            'SELECT d.day, count(f.id) FILTER (WHERE f.qty > 100), sum(f.amount) FILTER (WHERE f.grp = 3) FROM dates d LEFT JOIN facts f ON f.day = d.day GROUP BY d.day'),
	('condition in ON',   'SELECT d.day, count(f.id), max(f.qty) FROM dates d LEFT JOIN facts f ON f.day = d.day AND f.qty < 50 GROUP BY d.day'),
	('matches nothing',   'SELECT d.day, count(f.id), sum(f.amount) FROM dates d LEFT JOIN facts f ON f.day = d.day AND f.id < 0 GROUP BY d.day'),
	('global',            'SELECT count(*), count(f.id), sum(f.qty) FROM dates d LEFT JOIN facts f ON f.day = d.day'),
	('distinct',          'SELECT d.day, count(DISTINCT f.grp), count(DISTINCT f.qty) FROM dates d LEFT JOIN facts f ON f.day = d.day GROUP BY d.day'),
	-- the values of a column of a joined table have the worker's own ids: the leader interns them, and a value two workers saw counts once
	('distinct lookup',   'SELECT d.day, count(DISTINCT m.name), count(m.name) FROM dates d LEFT JOIN (facts f JOIN dims m ON m.id = f.grp) ON f.day = d.day GROUP BY d.day'),
	('distinct left',     'SELECT f.qty / 100, count(DISTINCT m.name), count(m.name) FROM facts f LEFT JOIN dims m ON m.id = f.grp GROUP BY 1'),
	('distinct global',   'SELECT count(DISTINCT m.name), count(m.name), count(*) FROM facts f LEFT JOIN dims m ON m.id = f.grp'),
	-- a lookup keyed by a column of an earlier one: a worker that scans a second lane starts again with the memo of that column's values
	('chained lookups',   'SELECT l.tag, count(*), sum(f.qty::float8), avg(f.qty::float8) FROM laned f JOIN dims m ON m.id = f.qty % 7 JOIN links l ON l.id = m.nxt GROUP BY l.tag'),
	-- the preserved side is keyed by a column of an earlier lookup, and the link no row reaches is added by the leader
	('chained preserved', 'SELECT l.tag, count(f.id), sum(f.qty::float8) FROM links l LEFT JOIN (laned f JOIN dims m ON m.id = f.qty % 7) ON l.id = m.nxt GROUP BY l.tag'),
	-- the rows of a CTE are the leader's: it shares them as a lookup table
	('cte lookup',        'WITH ps AS MATERIALIZED (SELECT id, nxt FROM dims WHERE id < 6) SELECT l.tag, count(*), sum(f.qty::float8), avg(f.qty::float8) FROM laned f JOIN ps ON ps.id = f.qty % 7 JOIN links l ON l.id = ps.nxt GROUP BY l.tag'),
	('distinct filter',   'SELECT f.grp % 3, count(DISTINCT m.name) FILTER (WHERE f.qty > 100), count(DISTINCT m.w) FROM facts f LEFT JOIN dims m ON m.id = f.grp - 1 GROUP BY 1');
CREATE FUNCTION rows_of(q text, push bool, workers int, leader bool DEFAULT true) RETURNS text LANGUAGE sql AS $$ SELECT sift_helpers.rows_agg_with_workers_and_leader(q, push, workers, leader) $$;
CREATE FUNCTION mode_of(q text, workers int) RETURNS text LANGUAGE sql AS $$ SELECT sift_helpers.plan_mode_workers(q, workers) $$;
SELECT s.name, mode_of(s.q, 4) AS mode,
       rows_of(s.q, true, 0) = rows_of(s.q, false, 0) AS scan_w0,
       rows_of(s.q, true, 2) = rows_of(s.q, false, 0) AS scan_w2,
       rows_of(s.q, true, 4) = rows_of(s.q, false, 0) AS scan_w4,
       rows_of(s.q, true, 3, false) = rows_of(s.q, false, 0) AS scan_w3_no_leader
FROM shapes s ORDER BY s.n;
       name        |   mode   | scan_w0 | scan_w2 | scan_w4 | scan_w3_no_leader 
-------------------+----------+---------+---------+---------+-------------------
 count by day      | parallel | t       | t       | t       | t
 float sum by day  | parallel | t       | t       | t       | t
 by label and grp  | parallel | t       | t       | t       | t
 filter            | parallel | t       | t       | t       | t
 condition in ON   | parallel | t       | t       | t       | t
 matches nothing   | none     | t       | t       | t       | t
 global            | parallel | t       | t       | t       | t
 distinct          | parallel | t       | t       | t       | t
 distinct lookup   | parallel | t       | t       | t       | t
 distinct left     | parallel | t       | t       | t       | t
 distinct global   | parallel | t       | t       | t       | t
 chained lookups   | parallel | t       | t       | t       | t
 chained preserved | parallel | t       | t       | t       | t
 cte lookup        | parallel | t       | t       | t       | t
 distinct filter   | parallel | t       | t       | t       | t
(15 rows)

-- the days no fact has (the calendar runs 18 days past the facts) are groups of the result, in the parallel scan too
SET sift.enable_agg = on;
SET max_parallel_workers_per_gather = 4;
SELECT count(*) AS days_without_facts FROM (SELECT d.day FROM dates d LEFT JOIN facts f ON f.day = d.day GROUP BY d.day HAVING count(f.id) = 0) x;
 days_without_facts 
--------------------
                 18
(1 row)

-- a scan that reads rows in the write buffer and rows of updated tuples
INSERT INTO facts SELECT 30000 + i, '2026-01-01'::date + (i % 30), i % 7, (i % 400)::float8 / 4, i % 300 FROM generate_series(1, 700) i;
UPDATE facts SET amount = amount + 0.5 WHERE id % 17 = 0;
DELETE FROM facts WHERE id % 61 = 0;
SELECT s.name, rows_of(s.q, true, 0) = rows_of(s.q, false, 0) AS dirty_w0, rows_of(s.q, true, 2) = rows_of(s.q, false, 0) AS dirty_w2,
       rows_of(s.q, true, 4) = rows_of(s.q, false, 0) AS dirty_w4
FROM shapes s ORDER BY s.n;
       name        | dirty_w0 | dirty_w2 | dirty_w4 
-------------------+----------+----------+----------
 count by day      | t        | t        | t
 float sum by day  | t        | t        | t
 by label and grp  | t        | t        | t
 filter            | t        | t        | t
 condition in ON   | t        | t        | t
 matches nothing   | t        | t        | t
 global            | t        | t        | t
 distinct          | t        | t        | t
 distinct lookup   | t        | t        | t
 distinct left     | t        | t        | t
 distinct global   | t        | t        | t
 chained lookups   | t        | t        | t
 chained preserved | t        | t        | t
 cte lookup        | t        | t        | t
 distinct filter   | t        | t        | t
(15 rows)

DROP TABLE facts, laned, dates, dims, links, shapes;
DROP EXTENSION pg_sift_helpers;
DROP EXTENSION pg_sift;
DROP FUNCTION mode_of, rows_of;