-- The direct probe of a correlated lookup child (combo/probe.c, combo/probe_bt.c) over a key of several columns on an index of three levels: a constant array expression, a group
-- column, an integer, two texts and a wide constant text, like the lookup tables of a report (a mapping keyed by scope, entity type, tag and sub-tag). The descent compares the key with
-- the separators and entries of a page from the first column that the bounds of its range do not both agree with the key on, and keeps the pages above the leaves pinned for the batch;
-- the answers are the child's plan's and the plan of the database's, under every collation, order and kind of text the probe compares (the C collation, another collation, an order by the
-- bytes, a descending column, a text that is stored compressed, entries with a NULL column, keys that are prefixes of each other), and what the probe saves is checked by counts:
-- sift._test_probe_stats gives the keys probed, the compares of a key with an index tuple, their columns, the compares through the opclass's order function and the pages read.
CREATE EXTENSION pg_sift;
CREATE EXTENSION pg_sift_test;
SET client_min_messages = warning;
SET max_parallel_workers_per_gather = 0;
SET seq_page_cost = 100; SET random_page_cost = 4;
SET enable_bitmapscan = off;
SET sift.agg_force = on;
\ir test/agg_probe_helpers.sql
-- the collation of the cases that are not the C collation: the cluster's, if it has one
SELECT coalesce((SELECT collname FROM pg_collation WHERE collname IN ('en_US.utf8', 'und-x-icu') ORDER BY collname LIMIT 1), 'C') AS coll \gset

-- the keys of the items of the entries (n, tag, sub), with keys that are prefixes of each other (tag1, tag10, tag100), absent keys and a second version of some keys
CREATE TABLE lk_src AS
SELECT DISTINCT o.n::int AS n, o.tag, o.sub
FROM entries e, jsonb_to_recordset(e.doc['items']) AS o(tag text, sub text, n float, score float) WHERE o.n IS NOT NULL;
CREATE TABLE lk_all (scope int[], grp text, n int, tag text, sub text, pad text, v text);
INSERT INTO lk_all SELECT NULL, 'g' || (n % 5), n, tag, sub, repeat('p', 120), 'v' || n || tag || sub FROM lk_src WHERE hashint4(n) % 4 <> 0 OR tag LIKE 'tag1%';
-- some rows are kept for one scope only, and a second version of the key (the first visible entry in index order is the match)
INSERT INTO lk_all SELECT ARRAY[n], 'g' || (n % 5), n, tag, sub, repeat('p', 120), 'scoped' FROM lk_src WHERE hashint4(n) % 7 = 0;
INSERT INTO lk_all SELECT NULL, 'g' || (n % 5), n, tag, sub, repeat('p', 120), 'second' FROM lk_src WHERE hashint4(n) % 5 = 0;
-- entries of the keys' neighbours: a column that is NULL, the empty string, a text that extends the key's, a key of another group
INSERT INTO lk_all SELECT NULL, 'g' || (n % 5), n, NULL, sub, repeat('p', 120), 'null tag' FROM lk_src WHERE hashint4(n) % 9 = 0;
INSERT INTO lk_all SELECT NULL, 'g' || (n % 5), n, '', sub, repeat('p', 120), 'empty tag' FROM lk_src WHERE hashint4(n) % 9 = 1;
INSERT INTO lk_all SELECT NULL, 'g' || (n % 5), n, tag || 'x', sub, repeat('p', 120), 'longer tag' FROM lk_src WHERE hashint4(n) % 3 = 0;
INSERT INTO lk_all SELECT NULL, 'g' || ((n + 1) % 5), n, tag, sub, repeat('p', 120), 'other group' FROM lk_src WHERE hashint4(n) % 3 = 1;

CREATE TABLE lk_c AS TABLE lk_all;
CREATE INDEX lk_c_k ON lk_c (coalesce(scope, '{}'::int[]), grp, n, tag, sub, pad);
CREATE TABLE lk_l AS TABLE lk_all;
SELECT set_config('probe.coll', :'coll', false) IS NOT NULL AS collation_set;
DO $$ BEGIN
  EXECUTE format('ALTER TABLE lk_l ALTER grp TYPE text COLLATE %1$I, ALTER tag TYPE text COLLATE %1$I, ALTER sub TYPE text COLLATE %1$I', current_setting('probe.coll'));
END $$;
CREATE INDEX lk_l_k ON lk_l (coalesce(scope, '{}'::int[]), grp, n, tag, sub, pad);
CREATE TABLE lk_d AS TABLE lk_all;
CREATE INDEX lk_d_k ON lk_d (coalesce(scope, '{}'::int[]), grp, n DESC, tag DESC NULLS FIRST, sub, pad);
CREATE TABLE lk_p AS TABLE lk_all;
CREATE INDEX lk_p_k ON lk_p (coalesce(scope, '{}'::int[]), grp text_pattern_ops, n, tag text_pattern_ops, sub text_pattern_ops, pad);
-- a text that is stored compressed in the index (long and repetitive), as the key's third column
CREATE TABLE lk_z AS SELECT scope, grp, n, tag || repeat('z', 3000) AS tag, sub, pad, v FROM lk_all WHERE tag IS NOT NULL;
CREATE INDEX lk_z_k ON lk_z (coalesce(scope, '{}'::int[]), grp, n, tag, sub, pad);
VACUUM (FREEZE, ANALYZE) lk_c, lk_l, lk_d, lk_p, lk_z, entries;
SELECT pg_relation_size('lk_c_k') / current_setting('block_size')::int > 150 AS three_levels, pg_relation_size('lk_z_k') > 0 AS z_built;

-- the condition on the key: the columns of the index, the value of each from the item o
CREATE FUNCTION kc(tag text DEFAULT 'o.tag', n text DEFAULT 'o.n::int') RETURNS text LANGUAGE sql AS $$
  SELECT format($q$coalesce(k.scope, '{}') = '{}' AND k.grp = 'g' || (%s %% 5) AND k.n = %s AND k.tag = %s AND k.sub = o.sub AND k.pad = repeat('p', 120)$q$, n, n, tag) $$;
-- the plan of each case reads its index (the other columns of the table are in no index)
SELECT t.name, p.* FROM (VALUES ('lk_c', 'C collation'), ('lk_l', 'collation of the cluster'), ('lk_d', 'descending'), ('lk_p', 'text_pattern_ops')) t(tab, name),
	 LATERAL probe_check(case_query(t.tab || ' k', kc(), 'LIMIT 1')) p ORDER BY t.tab;
-- the same key under the other column orders of one table, every match of a key in index order, and the first match under a filter
SELECT * FROM step('every match', 'lk_c k', kc(), 'OFFSET 0');
SELECT * FROM step('every match, collation', 'lk_l k', kc(), 'OFFSET 0');
SELECT * FROM step('filter on the row', 'lk_c k', kc() || ' AND k.v <> ''second''');
SELECT * FROM step('filter on the row, collation', 'lk_l k', kc() || ' AND k.v <> ''second''');
SELECT * FROM step('compressed key', 'lk_z k', kc('o.tag || repeat(''z'', 3000)'));
-- a key that stops after the first columns: the rows of the group, the first by the index order
SELECT * FROM step('shorter key', 'lk_c k', $c$coalesce(k.scope, '{}') = '{}' AND k.grp = 'g' || (o.n::int % 5) AND k.n = o.n::int$c$);
SELECT * FROM step('shorter key, descending', 'lk_d k', $c$coalesce(k.scope, '{}') = '{}' AND k.grp = 'g' || (o.n::int % 5) AND k.n = o.n::int$c$);
-- keys that are prefixes of the entries' and of each other
SELECT * FROM step('prefix keys', 'lk_c k', kc('substr(o.tag, 1, 4)'));
SELECT * FROM step('empty key text', 'lk_c k', kc(''''''));
-- the leaf pages change under the keys: rows deleted, updated and added, then the same keys again
DELETE FROM lk_c WHERE hashint4(n) % 6 = 0;
UPDATE lk_c SET v = v || 'u' WHERE hashint4(n) % 5 = 1;
INSERT INTO lk_c SELECT NULL, 'g' || (n % 5), n, tag, sub, repeat('p', 120), 'late' FROM lk_src WHERE hashint4(n) % 8 = 0;
SELECT * FROM step('after changes', 'lk_c k', kc());
VACUUM lk_c;
SELECT * FROM step('after vacuum', 'lk_c k', kc());

-- what the probe saves, by counts. The compares of a key with the tuples of its path, per key probed, and the columns each compares: the columns the range of the page agrees with
-- the key on are not compared again (the constant array, the group and the wide text are the same in every tuple); the pages above the leaves are read once for the batch
CREATE FUNCTION probe_counts(tab text, cond text) RETURNS TABLE (tuple_compares_per_key numeric, columns_per_compare numeric, generic_per_key numeric, pages_per_key numeric) LANGUAGE plpgsql AS $$
DECLARE s record;
BEGIN
  SET LOCAL sift.enable_agg = on;
  PERFORM sift._test_probe_stats(true);
  EXECUTE 'SELECT count(*) FROM (' || case_query(tab || ' k', cond) || ') q';
  SELECT * INTO s FROM sift._test_probe_stats(true);
  RETURN QUERY SELECT round(s.tuple_compares::numeric / s.probes, 1), round(s.column_compares::numeric / s.tuple_compares, 1), round(s.generic_compares::numeric / s.probes, 1), round(s.page_reads::numeric / s.probes, 1);
END $$;
SELECT tuple_compares_per_key < 16 AS compares_bounded, columns_per_compare < 3.2 AS prefix_skipped, generic_per_key = 0 AS no_generic, pages_per_key < 1.4 AS pins_kept FROM probe_counts('lk_c', kc());
-- under another collation the columns that differ from the key's are compared through the order function, the equal ones are not (the key's column is equal in most of the tuples it meets)
SELECT :'coll' = 'C' OR (columns_per_compare < 3.2 AND generic_per_key < 10) AS equal_bytes_not_compared FROM probe_counts('lk_l', kc());

DROP FUNCTION probe_counts(text, text), kc(text, text);
DROP FUNCTION step_io(text, text, text, text, text), case_query_col(text, text, text, text), step(text, text, text, text), case_query(text, text, text), probe_check(text), check_query(text);
DROP TABLE lk_src, lk_all, lk_c, lk_l, lk_d, lk_p, lk_z, entries;
DROP EXTENSION pg_sift_test;
DROP EXTENSION pg_sift;