-- 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';
 extversion 
------------
 0.16.0
(1 row)

SELECT phasevec.reader_register('r256', repeat('a', 64), 256) > 0 AS r256;
 r256 
------
 t
(1 row)

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;
 watch 
-------
 t
(1 row)

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;
 read 
------
 2000
(1 row)

-- no index: nothing to maintain, and none is built
SELECT phasevec.ivf_maintain('docs') AS no_index;
 no_index 
----------
 none
(1 row)

SELECT count(*) AS cents FROM phasevec.ivf_cents;
 cents 
-------
     0
(1 row)

SELECT phasevec.ivf_build('docs') AS lists;
 lists 
-------
    45
(1 row)

-- 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;
 reread 
--------
     50
(1 row)

SELECT count(*) AS delta FROM phasevec.ivf_dirty;
 delta 
-------
    50
(1 row)

SELECT phasevec.ivf_maintain('docs', 0.05, 60) AS below;
 below 
-------
 none
(1 row)

-- 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';
 one_row 
---------
       1
(1 row)

-- 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;
 reread 
--------
     60
(1 row)

SELECT phasevec.ivf_maintain('docs', 0.05, 60) AS past_fraction;
 past_fraction 
---------------
 rebuilt
(1 row)

SELECT array_agg(key) AS delta_kept FROM phasevec.ivf_dirty;
 delta_kept 
------------
 {2000}
(1 row)

SELECT n, absorbed FROM phasevec.ivf_cents;
  n   | absorbed 
------+----------
 2000 |        0
(1 row)

-- 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;
 changed 
---------
      40
(1 row)

SELECT phasevec.ivf_maintain('docs', 0.5, 30) AS absorbed;
 absorbed 
----------
 absorbed
(1 row)

SELECT array_agg(key) AS delta_kept FROM phasevec.ivf_dirty;
 delta_kept 
------------
 {2000}
(1 row)

SELECT n, absorbed FROM phasevec.ivf_cents;
  n   | absorbed 
------+----------
 2040 |       40
(1 row)

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';
 key | score  
-----+--------
 111 | 1.0000
(1 row)

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;
 same_as_find 
--------------
 t
(1 row)

-- 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';
 served_stale 
--------------
            0
(1 row)

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;
 chunks | chunk_readings 
--------+----------------
 t      |             40
(1 row)

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;
 same_as_find_loaded 
---------------------
 t
(1 row)

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';
 key | score  
-----+--------
 111 | 1.0000
(1 row)

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;
 rebuilt 
---------
 rebuilt
(1 row)

SELECT n, absorbed FROM phasevec.ivf_cents;
  n   | absorbed 
------+----------
 2000 |        0
(1 row)

SELECT count(*) AS chunks_after_build FROM phasevec.ivf_added;
 chunks_after_build 
--------------------
                  0
(1 row)

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;
 same_as_find 
--------------
 t
(1 row)

-- 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;
 r256b 
-------
 t
(1 row)

SELECT phasevec.reader_switch('docs', 'r256b');
 reader_switch 
---------------
 
(1 row)

SELECT read_as('r256b', 1, 1000, 0) AS half;
 half 
------
 1000
(1 row)

SELECT phasevec.ivf_maintain('docs') AS half_covered;
 half_covered 
--------------
 none
(1 row)

SELECT read_as('r256b', 1001, 2000, 0) AS rest;
 rest 
------
 1000
(1 row)

SELECT phasevec.ivf_maintain('docs') AS covered;
 covered 
---------
 rebuilt
(1 row)

SELECT rd.name AS index_reader FROM phasevec.ivf_cents c JOIN phasevec.readers rd ON rd.id = c.reader;
 index_reader 
--------------
 r256b
(1 row)

SELECT watch, action, delta, indexed FROM phasevec.ivf_maintain_all();
 watch | action | delta | indexed 
-------+--------+-------+---------
 docs  | none   |     0 |    2000
(1 row)

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;