/*
 * The WAL records the index writes (DESIGN.md §25).
 *
 * The extension logs through one of two resource managers, chosen per index
 * at CREATE INDEX and recorded on its meta page: PostgreSQL's own generic
 * one, or pg_lion's, which can only be registered when the library is in
 * shared_preload_libraries.  This test drives every write path and reads the
 * WAL back with pg_walinspect to name the records they produced.
 *
 * It therefore has TWO expected files, because the whole point is that the
 * output differs: walrecords.out is the generic-mode run (`make
 * installcheck`) and walrecords_1.out is the rmgr-mode one (`make
 * installcheck-rmgr`).  pg_regress accepts either.
 *
 * pg_waldump cannot name these records - it does not load the module, so it
 * prints the resource manager as "custom170" and the record type as UNKNOWN -
 * which is why this uses pg_walinspect, which runs in the backend, where
 * rm_identify() and rm_desc() are ours.
 */
\set ECHO none
\set VERBOSITY terse
SET client_min_messages = warning;
LOAD 'pg_lion';
\set ECHO all
CREATE EXTENSION IF NOT EXISTS pg_lion;
CREATE EXTENSION IF NOT EXISTS pg_walinspect;

/* VACUUM can only set all-visible for commits that reached disk. */
SET synchronous_commit = on;

CREATE TABLE lion_wr (id int, k int);
CREATE INDEX lion_wr_k ON lion_wr USING lion (k);

-- Which logger this index got, and whether this server has the resource
-- manager at all.  Everything below reads one way or the other from here.
SELECT lion_index_wal_mode('lion_wr_k') AS wal_mode;
SELECT count(*) = 1 AS rmgr_registered
  FROM pg_get_wal_resource_managers() WHERE rm_name = 'pg_lion';

SELECT pg_current_wal_lsn() AS lsn0 \gset

-- A new key: the entry goes onto a directory leaf.
INSERT INTO lion_wr VALUES (1, 1);

-- Many rows under two keys: the payload spills onto container pages, the
-- containers grow in place (the hot-key insert record), and the pages they
-- live on split.
INSERT INTO lion_wr SELECT i, i % 2 FROM generate_series(2, 20000) i;

-- One more row under a key that is already there: nothing but the in-place
-- member insert and the entry counter that travels with it.
INSERT INTO lion_wr VALUES (20001, 1);

-- Many distinct keys: the directory has to split, which is the record with
-- the most buffers in it (left, right, old right sibling, meta page).
INSERT INTO lion_wr SELECT i, i FROM generate_series(100000, 130000) i;

-- Removal: VACUUM rewrites containers under a cleanup lock, deletes the
-- entries it empties and hands the pages they owned back to the free space
-- map, takes the directory leaves it empties out of the tree (UNLINK,
-- DESIGN.md §21) - and then counts the keys that are left onto the meta
-- page, for the planner (the META record, DESIGN.md §33).
DELETE FROM lion_wr WHERE k >= 100000;
VACUUM (FREEZE) lion_wr;

-- ... and a key whose posting set owns whole container pages, so that
-- deleting it frees them: the entry goes, then its pages are marked deleted
-- and recorded in the map.
DELETE FROM lion_wr WHERE k < 2;
VACUUM (FREEZE) lion_wr;

SELECT pg_switch_wal() IS NOT NULL AS wal_switched;
SELECT pg_current_wal_lsn() AS lsn1 \gset

/*
 * The record types those statements produced.  Only the NAMES are pinned:
 * how many of each there are depends on page fill and on where the last
 * checkpoint fell.
 */
SELECT record_type
  FROM pg_get_wal_records_info(:'lsn0', :'lsn1')
 WHERE resource_manager IN ('pg_lion', 'Generic')
 GROUP BY record_type
 ORDER BY record_type;

/*
 * rm_desc(): pg_walinspect prints whatever the resource manager says about a
 * record.  Only the SHAPE is pinned - every record has a description, and an
 * rmgr-mode one names the blocks it touched - because the offsets in it
 * depend on page fill.
 */
SELECT bool_and(description IS NOT NULL AND description <> '') AS all_described
  FROM pg_get_wal_records_info(:'lsn0', :'lsn1')
 WHERE resource_manager = 'pg_lion';

SELECT lion_index_verify('lion_wr_k', true);

/*
 * ITEM_REPLACE AS A DELTA (DESIGN.md §25).
 *
 * A container that has used up its growth slack is rewritten through the
 * general path.  That record carries the byte ranges that differ from what
 * the page already holds, not the whole rewritten item: 1.6 KB for a dense
 * ARRAY, which would make the container path cost more WAL than a generic
 * byte diff of the same change.
 *
 * What is pinned is the shape: the delta operation appears, and the records
 * that use it stay small.  How many there are depends on page fill, and a
 * rewrite that really does change most of the item (an ARRAY that turns into
 * a NARROW) still goes in whole, which is why the bound is on the records
 * that carry a delta rather than on every ITEM_REPLACE.  A full-page image
 * is not part of what the record says, so it is subtracted: the first touch
 * of a page after a checkpoint carries one whatever the operation is.
 * Sixteen keys keep every container an ARRAY, of some 900 members at most,
 * which takes members in place until its slack is gone; four keys would make
 * NARROWs (DESIGN.md §38), which take every member in place, and no
 * container would be rewritten.
 */
CREATE TABLE lion_wd (id int, k int);
CREATE INDEX lion_wd_k ON lion_wd USING lion (k) WITH (inline_limit = 64);
INSERT INTO lion_wd SELECT i, i % 16 FROM generate_series(1, 20000) i;
CHECKPOINT;
SELECT pg_current_wal_lsn() AS lsn2 \gset
INSERT INTO lion_wd SELECT i, i % 16 FROM generate_series(20001, 21600) i;
SELECT pg_switch_wal() IS NOT NULL AS wal_switched_again;
SELECT pg_current_wal_lsn() AS lsn3 \gset

SELECT count(*) > 0 AS wrote_deltas,
	   coalesce(max(record_length - fpi_length), 0) <= 512 AS deltas_are_small
  FROM pg_get_wal_records_info(:'lsn2', :'lsn3')
 WHERE resource_manager = 'pg_lion' AND record_type = 'ITEM_REPLACE'
   AND description LIKE '%delta off%';

/*
 * INLINE ENTRY SLACK (DESIGN.md §4).  An insert into a key whose payload
 * still has room writes inside the entry: no other entry on the leaf moves,
 * and the record is a delta of the counters and the tail of the payload.
 */
CREATE TABLE lion_wi (id int, k int);
CREATE INDEX lion_wi_k ON lion_wi USING lion (k);
INSERT INTO lion_wi SELECT i, i % 400 FROM generate_series(1, 20000) i;
CHECKPOINT;
SELECT pg_current_wal_lsn() AS lsn4 \gset
INSERT INTO lion_wi SELECT i, i % 400 FROM generate_series(20001, 20400) i;
SELECT pg_switch_wal() IS NOT NULL AS wal_switched_thrice;
SELECT pg_current_wal_lsn() AS lsn5 \gset

SELECT count(*) > 0 AS wrote_entry_deltas,
	   coalesce(max(record_length - fpi_length), 0) <= 512 AS entry_deltas_are_small
  FROM pg_get_wal_records_info(:'lsn4', :'lsn5')
 WHERE resource_manager = 'pg_lion' AND record_type = 'ENTRY'
   AND description LIKE '%delta off%';

SELECT lion_index_verify('lion_wd_k', true);
SELECT lion_index_verify('lion_wi_k', true);

/*
 * DOWNLINK (DESIGN.md §22).  A posting set has no internal level while it has
 * at most 16 leaves, and the sets above have fewer; past that it is a tree,
 * and every split of a leaf posts its new leaf's downlink on the level above.
 * The build makes the trees: it fills a leaf to posting_fillfactor, so at 10
 * every container is a leaf of its own, and each of two keys over 1,231 heap
 * pages has twenty.  The rows inserted after it fill the last leaves and split
 * them, and the WAL read is kept to those inserts.
 */
CREATE TABLE lion_wl (id int, k int, pad char(80));
INSERT INTO lion_wl SELECT i, i % 2, 'x' FROM generate_series(1, 80000) i;
CREATE INDEX lion_wl_k ON lion_wl USING lion (k) WITH (posting_fillfactor = 10);
SELECT max_posting_height, posting_internal_pages
  FROM lion_index_stats('lion_wl_k');
SELECT pg_current_wal_lsn() AS lsn6 \gset
INSERT INTO lion_wl SELECT i, i % 2, 'x' FROM generate_series(80001, 160000) i;
SELECT pg_switch_wal() IS NOT NULL AS wal_switched_four_times;
SELECT pg_current_wal_lsn() AS lsn7 \gset

SELECT count(*) > 0 AS posted_downlinks
  FROM pg_get_wal_records_info(:'lsn6', :'lsn7')
 WHERE resource_manager = 'pg_lion' AND record_type = 'DOWNLINK';
SELECT lion_index_verify('lion_wl_k', true);

/*
 * ITEM_REPLACE AS A SPLICE (DESIGN.md §25).
 *
 * A bulk-built container has no growth slack, so the first member an insert
 * adds to it takes the general path and rewrites the item.  When the member
 * lands in the MIDDLE of an ARRAY, every member above it moves up by two
 * bytes, which a DELTA can only say by repeating them; the SPLICE operation
 * says it with the two bytes, where they go and the new count.  Updates on a
 * heap with free space put the new row versions on the pages the old ones
 * are on, so their TIDs land inside the containers rather than after them.
 * Thirty-two keys keep every container an ARRAY (fewer and denser ones would
 * be NARROWs, which grow in place), and there is no CHECKPOINT between the
 * build and the updates: the build's own pages need no full-page image, and
 * a block that carries one does not show its operations.
 */
CREATE TABLE lion_ws (id int, k int) WITH (fillfactor = 70);
INSERT INTO lion_ws SELECT i, i % 32 FROM generate_series(1, 20000) i;
CREATE INDEX lion_ws_k ON lion_ws USING lion (k) WITH (inline_limit = 64);
SELECT containers = array_containers AS all_arrays
  FROM lion_index_stats('lion_ws_k');
SELECT pg_current_wal_lsn() AS lsn8 \gset
UPDATE lion_ws SET k = (k + 1) % 32 WHERE id % 97 = 0;
SELECT pg_switch_wal() IS NOT NULL AS wal_switched_five_times;
SELECT pg_current_wal_lsn() AS lsn9 \gset

SELECT count(*) > 0 AS wrote_splices,
	   coalesce(max(record_length - fpi_length), 0) <= 160 AS splices_are_small
  FROM pg_get_wal_records_info(:'lsn8', :'lsn9')
 WHERE resource_manager = 'pg_lion' AND record_type = 'ITEM_REPLACE'
   AND description LIKE '%splice off%';
SELECT lion_index_verify('lion_ws_k', true);

/*
 * UNLINK (DESIGN.md §21, §25).  A directory leaf taken out of the tree is
 * rewritten whole, HALF_DEAD and then DELETED, by records that build the page
 * from nothing.  The page is registered as one the record initialises: redo
 * zeroes it and replays the record, and an rmgr-mode record takes no image of
 * it.  So after a checkpoint each page's first record takes the only image of
 * it, and no record takes one of a page an earlier record of the same VACUUM
 * already wrote.  (A generic record logs a page it initialises as an image,
 * of a nearly empty page here.)  Under wal_consistency_checking every block of
 * every record carries an image as well, which redo compares its page with
 * and does not apply: only an image redo applies (APPLY) is one.
 */
CREATE TABLE lion_wu (k text) WITH (autovacuum_enabled = off);
INSERT INTO lion_wu
	SELECT 'key' || lpad(g::text, 6, '0') || repeat('x', 40) FROM generate_series(1, 20000) g;
CREATE INDEX lion_wu_k ON lion_wu USING lion (k);
DELETE FROM lion_wu WHERE k < 'key015000';
CHECKPOINT;
SELECT pg_current_wal_lsn() AS lsn10 \gset
VACUUM lion_wu;
SELECT pg_switch_wal() IS NOT NULL AS wal_switched_six_times;
SELECT pg_current_wal_lsn() AS lsn11 \gset

WITH b AS (
	SELECT start_lsn, relblocknumber AS blk,
		   coalesce('APPLY' = ANY (block_fpi_info), false) AS image
	  FROM pg_get_wal_block_info(:'lsn10', :'lsn11', false)
	 WHERE relfilenode = pg_relation_filenode('lion_wu_k'))
SELECT (SELECT deleted_pages > 0 FROM lion_index_stats('lion_wu_k')) AS unlinked,
	   lion_index_wal_mode('lion_wu_k') <> 'rmgr' OR
	   NOT EXISTS (SELECT 1 FROM b
				   WHERE b.image AND
						  EXISTS (SELECT 1 FROM b e
								   WHERE e.blk = b.blk AND e.start_lsn < b.start_lsn))
	   AS one_image_a_page;
SELECT lion_index_verify('lion_wu_k', true);

/*
 * SEVERAL SLICES TO A RECORD (DESIGN.md §51, "Insert cost, revisited").  A
 * row of a sliced column goes into the slice of every bit its value's
 * encoding has set: -1 has 31 of an int4's, and the existence slice makes
 * 32.  Where the slices are posting trees (inline_limit = 64 spills them) the
 * member goes into the last container of each in place, and the adds into
 * slices whose entries share a directory leaf go out several posting leaves
 * and the leaf to a record: three and the leaf in generic mode, the four
 * blocks every record keeps to; seven and the leaf with the resource
 * manager.  So a row's 32 slices write some eleven records, or five, not 32.
 * Where they are INLINE entries that grow within their slack, the row's
 * slices on one leaf are one record of that leaf alone.
 */
CREATE TABLE lion_wsl (id int, v int4);
INSERT INTO lion_wsl SELECT i, -1 FROM generate_series(1, 20000) i;
CREATE INDEX lion_wsl_v ON lion_wsl USING lion (v) WITH (slices = on, inline_limit = 64);
CREATE INDEX lion_wsl_w ON lion_wsl USING lion (v) WITH (slices = on);
SELECT pg_current_wal_lsn() AS lsn12 \gset
INSERT INTO lion_wsl SELECT i, -1 FROM generate_series(20001, 20200) i;
SELECT pg_switch_wal() IS NOT NULL AS wal_switched_seven_times;
SELECT pg_current_wal_lsn() AS lsn13 \gset

WITH r AS (
	SELECT relfilenode = pg_relation_filenode('lion_wsl_v') AS chain,
		   start_lsn, count(*) AS nblocks
	  FROM pg_get_wal_block_info(:'lsn12', :'lsn13', false)
	 WHERE relfilenode IN (pg_relation_filenode('lion_wsl_v'),
						   pg_relation_filenode('lion_wsl_w'))
	 GROUP BY 1, 2)
SELECT max(nblocks) FILTER (WHERE chain) AS chain_most_blocks,
	   count(*) FILTER (WHERE chain AND nblocks = max_blocks) >= 200 * 3
	   AS chain_full_records,
	   count(*) FILTER (WHERE chain) <= 200 * 14 AS chain_records_per_row_at_most_14,
	   max(nblocks) FILTER (WHERE NOT chain) AS inline_most_blocks,
	   count(*) FILTER (WHERE NOT chain) <= 200 * 4 AS inline_records_per_row_at_most_4
  FROM r, (SELECT CASE lion_index_wal_mode('lion_wsl_v') WHEN 'rmgr' THEN 8 ELSE 4 END
		   AS max_blocks) m;
SELECT lion_index_verify('lion_wsl_v', true);
SELECT lion_index_verify('lion_wsl_w', true);

DROP TABLE lion_wr, lion_wd, lion_wi, lion_wl, lion_ws, lion_wu, lion_wsl;
DROP EXTENSION pg_walinspect;