-- The workers of a parallel aggregate read the pages of the write buffer of a sift_col index; the rows a transaction has inserted after its first one are held back
-- in the backend until it reads the index (col/wbuf/pending.c), and a worker has none of its own to flush. With the leader not scanning, nothing flushed them: the
-- transaction's own rows were missing from the answer of the workers, from the second transaction of a session on (the first read of a session flushed them).
CREATE EXTENSION pg_sift;
SET client_min_messages = warning;
CREATE TABLE own (id int PRIMARY KEY, g int NOT NULL, v int NOT NULL) WITH (autovacuum_enabled = false);
INSERT INTO own SELECT i, i % 5, i FROM generate_series(1, 20000) i;
CREATE INDEX own_c ON own USING sift_col (id, g, v);
VACUUM (FREEZE, ANALYZE) own;
SET max_parallel_workers_per_gather = 2;
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 sift.agg_par_min_rows = 0;
SET parallel_leader_participation = off;
-- the same aggregate, in the transaction that inserted rows, under the workers and under the heap's plan (sift.enable_agg off); three transactions, each rolled back
BEGIN;
INSERT INTO own SELECT i, i % 5, i FROM generate_series(20001, 20100) i;
SELECT count(*), sum(v) FROM own;
 count |    sum    
-------+-----------
 20100 | 202015050
(1 row)

SELECT g, count(*), sum(v) FROM own GROUP BY g ORDER BY g;
 g | count |   sum    
---+-------+----------
 0 |  4020 | 40411050
 1 |  4020 | 40394970
 2 |  4020 | 40398990
 3 |  4020 | 40403010
 4 |  4020 | 40407030
(5 rows)

ROLLBACK;
BEGIN;
INSERT INTO own SELECT i, i % 5, i FROM generate_series(20001, 20100) i;
SELECT count(*), sum(v) FROM own;
 count |    sum    
-------+-----------
 20100 | 202015050
(1 row)

SELECT g, count(*), sum(v) FROM own GROUP BY g ORDER BY g;
 g | count |   sum    
---+-------+----------
 0 |  4020 | 40411050
 1 |  4020 | 40394970
 2 |  4020 | 40398990
 3 |  4020 | 40403010
 4 |  4020 | 40407030
(5 rows)

SET LOCAL sift.enable_agg = off;
SELECT count(*), sum(v) FROM own;
 count |    sum    
-------+-----------
 20100 | 202015050
(1 row)

ROLLBACK;
BEGIN;
INSERT INTO own SELECT i, i % 5, i FROM generate_series(20001, 20100) i;
INSERT INTO own SELECT i, i % 5, i FROM generate_series(20101, 20200) i;
SELECT count(*), sum(v) FROM own;
 count |    sum    
-------+-----------
 20200 | 204030100
(1 row)

ROLLBACK;
SELECT count(*), sum(v) FROM own;
 count |    sum    
-------+-----------
 20000 | 200010000
(1 row)

-- the plan of a child that a worker starts per combination reads an index of its own table, and the rows the transaction holds back for it are not in a page either
CREATE TABLE zones (id int, name text) WITH (autovacuum_enabled = false);
INSERT INTO zones SELECT i, 'z' || i % 7 FROM generate_series(0, 59) i;
CREATE INDEX zones_c ON zones USING sift_col (id, name);
CREATE TABLE zev (id int PRIMARY KEY, zone_id int) WITH (autovacuum_enabled = false);
INSERT INTO zev SELECT i, i % 65 FROM generate_series(1, 12000) i;
CREATE INDEX zev_c ON zev USING sift_col (id, zone_id);
VACUUM (FREEZE, ANALYZE) zones, zev;
SET enable_seqscan = off;
EXPLAIN (COSTS OFF) SELECT s.name, count(*) FROM zev e JOIN LATERAL (SELECT name FROM zones z WHERE z.id = e.zone_id LIMIT 1) s ON true GROUP BY 1;
                       QUERY PLAN                       
--------------------------------------------------------
 Custom Scan (SiftAgg)
   Index: zev_c
   Group Keys: zone_id: s.name
   Pushed Aggregates: count(*)
   Lookup Joins: s (subquery)
   Name: s (subquery)
   ->  Limit
         ->  Index Scan using zones_c on zones z
               Index Cond: (id = (InitPlan input).col1)
   Workers Planned: 2
(10 rows)

BEGIN;
INSERT INTO zones VALUES (60, 'new');
INSERT INTO zones SELECT i, 'new' FROM generate_series(61, 64) i;
SELECT s.name, count(*) FROM zev e JOIN LATERAL (SELECT name FROM zones z WHERE z.id = e.zone_id LIMIT 1) s ON true GROUP BY 1 ORDER BY 1;
 name | count 
------+-------
 new  |   920
 z0   |  1661
 z1   |  1662
 z2   |  1662
 z3   |  1662
 z4   |  1478
 z5   |  1478
 z6   |  1477
(8 rows)

ROLLBACK;
BEGIN;
INSERT INTO zones VALUES (60, 'new');
INSERT INTO zones SELECT i, 'new' FROM generate_series(61, 64) i;
SELECT s.name, count(*) FROM zev e JOIN LATERAL (SELECT name FROM zones z WHERE z.id = e.zone_id LIMIT 1) s ON true GROUP BY 1 ORDER BY 1;
 name | count 
------+-------
 new  |   920
 z0   |  1661
 z1   |  1662
 z2   |  1662
 z3   |  1662
 z4   |  1478
 z5   |  1478
 z6   |  1477
(8 rows)

ROLLBACK;
DROP TABLE zones, zev;
-- the rows held back for the index of a partition
SET sift.agg_preagg = off;
SET seq_page_cost = 1000;
SET random_page_cost = 1000;
CREATE TABLE pt (id int, g int NOT NULL, v int NOT NULL) PARTITION BY RANGE (id);
CREATE TABLE pt1 PARTITION OF pt FOR VALUES FROM (0) TO (10000) WITH (fillfactor = 10, autovacuum_enabled = false);
CREATE TABLE pt2 PARTITION OF pt FOR VALUES FROM (10000) TO (20000) WITH (fillfactor = 10, autovacuum_enabled = false);
INSERT INTO pt SELECT i, i % 5, i FROM generate_series(1, 19999) i;
CREATE INDEX pt_c ON pt USING sift_col (id, g, v);
VACUUM (FREEZE, ANALYZE) pt;
EXPLAIN (COSTS OFF) SELECT g, count(*), sum(v) FROM pt GROUP BY g;
              QUERY PLAN               
---------------------------------------
 Custom Scan (SiftAgg)
   Index: (partitions)
   Partitions: pt1, pt2
   Group Keys: g
   Pushed Aggregates: count(*), sum(v)
   Workers Planned: 2
(6 rows)

BEGIN;
INSERT INTO pt VALUES (9000, 1, 9000);
INSERT INTO pt SELECT 10000 + i, 1, i FROM generate_series(1, 100) i;
INSERT INTO pt SELECT 100 + i, 1, i FROM generate_series(1, 100) i;
SELECT g, count(*), sum(v) FROM pt GROUP BY g ORDER BY g;
 g | count |   sum    
---+-------+----------
 0 |  3999 | 39990000
 1 |  4201 | 40013100
 2 |  4000 | 39998000
 3 |  4000 | 40002000
 4 |  4000 | 40006000
(5 rows)

ROLLBACK;
DROP TABLE pt;
DROP TABLE own;
DROP EXTENSION pg_sift;