-- key_columns = 'auto' is decided again when every segment of the index is merged: after churn (new row versions in the tail of the heap) a unit holds many keys, the
-- cube no longer holds the column in it, and the merge adds the key list that a REINDEX would have chosen. A column the cube holds in 40% of the units or more is not
-- added, an explicit key_columns is never touched, and a second merge changes nothing. The lists that the build wrote for a column the cube holds (the option
-- does not name it) are written again by a merge, so that the units of the rewrite do not read every unit whose zone holds a key.
SET client_min_messages = warning;
CREATE EXTENSION pg_sift_test CASCADE;
CREATE EXTENSION pg_sift_helpers;
SET sift.col_auto_merge = off;
SET max_parallel_workers_per_gather = 0;

CREATE FUNCTION mk_opt(idx regclass) RETURNS text LANGUAGE sql AS $$ SELECT coalesce(array_to_string(reloptions, ','), '') FROM pg_class WHERE oid = idx $$;
CREATE FUNCTION mk_held(idx regclass) RETURNS text LANGUAGE sql AS $$ SELECT held FROM sift.index_advice(idx) WHERE leaf = 'set_id' $$;
CREATE FUNCTION mk_rows(q text, push bool) RETURNS text LANGUAGE sql AS $$ SELECT sift_helpers.rows_agg(q, push) $$;

-- clustered by set_id: about 6 owners in a unit, which the cube holds
CREATE TABLE mk (id int, set_id int, v int, pad text) WITH (autovacuum_enabled = off);
INSERT INTO mk SELECT i, i / 500, i % 100, repeat('p', 200) FROM generate_series(0, 29999) i;
VACUUM (FREEZE, DISABLE_PAGE_SKIPPING) mk;
CREATE INDEX mk_auto ON mk USING sift_col (id, set_id, v);
CREATE INDEX mk_explicit ON mk USING sift_col (id, set_id, v) WITH (key_columns = 'v');
CREATE INDEX mk_off ON mk USING sift_col (id, set_id, v) WITH (key_columns = 'off');
SELECT mk_opt('mk_auto') AS auto, mk_held('mk_auto') AS held;

-- two rounds of non-HOT updates that scatter the owners, each promoted into its own segment
UPDATE mk SET set_id = (id::bigint * 7919 % 600)::int;
VACUUM mk;
SELECT sift.col_promote('mk_auto') > 0 AS promoted_auto, sift.col_promote('mk_explicit') > 0 AS promoted_explicit, sift.col_promote('mk_off') > 0 AS promoted_off;
UPDATE mk SET set_id = (id::bigint * 104729 % 600)::int;
VACUUM mk;
SELECT sift.col_promote('mk_auto') > 0 AS promoted_auto, sift.col_promote('mk_explicit') > 0 AS promoted_explicit, sift.col_promote('mk_off') > 0 AS promoted_off;

SELECT sift.col_merge('mk_auto', true) > 0 AS merged_auto, sift.col_merge('mk_explicit', true) > 0 AS merged_explicit, sift.col_merge('mk_off', true) > 0 AS merged_off;
SELECT mk_opt('mk_auto') AS auto, mk_held('mk_auto') AS held;
SELECT mk_opt('mk_explicit') AS explicit, mk_opt('mk_off') AS off;
-- nothing alternates: the next merge keeps the choice
SELECT sift.col_merge('mk_auto', true) >= 0 AS merged_again;
SELECT mk_opt('mk_auto') AS auto, mk_held('mk_auto') AS held;

-- the answers are the plain plan's
SELECT mk_rows(q, true) = mk_rows(q, false) AS same FROM (VALUES
	('SELECT count(DISTINCT id), count(*) FROM mk WHERE set_id = 5'),
	('SELECT set_id, count(DISTINCT id) FROM mk WHERE set_id IN (3, 77, 410) GROUP BY 1')) t(q);

-- the lists of a column the cube holds survive a merge of every segment
CREATE FUNCTION mk_lists(idx regclass) RETURNS bool LANGUAGE sql AS $$ SELECT coalesce(sum(bytes), 0) > 0 FROM sift.col_sections(idx) WHERE section = 'sets' $$;
CREATE TABLE mk2 (id int, set_id int, v int, pad text) WITH (autovacuum_enabled = off);
INSERT INTO mk2 SELECT i, i / 500, i % 100, repeat('p', 200) FROM generate_series(0, 29999) i;
VACUUM (FREEZE, DISABLE_PAGE_SKIPPING) mk2;
CREATE INDEX mk2_auto ON mk2 USING sift_col (id, set_id, v);
SELECT mk_opt('mk2_auto') AS auto, mk_lists('mk2_auto') AS lists;
INSERT INTO mk2 SELECT i, i / 500, i % 100, repeat('p', 200) FROM generate_series(30000, 59999) i;
VACUUM mk2;
SELECT sift.col_promote('mk2_auto') > 0 AS promoted;
SELECT sift.col_merge('mk2_auto', true) > 0 AS merged;
SELECT mk_opt('mk2_auto') AS auto, mk_lists('mk2_auto') AS lists;
SELECT mk_rows(q, true) = mk_rows(q, false) AS same FROM (VALUES
	('SELECT count(DISTINCT id), count(*) FROM mk2 WHERE set_id = 5'),
	('SELECT set_id, count(DISTINCT id) FROM mk2 WHERE set_id IN (3, 77, 110) GROUP BY 1')) t(q);

-- and a promotion writes them too: a column the cube holds (400 owners in bursts of 500 rows, scattered over the units) is not in the option, and the segment of a
-- promotion reads as few units for a key as the one a REINDEX writes
CREATE FUNCTION mk_units(q text) RETURNS int LANGUAGE sql AS $$ SELECT sift_helpers.units_read_scans_off(q) $$;
CREATE TABLE mk3 (id int, set_id int, v int, pad text) WITH (autovacuum_enabled = off);
ALTER TABLE mk3 ALTER COLUMN pad SET STORAGE PLAIN;
INSERT INTO mk3 SELECT i, (hashint4(i / 500) & 2147483647) % 400, i % 1000, repeat('p', 100) FROM generate_series(1, 150000) i;
VACUUM (FREEZE, ANALYZE) mk3;
CREATE INDEX mk3_auto ON mk3 USING sift_col (id, set_id, v);
INSERT INTO mk3 SELECT i, (hashint4(i / 500) & 2147483647) % 400, i % 1000, repeat('p', 100) FROM generate_series(150001, 300000) i;
SELECT sift.col_promote('mk3_auto') > 0 AS promoted;
CREATE TEMP TABLE mk3_units AS SELECT mk_units('SELECT count(*), sum(v) FROM mk3 WHERE set_id = 5 AND v % 7 = 0') AS promoted;
REINDEX INDEX mk3_auto;
SELECT mk_opt('mk3_auto') AS auto, promoted <= mk_units('SELECT count(*), sum(v) FROM mk3 WHERE set_id = 5 AND v % 7 = 0') AS promotion_prunes_like_reindex FROM mk3_units;
SELECT mk_rows(q, true) = mk_rows(q, false) AS same FROM (VALUES
	('SELECT count(*), sum(v) FROM mk3 WHERE set_id = 5 AND v % 7 = 0')) t(q);

SET client_min_messages = warning;
DROP TABLE mk, mk2, mk3;
DROP FUNCTION mk_opt(regclass), mk_held(regclass), mk_rows(text, bool), mk_lists(regclass), mk_units(text);
DROP EXTENSION pg_sift_helpers;
DROP EXTENSION pg_sift_test, pg_sift CASCADE;