-- The rewrite of the overlapping runs of a build is paced: a pass (sift.col_maintain, a VACUUM's worker) rewrites about sift.col_auto_merge_pass_mb of them, a window
-- of the runs that meet in the key at a time, and the next pass goes on from the directory this one left. Each pass leaves the index valid and equal to the heap's,
-- changes made between the passes are kept, and the passes end with the key range cut into partitions: a key in one unit.
CREATE EXTENSION pg_sift;
CREATE EXTENSION pg_sift_helpers;
SET client_min_messages = warning;
SET max_parallel_workers_per_gather = 0;
SET max_parallel_maintenance_workers = 0;
SET sift.async_maintenance = off;
SET sift.vacuum_rewrite_limit = 1;
SET sift.col_auto_merge_pass_mb = 2;
SET sift.test_col_segment_kb = 7000;
SET sift.col_build_key_ranges = off;
CREATE TABLE kp (id int, tenant_id int, d date, amount int, pad text) WITH (autovacuum_enabled = false);
INSERT INTO kp SELECT i, (i::bigint * 7919 % 400)::int, date '2026-01-01' + (i * 31) % 300, i % 100, md5(i::text) || md5((i + 1)::text) || md5((i + 2)::text) FROM generate_series(1, 337920) i;
VACUUM (FREEZE, ANALYZE) kp;
CREATE INDEX kp_idx ON kp USING sift_col (id, tenant_id, d, amount, pad) WITH (key_cut = 'tenant_id', dict_lookup = off);
RESET sift.test_col_segment_kb;
RESET sift.col_build_key_ranges;
-- eight runs, each as large as a partition and holding rows of every tenant
SELECT count(*) AS runs, min(bytes) > 524288 AS partition_sized FROM sift.col_segments('kp_idx') WHERE kind = 'immutable';
 runs | partition_sized 
------+-----------------
    8 | t
(1 row)

CREATE FUNCTION kp_same() RETURNS bool LANGUAGE plpgsql AS $$
DECLARE a text; b text; q text;
BEGIN
	FOREACH q IN ARRAY ARRAY['SELECT count(*), sum(amount), min(id), max(d) FROM kp WHERE tenant_id = 42', 'SELECT tenant_id, count(*), sum(amount) FROM kp WHERE tenant_id IN (3, 77, 399) GROUP BY 1 ORDER BY 1', 'SELECT count(*), sum(amount) FROM kp'] LOOP
		SET LOCAL enable_seqscan = off; SET LOCAL enable_bitmapscan = off; SET LOCAL enable_indexscan = off;
		EXECUTE format('SELECT string_agg(x::text, E''\n'') FROM (%s) x', q) INTO a;
		SET LOCAL sift.enable_agg = off; SET LOCAL enable_seqscan = on;
		EXECUTE format('SELECT string_agg(x::text, E''\n'') FROM (%s) x', q) INTO b;
		SET LOCAL sift.enable_agg = on;
		IF a IS DISTINCT FROM b THEN RETURN false; END IF;
	END LOOP;
	RETURN true;
END $$;
CREATE FUNCTION kp_units() RETURNS int LANGUAGE sql AS $$
	SELECT substring(sift_helpers.plan_units_lines('SELECT count(*), sum(amount) FROM kp WHERE tenant_id = 42') from 'Units: (\d+) read')::int $$;
SET enable_seqscan = off; SET enable_bitmapscan = off; SET enable_indexscan = off;
SELECT kp_units() AS units_before, kp_same() AS same_before;
 units_before | same_before 
--------------+-------------
            8 | t
(1 row)

-- the passes: the bytes a pass replaces stay within the limit and one window over it, the answers are the heap's after each, and changes between them are kept
CREATE TABLE kp_log (pass int, replaced_bytes bigint, bounded boolean, same boolean);
CREATE TABLE kp_before (gen bigint, bytes bigint);
DO $$
DECLARE
	p int := 0; gone bigint; maxb bigint; changed boolean;
BEGIN
	LOOP
		TRUNCATE kp_before;
		INSERT INTO kp_before SELECT gen, bytes FROM sift.col_segments('kp_idx') WHERE kind = 'immutable';
		PERFORM sift.col_maintain('kp_idx');
		SELECT coalesce(sum(bytes), 0), coalesce(max(bytes), 0) INTO gone, maxb FROM kp_before WHERE gen NOT IN (SELECT gen FROM sift.col_segments('kp_idx') WHERE kind = 'immutable');
		EXIT WHEN gone = 0 OR p >= 60;
		p := p + 1;
		INSERT INTO kp_log VALUES (p, gone, gone <= 2 * 1048576 + 2 * maxb, kp_same());
		IF p = 2 THEN
			INSERT INTO kp SELECT 1000000 + g, g % 400, date '2027-01-01', 7, md5(g::text) FROM generate_series(1, 3000) g;
			UPDATE kp SET amount = amount + 1 WHERE id % 50 = 3;
			DELETE FROM kp WHERE id % 97 = 5;
		END IF;
	END LOOP;
END $$;
SELECT count(*) > 3 AS several_passes, bool_and(bounded) AS within_the_limit, bool_and(same) AS rows_equal_after_each_pass FROM kp_log;
 several_passes | within_the_limit | rows_equal_after_each_pass 
----------------+------------------+----------------------------
 t              | t                | t
(1 row)

SELECT sift.col_verify('kp_idx', true);
 col_verify 
------------
 
(1 row)

SELECT kp_units() AS units_after, kp_same() AS same_after;
 units_after | same_after 
-------------+------------
           1 | t
(1 row)

SELECT sift.col_maintain('kp_idx') IS NOT NULL AS asked_again;
 asked_again 
-------------
 t
(1 row)

SELECT kp_same() AS same_end;
 same_end 
----------
 t
(1 row)

DROP TABLE kp, kp_log, kp_before;
DROP FUNCTION kp_same(), kp_units();
DROP EXTENSION pg_sift_helpers;
DROP EXTENSION pg_sift;