-- The query path follows the lengths, counts and offsets of the term blob it reads without the exhaustive checks of sift.verify, so a
-- damaged blob (a bad page, or a query with verify off) must raise ERRCODE_DATA_CORRUPTED (or the dictionary mismatch of
-- ERRCODE_INDEX_CORRUPTED) and never read or write outside the blob: stored bytes set to other values, in LIST and GROUPED blobs
-- (bitmap and array containers, with and without tf nibbles and escapes). Meaningful under SANITIZE=1 (make check SANITIZE=1)
CREATE EXTENSION pg_sift_test CASCADE;
CREATE TABLE rc (id serial, body text);
INSERT INTO rc (body)
SELECT 'dense' || CASE WHEN i % 4 = 0 THEN ' quarter' ELSE '' END
	|| CASE WHEN i % 4 = 1 THEN ' q1 q1' ELSE '' END
	|| CASE WHEN i % 29 = 0 THEN repeat(' heavy', 20) ELSE '' END
	|| CASE WHEN i % 41 = 0 THEN ' lst lst' ELSE '' END
	|| CASE WHEN i % 997 = 0 THEN repeat(' lst', 22) ELSE '' END
	|| CASE WHEN i % 3001 = 0 THEN ' few' ELSE '' END
	|| CASE WHEN i % 2 = 0 THEN ' dense2' ELSE '' END
	|| CASE WHEN i % 5 = 0 THEN ' dense2' ELSE '' END
	|| CASE WHEN i % 7 = 0 THEN repeat(' dense2', 22) ELSE '' END
FROM generate_series(1, 20000) i;
-- LIST blobs below sift.grouped_min_df, GROUPED above; the second index makes the small terms GROUPED too, with ARRAY containers (format 2), the
-- third with PACKED ones (format 3)
CREATE INDEX rc_list ON rc USING sift (body);
SET sift.grouped_min_df = 10;
SET sift.test_segfmt = 2;
CREATE INDEX rc_grp ON rc USING sift (body);
SET sift.test_segfmt = 3;
CREATE INDEX rc_pk ON rc USING sift (body);
RESET sift.grouped_min_df;
RESET sift.test_segfmt;

SELECT idx, term, CASE WHEN flags & 4 = 0 THEN 'tf' ELSE 'all tf 1' END AS tfs, CASE WHEN flags & 1 = 1 THEN 'LIST' ELSE 'GROUPED' END AS kind
FROM (VALUES ('rc_list', 'quarter'), ('rc_list', 'q1'), ('rc_list', 'heavy'), ('rc_list', 'few'), ('rc_list', 'dense'), ('rc_list', 'dense2'),
			 ('rc_grp', 'quarter'), ('rc_grp', 'q1'), ('rc_grp', 'heavy'), ('rc_grp', 'lst'), ('rc_pk', 'quarter'), ('rc_pk', 'heavy')) v(idx, term),
	LATERAL sift._test_term_blob(idx::regclass, term)
ORDER BY 1, 2;

-- the verdict of one read: the postings and their tf sum, or the error
CREATE FUNCTION rc_read(blob bytea, flags int4, df int4, codec_len int4, nnib int4) RETURNS text LANGUAGE plpgsql AS $$
BEGIN
	RETURN (SELECT format('%s postings, tf sum %s', postings, tf_sum) FROM sift._test_blob_read(blob, flags, df, codec_len, nnib));
EXCEPTION WHEN OTHERS THEN
	RETURN SQLSTATE || ' ' || SQLERRM;
END $$;

-- an intact blob reads; the damage is the bytes [off, off + len) set to val
CREATE FUNCTION rc_damage(idx regclass, term text, off int4, len int4, val int4) RETURNS text LANGUAGE sql AS $$
	SELECT rc_read((SELECT string_agg(set_byte('\x00'::bytea, 0, CASE WHEN i >= off AND i < off + len THEN val ELSE get_byte(blob, i) END), ''::bytea)
					FROM generate_series(0, length(blob) - 1) i), flags, df, codec_len, nnib)
	FROM sift._test_term_blob(idx, term)
$$;

-- one damage of each kind the reads check: LIST blocks (high bits, ranges, header and block ends, the page order of the postings), GROUPED
-- groups (extents, offsets, bitmap and array containers, the page order), and the tf stream (nibbles and escape bytes of each group)
SELECT what, idx, term, rc_damage(idx::regclass, term, off, len, val) AS verdict
FROM (VALUES
	('intact LIST', 'rc_list', 'quarter', 0, 0, 0),
	('intact GROUPED with tf', 'rc_grp', 'heavy', 0, 0, 0),
	('varint without an end', 'rc_list', 'quarter', 2, 12, 255),
	('block range', 'rc_list', 'quarter', 4, 1, 128),
	('block high bits', 'rc_list', 'quarter', 5, 1, 128),
	('block header past the blob', 'rc_list', 'quarter', 419, 1, 255),
	('page order', 'rc_list', 'quarter', 558, 1, 127),
	('block past the blob', 'rc_list', 'quarter', 2492, 1, 255),
	('LIST escape bytes', 'rc_list', 'q1', 3329, 1, 127),
	('LIST keys past the limit', 'rc_list', 'q1', 697, 1, 127),
	('GROUPED offsets', 'rc_list', 'dense', 8, 1, 0),
	('GROUPED group extents', 'rc_list', 'dense', 12, 1, 0),
	('GROUPED bitmap and page mask', 'rc_list', 'dense', 12, 1, 1),
	('GROUPED payload', 'rc_list', 'dense', 20, 1, 127),
	('GROUPED short bitmap', 'rc_list', 'dense', 21, 1, 0),
	('GROUPED array size', 'rc_grp', 'heavy', 20, 1, 0),
	('GROUPED array entry', 'rc_grp', 'heavy', 32, 1, 128),
	('GROUPED array page order', 'rc_grp', 'heavy', 33, 1, 255),
	('intact GROUPED packed', 'rc_pk', 'heavy', 0, 0, 0),
	('GROUPED packed offset bits', 'rc_pk', 'heavy', 32, 1, 6),
	('GROUPED packed offset bits too wide', 'rc_pk', 'heavy', 32, 1, 200),
	('GROUPED packed page mask', 'rc_pk', 'heavy', 33, 1, 0),
	('GROUPED packed page ends', 'rc_pk', 'heavy', 668, 1, 255),
	('GROUPED packed type', 'rc_pk', 'heavy', 12, 1, 255),
	('tf nibbles', 'rc_list', 'dense2', 3120, 1, 1),
	('tf group start', 'rc_list', 'dense2', 3121, 1, 127),
	('tf escape bytes', 'rc_list', 'dense2', 3124, 1, 127)
) v(what, idx, term, off, len, val);

-- the exhaustive check of sift.verify on a PACKED group: sound, then the byte damaged (offset bits, page mask, page ends, offsets out of order on a page)
SELECT what, (SELECT sift._test_blob_check(string_agg(set_byte('\x00'::bytea, 0, CASE WHEN i >= off AND i < off + len THEN val ELSE get_byte(blob, i) END), ''::bytea))
			  FROM sift._test_term_blob('rc_pk', 'heavy'), generate_series(0, length(blob) - 1) i) AS verdict
FROM (VALUES
	('sound', 0, 0, 0),
	('offset bits', 32, 1, 6),
	('page mask', 33, 1, 0),
	('page ends', 668, 1, 255),
	('last page end', 754, 1, 0),
	('offsets out of order', 65, 1, 255)
) v(what, off, len, val);

-- the dictionary's count of the term's nibbles: fewer than the L1 bits count, none, and 2^32 - 1 (its byte count wraps in 32 bits)
SELECT n.nnib, rc_read(t.blob, t.flags, t.df, t.codec_len, n.nnib) AS verdict
FROM sift._test_term_blob('rc_list', 'q1') t, unnest(ARRAY[1, 2500, 4999, 5000, 0, -1]) n(nnib);

-- a blob that ends inside the padding after its codec, with a group count that would fit if the length arithmetic wrapped
SELECT rc_read(set_byte(set_byte(set_byte(substr(blob, 1, codec_len + 1), 8, 255), 9, 255), 10, 255), flags, df, codec_len, nnib) AS verdict
FROM sift._test_term_blob('rc_grp', 'heavy');

-- every byte of every term blob set to a few values, and every length the dictionary holds for the blob changed
CREATE FUNCTION rc_sweep(idx regclass, term text) RETURNS text LANGUAGE plpgsql AS $$
DECLARE
	t record;
	v text;
	ok int := 0;
	corrupt int := 0;
	other int := 0;
	val int;
BEGIN
	SELECT * INTO t FROM sift._test_term_blob(idx, term);
	FOR o IN 0 .. length(t.blob) - 1 LOOP
		FOREACH val IN ARRAY ARRAY[0, 1, 127, 128, 255] LOOP
			v := rc_read(set_byte(t.blob, o, val), t.flags, t.df, t.codec_len, t.nnib);
			IF v ~ '^[0-9]+ postings' THEN ok := ok + 1; ELSIF left(v, 5) IN ('XX001', 'XX002') THEN corrupt := corrupt + 1; ELSE other := other + 1; RAISE NOTICE '%', v; END IF;
		END LOOP;
	END LOOP;
	FOREACH val IN ARRAY ARRAY[0, 1, 2, 3, 255, 65535, 1000000, 2147483647, -1] LOOP
		FOR f IN 1 .. 3 LOOP
			v := rc_read(t.blob, t.flags, CASE f WHEN 1 THEN val ELSE t.df END, CASE f WHEN 2 THEN val ELSE t.codec_len END, CASE f WHEN 3 THEN val ELSE t.nnib END);
			IF v ~ '^[0-9]+ postings' THEN ok := ok + 1; ELSIF left(v, 5) IN ('XX001', 'XX002') THEN corrupt := corrupt + 1; ELSE other := other + 1; RAISE NOTICE '%', v; END IF;
		END LOOP;
	END LOOP;
	RETURN format('%s %s: %s', idx, term, CASE WHEN other = 0 AND corrupt > 0 AND ok > 0 THEN 'every read is ok or corrupt' ELSE format('ok %s, corrupt %s, other %s', ok, corrupt, other) END);
END $$;

SELECT rc_sweep('rc_list', 'quarter');
SELECT rc_sweep('rc_list', 'q1');
SELECT rc_sweep('rc_list', 'heavy');
SELECT rc_sweep('rc_list', 'few');
SELECT rc_sweep('rc_list', 'dense');
SELECT rc_sweep('rc_list', 'dense2');
SELECT rc_sweep('rc_grp', 'quarter');
SELECT rc_sweep('rc_grp', 'q1');
SELECT rc_sweep('rc_grp', 'heavy');
SELECT rc_sweep('rc_grp', 'lst');
SELECT rc_sweep('rc_pk', 'quarter');
SELECT rc_sweep('rc_pk', 'heavy');

DROP TABLE rc;
DROP EXTENSION pg_sift_test;
DROP EXTENSION pg_sift;
DROP FUNCTION rc_damage(regclass,text,integer,integer,integer), rc_read(bytea,integer,integer,integer,integer), rc_sweep(regclass,text);