-- The write buffer of a text column is sorted for the global order like a segment's entries, whatever its size (here 6,000 rows): under a collation that
-- orders by bytes its rows are cut into ranges by their first 8 bytes and only the ranges a read reaches are sorted. The strings share their first 8 bytes ('customer-...'), are shorter than 8,
-- hold accents (bytes past 0x7f), repeat, are empty or NULL. The answers are the heap's, in both directions, NULLS FIRST and LAST, with OFFSET and bounds, after rows of the buffer are deleted,
-- updated and rolled back, and after a promotion. A column under a locale (en-x-icu) sorts its buffer whole up to 16,384 rows (more reads the units): its answers are the heap's too.
CREATE EXTENSION pg_sift_test CASCADE;
CREATE EXTENSION pg_sift_helpers;
SET sift.async_maintenance = off;
SET max_parallel_workers_per_gather = 0;
SET client_min_messages = warning;
\set VERBOSITY terse

CREATE FUNCTION tx_str(i int) RETURNS text LANGUAGE sql IMMUTABLE AS $$
	SELECT CASE i % 13 WHEN 0 THEN '' WHEN 1 THEN 'ab' ELSE '' END
		|| (ARRAY['customer-', 'customer-0', 'order-', 'Zeta-', 'été-', 'a', 'zz'])[1 + (i * 7) % 7] || lpad(((i::bigint * 7919) % 3000)::text, 4, '0') $$;
CREATE TABLE tx (id int, c text COLLATE "C", l text COLLATE "en-x-icu", pad int);
INSERT INTO tx SELECT i, CASE WHEN i % 41 = 0 THEN NULL ELSE tx_str(i) END, CASE WHEN i % 43 = 0 THEN NULL ELSE tx_str(i * 3) END, 0 FROM generate_series(1, 8000) i;
CREATE INDEX tx_i ON tx USING sift_col (id sift.col_int4_ops, c sift.col_text_ops, l sift.col_text_ops, pad sift.col_int4_ops) WITH (order_array_columns = 'c, l', write_buffer_limit = 65536);
VACUUM tx;
INSERT INTO tx SELECT 8000 + i, CASE WHEN i % 41 = 0 THEN NULL ELSE tx_str(i * 5) END, CASE WHEN i % 43 = 0 THEN NULL ELSE tx_str(i * 11) END, 0 FROM generate_series(1, 6000) i;
SELECT sum(rows) > 4096 AS buffer_over_the_old_limit FROM sift.col_segments('tx_i') WHERE kind = 'buffer';
ANALYZE tx;

CREATE FUNCTION tx_chk(q text) RETURNS text LANGUAGE sql AS $$ SELECT sift_helpers.topn_check_ordered_rows_with_nulls(q) $$;
CREATE FUNCTION tx_all() RETURNS TABLE (col text, result text, n bigint) LANGUAGE sql AS
$$ SELECT c, tx_chk(format('SELECT %1$s FROM tx %6$s ORDER BY %1$s %2$s %3$s LIMIT %4$s OFFSET %5$s', c, d, nl, lim, o, w)), 1::bigint
   FROM unnest(ARRAY['c', 'l']) c, unnest(ARRAY['ASC', 'DESC']) d, unnest(ARRAY['NULLS FIRST', 'NULLS LAST']) nl, unnest(ARRAY[1, 10, 100, 2000]) lim, unnest(ARRAY[0, 7]) o,
		unnest(ARRAY['', 'WHERE id > 0', 'WHERE c >= ''customer-0100'' AND c < ''customer-0300''', 'WHERE c >= ''order-''', 'WHERE c < ''a''']) w $$;
SELECT col, result, count(*) FROM tx_all() GROUP BY 1, 2 ORDER BY 1, 2;

-- rows of the buffer deleted, updated (the key moves), an insert and an update rolled back
DELETE FROM tx WHERE id > 8000 AND id % 7 = 0;
UPDATE tx SET c = 'customer-' || id::text, l = 'Zeta-' || id::text WHERE id > 8000 AND id % 11 = 0;
BEGIN;
UPDATE tx SET c = 'a' WHERE id > 8000 AND id % 3 = 0;
INSERT INTO tx SELECT 40000 + i, 'customer-', 'customer-', 1 FROM generate_series(1, 200) i;
ROLLBACK;
ANALYZE tx;
SELECT sum(rows) > 4096 AS buffer_over_the_old_limit FROM sift.col_segments('tx_i') WHERE kind = 'buffer';
SELECT col, result, count(*) FROM tx_all() GROUP BY 1, 2 ORDER BY 1, 2;

SELECT sift.col_promote('tx_i') >= 0 AS promoted;
INSERT INTO tx SELECT 50000 + i, tx_str(i * 13), tx_str(i * 17), 0 FROM generate_series(1, 5000) i;
ANALYZE tx;
SELECT col, result, count(*) FROM tx_all() GROUP BY 1, 2 ORDER BY 1, 2;
-- a buffer past the limit of a locale column: the units answer, the answers are the heap's
INSERT INTO tx SELECT 100000 + i, tx_str(i * 19), tx_str(i * 23), 0 FROM generate_series(1, 17000) i;
ANALYZE tx;
SELECT sum(rows) > 16384 AS buffer_over_the_locale_limit FROM sift.col_segments('tx_i') WHERE kind = 'buffer';
SELECT col, result, count(*) FROM tx_all() GROUP BY 1, 2 ORDER BY 1, 2;
SELECT sift.col_verify('tx_i');

DROP FUNCTION tx_all(), tx_chk(text), tx_str(int);
DROP TABLE tx;
DROP EXTENSION pg_sift_test;
DROP EXTENSION pg_sift_helpers;
DROP EXTENSION pg_sift;