-- phrase_gate: a ranked phrase query on a positions=on index checks adjacency
-- per WAND candidate (the lazy phrase gate, 1.9.1) instead of building the full
-- phrase match set first.  It must return exactly what the collect path
-- (pg_fts.lazy_phrase = off) returns -- ids, order and distances -- for 2..4
-- term phrases, NEAR, a repeated term, ties, several segments, deletes, an
-- absent term, the fts_search() SRF, and high-tf documents.
SET client_min_messages = warning;
SET enable_seqscan = off;
SET enable_bitmapscan = off;
CREATE TABLE pg (id int PRIMARY KEY, d ftsdoc);
-- vocabulary w0..w7 in a deterministic pseudo-random order: adjacent pairs,
-- reversed pairs, gaps of every size, and repeats all occur
-- plus planted 3- and 4-term phrases (and near-misses) so every shape has hits
INSERT INTO pg SELECT g, to_ftsdoc('simple',
  (SELECT string_agg('w' || ((g * 7 + i * i * 3 + (g % 11) * i) % 8), ' ')
     FROM generate_series(1, 3 + (g % 23)) i)
  || CASE WHEN g % 13 = 0 THEN ' w1 w2 w3 w4'
          WHEN g % 13 = 1 THEN ' w1 w2 w4 w3'
          WHEN g % 17 = 0 THEN ' w0 w5 w2'
          WHEN g % 17 = 1 THEN ' w0 w5 w6 w2' ELSE '' END)
FROM generate_series(1, 4000) g;
CREATE INDEX pg_fts_ix ON pg USING fts (d) WITH (positions = on);
VACUUM ANALYZE pg;

CREATE FUNCTION pg_same(qs text, k int) RETURNS text LANGUAGE plpgsql AS $$
DECLARE lz text; cl text; f1 text; f2 text; q ftsquery := to_ftsquery('simple', qs); n int;
BEGIN
  SET LOCAL pg_fts.lazy_phrase = on;
  SELECT string_agg(id || ':' || (d <=> q)::text, ','), count(*) INTO lz, n
    FROM (SELECT id, d FROM pg WHERE d @@@ q ORDER BY d <=> q LIMIT k) s;
  SELECT string_agg(ctid::text || ':' || score::text, ',') INTO f1 FROM fts_search('pg_fts_ix', q, k);
  SET LOCAL pg_fts.lazy_phrase = off;
  SELECT string_agg(id || ':' || (d <=> q)::text, ',') INTO cl
    FROM (SELECT id, d FROM pg WHERE d @@@ q ORDER BY d <=> q LIMIT k) s;
  SELECT string_agg(ctid::text || ':' || score::text, ',') INTO f2 FROM fts_search('pg_fts_ix', q, k);
  IF lz IS DISTINCT FROM cl OR f1 IS DISTINCT FROM f2 THEN
    RETURN 'DIFF ' || qs || ' k=' || k;
  END IF;
  RETURN 'ok ' || coalesce(n, 0);
END $$;

-- every phrase shape must agree; the second column proves the probes are not
-- comparing empty against empty (rule 9: non-zero hits)
SELECT qs, k, pg_same(qs, k) AS lazy_vs_collect
FROM (VALUES ('"w1 w2"'), ('"w2 w1"'), ('"w3 w3"'), ('"w0 w5 w2"'), ('"w1 w2 w3 w4"'),
             ('NEAR(w1 w6, 3)'), ('"w7 w0"'),
             -- 1.13.0: quoted phrases are exact <-> steps; exact gaps, a mixed
             -- exact/within chain and a nested (not left-deep) exact chain
             ('w1 <2> w3'), ('w1 <0> w1'), ('"w1 w2" <2> w4'), ('w1 <-> (w2 <-> w3)'),
             ('NEAR(w1 w2, 2) <-> w3')) qq(qs),
     (VALUES (1), (10), (100), (5000)) ks(k)
ORDER BY qs, k;

-- the phrase set the gate admits must equal @@@ exactly (an unbounded ranked
-- scan returns every match)
SET pg_fts.lazy_phrase = on;
SELECT (SELECT count(*) FROM (SELECT id FROM pg WHERE d @@@ to_ftsquery('simple','"w1 w2"')
          ORDER BY d <=> to_ftsquery('simple','"w1 w2"')) s)
     = (SELECT fts_count('pg_fts_ix', to_ftsquery('simple','"w1 w2"'))) AS ranked_set_is_exact,
       (SELECT fts_count('pg_fts_ix', to_ftsquery('simple','"w1 w2"'))) <
       (SELECT fts_count('pg_fts_ix', to_ftsquery('simple','w1 & w2'))) AS phrase_lt_and;
-- an absent term anywhere in the phrase: no rows, on both paths
SELECT count(*) AS absent_term_rows FROM (SELECT id FROM pg WHERE d @@@ to_ftsquery('simple','"w1 nosuch"')
  ORDER BY d <=> to_ftsquery('simple','"w1 nosuch"') LIMIT 10) s;
RESET pg_fts.lazy_phrase;

-- deletes: tombstoned docs must not be admitted
DELETE FROM pg WHERE id % 5 = 0;
VACUUM pg;
SELECT qs, pg_same(qs, 50) AS after_delete
FROM (VALUES ('"w1 w2"'), ('"w0 w5 w2"'), ('NEAR(w1 w6, 3)')) qq(qs) ORDER BY qs;

-- two segments (VACUUM flushes the pending rows into a NEW segment): one
-- cursor per (term, segment), so the gate must pick the pivot's cursors among
-- several per term.  The new segment's 'w2 w7 w7 w7 w1' docs hold only
-- reversed and gapped pairs and must never be admitted for "w1 w2".
INSERT INTO pg SELECT g, to_ftsdoc('simple',
  CASE WHEN g % 4 = 0 THEN 'w2 w7 w7 w7 w1'
       ELSE (SELECT string_agg('w' || ((g * 5 + i * 3) % 8), ' ') FROM generate_series(1, 4 + (g % 9)) i) END)
FROM generate_series(4001, 6000) g;
VACUUM pg;
SELECT fts_index_nsegments('pg_fts_ix') AS nsegments;
SELECT qs, pg_same(qs, 200) AS multiseg
FROM (VALUES ('"w1 w2"'), ('"w2 w1"'), ('"w0 w5 w2"'), ('"w2 w7"'), ('"w7 w1"')) qq(qs) ORDER BY qs;

-- moderate tf (Sum(tf) in the thousands per block, still fits a page): the
-- per-block positions decode with long per-posting runs
INSERT INTO pg SELECT g, to_ftsdoc('simple', repeat('w1 w3 ', 8 + g % 9) || 'w1 w2 ' || repeat('w2 w4 ', 6 + g % 5))
FROM generate_series(6501, 6900) g;
SELECT fts_merge('pg_fts_ix') IS NOT NULL AS merged2;
SELECT pg_same('"w1 w2"', 300) AS moderate_tf, pg_same('"w3 w1 w2"', 50) AS moderate_tf3,
       pg_same('"w2 w1"', 10) AS moderate_tf_reversed;
-- very high tf: a block whose positions exceed a page is written WITHOUT
-- positions, so the gate cannot decide it and the scan redoes the query on the
-- collect path -- same answer
INSERT INTO pg SELECT g, to_ftsdoc('simple', repeat('w1 ', 300 + g % 50) || 'w1 w2 ' || repeat('w2 ', 200))
FROM generate_series(7001, 7300) g;
SELECT fts_merge('pg_fts_ix') IS NOT NULL AS merged3;
SELECT pg_same('"w1 w2"', 400) AS high_tf;
SELECT pg_same('"w2 w1"', 10) AS high_tf_reversed;