-- 0.10.0: the index keeps itself. ivf_maintain rebuilds only past its threshold; a rebuild empties only
-- the part of the delta it could see; a watch without an index is not given one; after a reader switch
-- the index is rebuilt for the new reader once it covers min_coverage.
\set VERBOSITY terse

CREATE EXTENSION pg_phasevec;
SELECT extversion FROM pg_extension WHERE extname = 'pg_phasevec';
SELECT phasevec.reader_register('r256', repeat('a', 64), 256) > 0 AS r256;
CREATE TABLE docs (id int PRIMARY KEY, body text);
INSERT INTO docs SELECT g, 'doc ' || g FROM generate_series(1, 2000) g;
SELECT phasevec.watch('docs', 'docs'::regclass, 'id', 'r.body', 'r256') > 0 AS watch;
CREATE FUNCTION code_of(g int) RETURNS smallint[] LANGUAGE sql IMMUTABLE AS $$
    SELECT array(SELECT (((g * 31 + j * 17 * (g % 7 + 1) + (g / 13) * j) % 255) - 127)::smallint
                   FROM generate_series(1, 256) j) $$;
CREATE FUNCTION signs_of(c smallint[]) RETURNS bit(256) LANGUAGE sql IMMUTABLE AS $$
    SELECT string_agg(CASE WHEN x > 0 THEN '1' ELSE '0' END, '' ORDER BY o)::bit(256)
      FROM unnest(c) WITH ORDINALITY u(x, o) $$;
CREATE FUNCTION read_as(p_reader text, p_from int, p_to int, p_shift int) RETURNS bigint LANGUAGE sql AS $$
    SELECT count(phasevec.record('docs', s.key, p_reader, s.text_hash, 0::bigint, code_of(s.key::int + p_shift), NULL,
                                 signs_of(code_of(s.key::int + p_shift))))
      FROM phasevec.sources s WHERE s.key::int BETWEEN p_from AND p_to $$;
-- a row whose TEXT changes and is read again, as the reader would: a new hash, a new code
CREATE FUNCTION change_as(p_from int, p_to int, p_shift int) RETURNS bigint LANGUAGE plpgsql AS $$
BEGIN
    UPDATE docs SET body = 'doc ' || id || ' v' || p_shift WHERE id BETWEEN p_from AND p_to;
    RETURN (SELECT count(phasevec.record('docs', s.key, 'r256', s.text_hash, 0::bigint, code_of(s.key::int + p_shift), NULL,
                                         signs_of(code_of(s.key::int + p_shift))))
              FROM phasevec.sources s WHERE s.key::int BETWEEN p_from AND p_to);
END $$;
SELECT read_as('r256', 1, 2000, 0) AS read;

-- no index: nothing to maintain, and none is built
SELECT phasevec.ivf_maintain('docs') AS no_index;
SELECT count(*) AS cents FROM phasevec.ivf_cents;
SELECT phasevec.ivf_build('docs') AS lists;

-- 50 rows read again: below p_absorb (60), and absorbed + delta below p_frac * 2000 (100)
SELECT read_as('r256', 1, 50, 1) AS reread;
SELECT count(*) AS delta FROM phasevec.ivf_dirty;
SELECT phasevec.ivf_maintain('docs', 0.05, 60) AS below;
-- one of them read again keeps one delta row, with its newer transaction
SELECT count(*) AS one_row FROM phasevec.ivf_dirty WHERE key = '7';
-- a delta row of a transaction no snapshot can see yet (here, a future xid): never emptied
INSERT INTO phasevec.ivf_dirty (watch, key, txid)
    SELECT w.id, '2000', (pg_current_xact_id()::text::bigint + 1000000)::text::xid8 FROM phasevec.watches w WHERE w.name = 'docs';
-- 60 more: 111 >= 100, rebuilt
SELECT read_as('r256', 51, 110, 1) AS reread;
SELECT phasevec.ivf_maintain('docs', 0.05, 60) AS past_fraction;
SELECT array_agg(key) AS delta_kept FROM phasevec.ivf_dirty;
SELECT n, absorbed FROM phasevec.ivf_cents;

-- 40 rows whose text changed, absorbed into the lists (p_absorb 30, p_frac 0.5 is far): the delta empties but for the
-- future row, the index holds 2040 entries, and the rows read last are found from the lists
SELECT change_as(111, 150, 1) AS changed;
SELECT phasevec.ivf_maintain('docs', 0.5, 30) AS absorbed;
SELECT array_agg(key) AS delta_kept FROM phasevec.ivf_dirty;
SELECT n, absorbed FROM phasevec.ivf_cents;
SELECT key, round(score::numeric, 4) AS score FROM phasevec.find_ivf('docs', 0, code_of(112), 1, false, 200, signs_of(code_of(112)), 1000)
 WHERE key = '111';
SELECT (SELECT array_agg(key ORDER BY score DESC, key) FROM phasevec.find('docs', 0, code_of(130), 10, false, 2100, signs_of(code_of(130))))
     = (SELECT array_agg(key ORDER BY score DESC, key) FROM phasevec.find_ivf('docs', 0, code_of(130), 10, false, 2100, signs_of(code_of(130)), 1000))
       AS same_as_find;
-- a row absorbed, then its text changes: not served (the validation drops its list entry)
BEGIN;
UPDATE docs SET body = 'changed' WHERE id = 120;
SELECT count(*) AS served_stale FROM phasevec.find_ivf('docs', 0, code_of(121), 10, false, 2000, signs_of(code_of(121)), 1000) WHERE key = '120';
ROLLBACK;
-- what was absorbed lives in chunks, never in the rewritten lists; a backend that loads the index itself
-- (no shared copy) joins lists and chunks and still answers exactly what find answers
SELECT count(*) > 0 AS chunks, sum(n) AS chunk_readings FROM phasevec.ivf_added;
SET phasevec.shared_ivf_cache = off;
SELECT (SELECT array_agg(key ORDER BY score DESC, key) FROM phasevec.find('docs', 0, code_of(130), 10, false, 2100, signs_of(code_of(130))))
     = (SELECT array_agg(key ORDER BY score DESC, key) FROM phasevec.find_ivf('docs', 0, code_of(130), 10, false, 2100, signs_of(code_of(130)), 1000))
       AS same_as_find_loaded;
SELECT key, round(score::numeric, 4) AS score FROM phasevec.find_ivf('docs', 0, code_of(112), 1, false, 200, signs_of(code_of(112)), 1000)
 WHERE key = '111';
RESET phasevec.shared_ivf_cache;
-- absorbed (40) + delta (1) reach 2 % of the build: rebuilt, the count of absorbed back to 0
SELECT phasevec.ivf_maintain('docs', 0.02, 30) AS rebuilt;
SELECT n, absorbed FROM phasevec.ivf_cents;
SELECT count(*) AS chunks_after_build FROM phasevec.ivf_added;
SELECT (SELECT array_agg(key ORDER BY score DESC, key) FROM phasevec.find('docs', 0, code_of(500), 10, false, 2000, signs_of(code_of(500))))
     = (SELECT array_agg(key ORDER BY score DESC, key) FROM phasevec.find_ivf('docs', 0, code_of(500), 10, false, 2000, signs_of(code_of(500)), 1000))
       AS same_as_find;

-- a reader switch: the index is the old reader's; it is rebuilt only once the new one covers 95 %
SELECT phasevec.reader_register('r256b', repeat('b', 64), 256) > 0 AS r256b;
SELECT phasevec.reader_switch('docs', 'r256b');
SELECT read_as('r256b', 1, 1000, 0) AS half;
SELECT phasevec.ivf_maintain('docs') AS half_covered;
SELECT read_as('r256b', 1001, 2000, 0) AS rest;
SELECT phasevec.ivf_maintain('docs') AS covered;
SELECT rd.name AS index_reader FROM phasevec.ivf_cents c JOIN phasevec.readers rd ON rd.id = c.reader;
SELECT watch, action, delta, indexed FROM phasevec.ivf_maintain_all();

DROP TABLE docs;
DROP FUNCTION code_of(int), signs_of(smallint[]), change_as(int, int, int);
DROP FUNCTION read_as(text, int, int, int);
DROP EXTENSION pg_phasevec;