-- The order of a dictionary that is searched in place takes the bits of a code, ceil(log2 n), where it took 32 (column format minor 39;
-- sift.test_col_dict_unpacked writes the earlier order): the answers are the same in both, for an equality, an IN list, a prefix and a range, in segments of one table
-- written before and after, after a merge, and col_verify checks the order of every segment.
CREATE EXTENSION pg_sift_test CASCADE;
SET client_min_messages = warning;
SET sift.async_maintenance = off;
SET max_parallel_maintenance_workers = 0;

CREATE TABLE dp (id int, s text, u uuid);
INSERT INTO dp SELECT i, CASE WHEN i % 41 = 0 THEN NULL ELSE 'k' || md5(i::text) END, md5(i::text)::uuid FROM generate_series(1, 20000) i;
VACUUM (ANALYZE) dp;

SET sift.test_col_dict_unpacked = on;
CREATE INDEX dp_old ON dp USING sift_col (s sift.col_text_ops, u sift.col_uuid_ops, id sift.col_int4_ops) WITH (point_columns = 's, u', dict_lookup = 'on', write_buffer_limit = 256);
RESET sift.test_col_dict_unpacked;
CREATE INDEX dp_new ON dp USING sift_col (s sift.col_text_ops, u sift.col_uuid_ops, id sift.col_int4_ops) WITH (point_columns = 's, u', dict_lookup = 'on', write_buffer_limit = 256);
SELECT 'dp_old' AS idx, min(format) AS format FROM sift.col_segments('dp_old') WHERE kind = 'immutable' UNION ALL SELECT 'dp_new', min(format) FROM sift.col_segments('dp_new') WHERE kind = 'immutable' ORDER BY 1;
-- 19,513 strings of s and 20,000 of u take 15 bits a code, rounded up to 4 bytes, where they took 4 bytes each (the sections that follow may pad a few bytes more)
SELECT abs((SELECT sum(bytes) FROM sift.col_sections('dp_old') WHERE section = 'dict') - (SELECT sum(bytes) FROM sift.col_sections('dp_new') WHERE section = 'dict')
	   - ((4 * 19513 - ((19513 * 15 + 7) / 8 + 3) / 4 * 4) + (4 * 20000 - ((20000 * 15 + 7) / 8 + 3) / 4 * 4))) <= 8 AS saved_as_computed;
SELECT sift.col_verify('dp_old', true), sift.col_verify('dp_new', true);

CREATE FUNCTION dp_cmp(preds text, heapwhere text) RETURNS text LANGUAGE plpgsql AS $$
DECLARE a bigint; b bigint; c bigint;
BEGIN
	SELECT count(*) INTO a FROM sift._test_col_tids('dp_old', preds);
	SELECT count(*) INTO b FROM sift._test_col_tids('dp_new', preds);
	EXECUTE format('SELECT count(*) FROM dp WHERE %s', heapwhere) INTO c;
	RETURN CASE WHEN a = b AND b = c AND NOT EXISTS (SELECT tid FROM sift._test_col_tids('dp_old', preds) EXCEPT SELECT tid FROM sift._test_col_tids('dp_new', preds)) THEN 'ok' ELSE format('MISMATCH old=%s new=%s heap=%s', a, b, c) END;
END $$;
CREATE TABLE dp_q (n serial, preds text, heapwhere text);
INSERT INTO dp_q (preds, heapwhere)
SELECT format('0:eq:k%s', md5(i::text)), format('s = %L', 'k' || md5(i::text)) FROM unnest(ARRAY[1, 17, 4000, 19999, 20000, 20001, 41]) i
UNION ALL SELECT format('1:eq:%s', md5(i::text)::uuid), format('u = %L', md5(i::text)::uuid) FROM unnest(ARRAY[1, 17, 4000, 20000, 20001]) i
UNION ALL SELECT format('0:in:k%s|k%s|none', md5('3'), md5('19000')), format('s IN (%L, %L, ''none'')', 'k' || md5('3'), 'k' || md5('19000'))
UNION ALL SELECT '0:prefix:k0', 's LIKE ''k0%''' UNION ALL SELECT '0:prefix:kff', 's LIKE ''kff%'''
UNION ALL SELECT '0:ge:kc', 's >= ''kc''' UNION ALL SELECT '0:lt:k1', 's < ''k1''';
SELECT n, preds, dp_cmp(preds, heapwhere) FROM dp_q ORDER BY n;

-- a segment of each layout in one index, and a merge to one of the earlier minor 39
INSERT INTO dp SELECT i, 'k' || md5(i::text), md5(i::text)::uuid FROM generate_series(20001, 25000) i;
SELECT sift.col_promote('dp_new') > 0 AS promoted_new;
SET sift.test_col_dict_unpacked = on;
SELECT sift.col_promote('dp_old') > 0 AS promoted_old;
RESET sift.test_col_dict_unpacked;
SELECT 'dp_old' AS idx, format, count(*) AS segments FROM sift.col_segments('dp_old') WHERE kind = 'immutable' GROUP BY 2 UNION ALL SELECT 'dp_new', format, count(*) FROM sift.col_segments('dp_new') WHERE kind = 'immutable' GROUP BY 2 ORDER BY 1, 2;
INSERT INTO dp_q (preds, heapwhere) VALUES (format('0:eq:k%s', md5('22222')), format('s = %L', 'k' || md5('22222'))), (format('1:eq:%s', md5('24999')::uuid), format('u = %L', md5('24999')::uuid)), ('0:prefix:k5', 's LIKE ''k5%''');
SELECT n, dp_cmp(preds, heapwhere) FROM dp_q ORDER BY n;
SELECT sift.col_merge('dp_new', true) >= 0 AS merged;
SELECT format, count(*) AS segments FROM sift.col_segments('dp_new') WHERE kind = 'immutable' GROUP BY 1 ORDER BY 1;
SELECT sift.col_verify('dp_new', true);
SELECT n, dp_cmp(preds, heapwhere) FROM dp_q ORDER BY n;

DROP TABLE dp_q;
DROP FUNCTION dp_cmp(text, text);
DROP TABLE dp;
DROP EXTENSION pg_sift_test, pg_sift CASCADE;