-- Parallel CREATE INDEX: every participant scans its share of the heap into sorted runs, the leader
-- merges them into the segments a serial build writes.
CREATE EXTENSION pg_sift;
SET maintenance_work_mem = '128MB';
SET max_parallel_maintenance_workers = 2;
SET enable_seqscan = off;
CREATE TABLE pb (id serial, body text) WITH (parallel_workers = 2);
INSERT INTO pb (body)
SELECT 'k' || (i % 7) || ' g' || (i % 1000) || ' common word' || (i % 50) || ' ' || repeat('filler ', i % 5)
FROM generate_series(1, 30000) i;
UPDATE pb SET body = NULL WHERE id % 1000 = 0;
DELETE FROM pb WHERE id % 13 = 0;
SET max_parallel_maintenance_workers = 0;
-- no dead version is left for the index build (a session of another test with an older snapshot makes the planner leave the index out: indcheckxmin)
CREATE TEMP TABLE pb_rows AS SELECT * FROM pb ORDER BY ctid;
TRUNCATE pb;
INSERT INTO pb SELECT * FROM pb_rows;
DROP TABLE pb_rows;
CREATE INDEX pb_ser ON pb USING sift (body);
CREATE TABLE ranked_ser AS
	SELECT id, row_number() OVER () AS pos FROM
	(SELECT id FROM pb WHERE body @@> 'g5 OR g7 OR k3 AND word9' ORDER BY sift.score(ctid) DESC, id LIMIT 50) s;
CREATE TABLE segs_ser AS SELECT gen, ndocs, sum_dl, nterms, bytes FROM sift.segments('pb_ser');
DROP INDEX pb_ser;
SET max_parallel_maintenance_workers = 2;
SET sift.test_par_workers = 0;
CREATE INDEX pb_par ON pb USING sift (body);
SELECT current_setting('sift.test_par_workers')::int AS workers;
 workers 
---------
       2
(1 row)

SELECT count(*) AS same_segments FROM sift.segments('pb_par') p JOIN segs_ser s USING (gen, ndocs, sum_dl, nterms);
 same_segments 
---------------
             1
(1 row)

SELECT count(*) AS serial_segments FROM segs_ser;
 serial_segments 
-----------------
               1
(1 row)

SELECT sift.verify('pb_par', true);
 verify 
--------
 
(1 row)

SELECT (SELECT count(*) FROM pb WHERE body @@> 'k3') = (SELECT count(*) FROM pb WHERE id % 7 = 3 AND id % 1000 <> 0) AS term_count_ok;
 term_count_ok 
---------------
 t
(1 row)

SELECT (SELECT count(*) FROM pb WHERE body @@> 'common') = (SELECT count(*) FROM pb WHERE body IS NOT NULL) AS all_docs_ok;
 all_docs_ok 
-------------
 t
(1 row)

SELECT count(*) AS ranked_diff FROM
	(SELECT id, row_number() OVER () AS pos FROM
	 (SELECT id FROM pb WHERE body @@> 'g5 OR g7 OR k3 AND word9' ORDER BY sift.score(ctid) DESC, id LIMIT 50) s
	 EXCEPT SELECT id, pos FROM ranked_ser) d;
 ranked_diff 
-------------
           0
(1 row)

-- the index keeps taking inserts and merges
INSERT INTO pb (body) SELECT 'late word' || i FROM generate_series(1, 500) i;
SELECT count(*) FROM pb WHERE body @@> 'late';
 count 
-------
   500
(1 row)

SELECT sift.merge('pb_par') >= 0 AS merged;
 merged 
--------
 t
(1 row)

SELECT sift.verify('pb_par', true);
 verify 
--------
 
(1 row)

SET sift.test_par_workers = 0;
REINDEX INDEX pb_par;
SELECT current_setting('sift.test_par_workers')::int AS workers;
 workers 
---------
       2
(1 row)

SELECT sift.verify('pb_par', true);
 verify 
--------
 
(1 row)

-- MVCC snapshot variant and an expression index
SET sift.test_par_workers = 0;
CREATE INDEX CONCURRENTLY pb_cic ON pb USING sift (body);
SELECT current_setting('sift.test_par_workers')::int AS cic_workers;
 cic_workers 
-------------
           2
(1 row)

SET sift.test_par_workers = 0;
CREATE INDEX pb_expr ON pb USING sift ((body || ' tail'));
SELECT current_setting('sift.test_par_workers')::int AS expr_workers;
 expr_workers 
--------------
            2
(1 row)

SELECT sift.verify('pb_cic', true);
 verify 
--------
 
(1 row)

SELECT sift.verify('pb_expr', true);
 verify 
--------
 
(1 row)

-- build_segments asks for an exact layout, so it builds serially
SET sift.test_par_workers = 0;
CREATE INDEX pb_one ON pb USING sift (body) WITH (build_segments = 3);
SELECT current_setting('sift.test_par_workers')::int AS workers;
 workers 
---------
       0
(1 row)

SELECT count(*) AS three_segments FROM sift.segments('pb_one');
 three_segments 
----------------
              3
(1 row)

-- a heap bigger than a quarter of shared_buffers is scanned from where a scan stopped last, but
-- the participants need ascending keys
CREATE TABLE pw (id serial, body text) WITH (parallel_workers = 2);
INSERT INTO pw (body) SELECT 'w' || (i % 100) || ' ' || repeat('x', 600) FROM generate_series(1, 120000) i;
SELECT pg_relation_size('pw') > pg_size_bytes(current_setting('shared_buffers')) / 4 AS syncscan_size;
 syncscan_size 
---------------
 t
(1 row)

RESET enable_seqscan;
SET max_parallel_workers_per_gather = 0;
SELECT count(*) FROM (SELECT id FROM pw LIMIT 60000) s;
 count 
-------
 60000
(1 row)

CREATE INDEX pw_idx ON pw USING sift (body);
SELECT sift.verify('pw_idx', true);
 verify 
--------
 
(1 row)

SELECT count(*) FROM pw WHERE body @@> 'w7';
 count 
-------
  1200
(1 row)

DROP TABLE pb, pw, ranked_ser, segs_ser;
DROP EXTENSION pg_sift;