-- A merge of segments by several processes: the participants write ranges of the terms straight from the dictionaries of the
-- input segments and the process of the merge stitches them. The merged segment is the one a merge by one process writes, byte for byte.
CREATE EXTENSION pg_sift;
CREATE EXTENSION pageinspect;
SET maintenance_work_mem = '64MB';
SET sift.test_range_bytes = 4000;
CREATE FUNCTION idx_digest(r regclass) RETURNS text LANGUAGE sql AS $$
	SELECT md5(string_agg(substring(get_raw_page(r::text, n::int) FROM 11), ''::bytea ORDER BY n))
	FROM generate_series(0, pg_relation_size(r) / 8192 - 1) n
$$;
CREATE TABLE mr (id int, body text, cat int, created date);
-- two indexes made alike, the segments of one merged by the backend alone, the other with w workers
CREATE FUNCTION prep(opts text, incl text DEFAULT '') RETURNS void LANGUAGE plpgsql AS $$
BEGIN
	DROP INDEX IF EXISTS mr_a;
	DROP INDEX IF EXISTS mr_b;
	TRUNCATE mr;
	EXECUTE format('CREATE INDEX mr_a ON mr USING sift (body) %s WITH (%s)', incl, opts);
	EXECUTE format('CREATE INDEX mr_b ON mr USING sift (body) %s WITH (%s)', incl, opts);
	FOR b IN 1..4 LOOP
		INSERT INTO mr (id, body, cat, created)
		SELECT i, 'k' || (i % 7) || ' g' || (i % 300) || ' z' || (i * i % 997) || ' common word' || (i % 50) || ' ' || repeat('filler ', i % 5)
			|| CASE WHEN i % 3 = 0 THEN ' rare' || (i % 500) ELSE '' END,
			CASE WHEN i % 11 = 0 THEN NULL ELSE i % 97 END, DATE '2026-01-01' + i % 300
		FROM generate_series(b * 10000 - 9999, b * 10000) i;
		PERFORM sift.promote('mr_a'), sift.promote('mr_b');
	END LOOP;
END
$$;
CREATE FUNCTION merged(w int, force bool DEFAULT false) RETURNS void LANGUAGE plpgsql AS $$
BEGIN
	PERFORM set_config('max_parallel_maintenance_workers', '0', false);
	PERFORM sift.merge('mr_a', force);
	PERFORM set_config('max_parallel_maintenance_workers', w::text, false);
	PERFORM set_config('sift.test_par_workers', '0', false);
	PERFORM set_config('sift.test_range_pieces', '0', false);
	PERFORM sift.merge('mr_b', force);
END
$$;
CREATE FUNCTION same() RETURNS TABLE (same_bytes bool, workers int, ranges int, segments_left int) LANGUAGE plpgsql AS $$
BEGIN
	PERFORM sift.verify('mr_a', true);
	PERFORM sift.verify('mr_b', true);
	same_bytes := idx_digest('mr_a') = idx_digest('mr_b');
	workers := current_setting('sift.test_par_workers')::int;
	ranges := current_setting('sift.test_range_pieces')::int;
	SELECT count(*)::int INTO segments_left FROM sift.segments('mr_b') WHERE kind = 'immutable';
	RETURN NEXT;
END
$$;
SELECT prep('merge_fanin = 4');
NOTICE:  index "mr_a" does not exist, skipping
NOTICE:  index "mr_b" does not exist, skipping
 prep 
------
 
(1 row)

SELECT merged(1);
 merged 
--------
 
(1 row)

SELECT * FROM same();
 same_bytes | workers | ranges | segments_left 
------------+---------+--------+---------------
 t          |       1 |      8 |             1
(1 row)

SELECT prep('merge_fanin = 4');
 prep 
------
 
(1 row)

SELECT merged(2);
 merged 
--------
 
(1 row)

SELECT * FROM same();
 same_bytes | workers | ranges | segments_left 
------------+---------+--------+---------------
 t          |       2 |     12 |             1
(1 row)

SELECT prep('merge_fanin = 4');
 prep 
------
 
(1 row)

SELECT merged(4);
 merged 
--------
 
(1 row)

SELECT * FROM same();
 same_bytes | workers | ranges | segments_left 
------------+---------+--------+---------------
 t          |       4 |     20 |             1
(1 row)

-- INCLUDE columns: the values of every doc are read from the heap by the process of the merge
SELECT prep('merge_fanin = 4', 'INCLUDE (cat, created)');
 prep 
------
 
(1 row)

SELECT merged(1);
 merged 
--------
 
(1 row)

SELECT * FROM same();
 same_bytes | workers | ranges | segments_left 
------------+---------+--------+---------------
 t          |       1 |      8 |             1
(1 row)

SELECT prep('merge_fanin = 4', 'INCLUDE (cat, created)');
 prep 
------
 
(1 row)

SELECT merged(2);
 merged 
--------
 
(1 row)

SELECT * FROM same();
 same_bytes | workers | ranges | segments_left 
------------+---------+--------+---------------
 t          |       2 |     12 |             1
(1 row)

SELECT prep('merge_fanin = 4', 'INCLUDE (cat, created)');
 prep 
------
 
(1 row)

SELECT merged(4);
 merged 
--------
 
(1 row)

SELECT * FROM same();
 same_bytes | workers | ranges | segments_left 
------------+---------+--------+---------------
 t          |       4 |     20 |             1
(1 row)

-- the merge of every segment, and an index without positions
SELECT prep('merge_fanin = 8');
 prep 
------
 
(1 row)

SELECT merged(3, true);
 merged 
--------
 
(1 row)

SELECT * FROM same();
 same_bytes | workers | ranges | segments_left 
------------+---------+--------+---------------
 t          |       3 |     16 |             1
(1 row)

SELECT prep('merge_fanin = 4, positions = off');
 prep 
------
 
(1 row)

SELECT merged(2);
 merged 
--------
 
(1 row)

SELECT * FROM same();
 same_bytes | workers | ranges | segments_left 
------------+---------+--------+---------------
 t          |       2 |     12 |             1
(1 row)

-- segments with vacuumed docs, merged by the VACUUM of the backend alone and then with workers
SELECT prep('merge_fanin = 8');
 prep 
------
 
(1 row)

DELETE FROM mr WHERE id % 7 = 0;
SET max_parallel_maintenance_workers = 0;
VACUUM mr;
SELECT idx_digest('mr_a') AS serial_digest \gset
SELECT prep('merge_fanin = 8');
 prep 
------
 
(1 row)

DELETE FROM mr WHERE id % 7 = 0;
SET max_parallel_maintenance_workers = 2;
SET sift.test_par_workers = 0;
SET sift.test_range_pieces = 0;
VACUUM mr;
RESET max_parallel_maintenance_workers;
SELECT idx_digest('mr_a') = :'serial_digest' AS same_bytes, current_setting('sift.test_par_workers')::int AS workers,
	current_setting('sift.test_range_pieces')::int AS ranges;
 same_bytes | workers | ranges 
------------+---------+--------
 t          |       2 |     24
(1 row)

-- a segment of an older format
SET sift.test_segfmt = 2;
SELECT prep('merge_fanin = 4');
 prep 
------
 
(1 row)

RESET sift.test_segfmt;
SELECT merged(2);
 merged 
--------
 
(1 row)

SELECT * FROM same();
 same_bytes | workers | ranges | segments_left 
------------+---------+--------+---------------
 t          |       2 |     12 |             1
(1 row)

-- a participant, a worker or the process of the merge, that fails in a range: the merge ends, nothing changed, the next one is whole
DROP INDEX IF EXISTS mr_b;
DROP INDEX IF EXISTS mr_a;
TRUNCATE mr;
CREATE INDEX mr_b ON mr USING sift (body) WITH (merge_fanin = 4);
INSERT INTO mr (body) SELECT 'k' || (i % 7) || ' g' || (i % 300) || ' common' FROM generate_series(1, 40000) i;
SELECT sift.promote('mr_b');
 promote 
---------
       1
(1 row)

INSERT INTO mr (body) SELECT 'k' || (i % 7) || ' g' || (i % 300) || ' common' FROM generate_series(1, 40000) i;
SELECT sift.promote('mr_b');
 promote 
---------
       1
(1 row)

INSERT INTO mr (body) SELECT 'k' || (i % 7) || ' g' || (i % 300) || ' common' FROM generate_series(1, 40000) i;
SELECT sift.promote('mr_b');
 promote 
---------
       1
(1 row)

INSERT INTO mr (body) SELECT 'k' || (i % 7) || ' g' || (i % 300) || ' common' FROM generate_series(1, 40000) i;
SELECT sift.promote('mr_b');
 promote 
---------
       1
(1 row)

SELECT count(*) AS segments FROM sift.segments('mr_b') WHERE kind = 'immutable';
 segments 
----------
        4
(1 row)

SET max_parallel_maintenance_workers = 3;
SET sift.test_fail_merge = 1;
SET sift.test_fail_who = 2;
SELECT sift.merge('mr_b');
ERROR:  canceling statement due to sift.test_fail_merge
CONTEXT:  parallel worker
SET sift.test_fail_who = 1;
SELECT sift.merge('mr_b');
ERROR:  canceling statement due to sift.test_fail_merge
RESET sift.test_fail_merge;
RESET sift.test_fail_who;
SELECT count(*) AS segments_after_failures FROM sift.segments('mr_b') WHERE kind = 'immutable';
 segments_after_failures 
-------------------------
                       4
(1 row)

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

SELECT count(*) AS workers_left FROM pg_stat_activity WHERE backend_type = 'parallel worker';
 workers_left 
--------------
            0
(1 row)

SET sift.test_par_workers = 0;
SELECT sift.merge('mr_b') AS merged;
 merged 
--------
      1
(1 row)

SELECT current_setting('sift.test_par_workers')::int > 0 AS parallel;
 parallel 
----------
 t
(1 row)

SELECT count(*) AS segments_after FROM sift.segments('mr_b') WHERE kind = 'immutable';
 segments_after 
----------------
              1
(1 row)

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

SELECT count(*) FROM mr WHERE body @@> 'common';
 count  
--------
 160000
(1 row)

DROP TABLE mr;
DROP FUNCTION same();
DROP FUNCTION merged(int, bool);
DROP FUNCTION prep(text, text);
DROP FUNCTION idx_digest(regclass);
DROP EXTENSION pageinspect;
DROP EXTENSION pg_sift;