-- The found flag of a connected index is filled when a segment is written (build, promotion, merge), from the parent rows visible to a snapshot, and the segment keeps
-- the stamp of that snapshot (docs/COL-CONNECT.md). What a segment holds is then as of its stamp; the parent keys that changed since are in the journal of the parent index.
CREATE EXTENSION pg_sift;
CREATE EXTENSION pg_sift_test;
SET max_parallel_workers_per_gather = 0;
CREATE TABLE dim_a (id int PRIMARY KEY, label text, price int) WITH (autovacuum_enabled = false);
INSERT INTO dim_a SELECT i, 'l' || i, i * 10 FROM generate_series(1, 100) i;
CREATE INDEX dim_a_col ON dim_a USING sift_col (id, label, price);
NOTICE:  sift_col index "dim_a_col": 1 automatic choice (track_changes)
DETAIL:  track_changes = id
HINT:  Reasons, costs and how to turn each off: SELECT * FROM sift.index_choices('dim_a_col');
CREATE TABLE events (id int PRIMARY KEY, dim_a_id int, v int NOT NULL) WITH (autovacuum_enabled = false);
INSERT INTO events SELECT i, 1 + i % 100, i FROM generate_series(1, 1000) i;
INSERT INTO events VALUES (1001, NULL, 1001), (1002, 999, 1002);
CREATE INDEX events_col ON events USING sift_col (id, dim_a_id, v) WITH (connect = 'dim_a_id -> dim_a(id)');
NOTICE:  sift_col index "events_col": 1 automatic choice (track_changes)
DETAIL:  track_changes = id
HINT:  Reasons, costs and how to turn each off: SELECT * FROM sift.index_choices('events_col');
-- rows flagged found whose parent row is not in the heap of the parent now (the flag is as of the stamp)
CREATE FUNCTION found_wrong() RETURNS bigint LANGUAGE sql AS $$
	SELECT count(*) FROM sift._test_col_read('events_col', 3) f
		JOIN events e ON e.ctid = f.tid
		LEFT JOIN dim_a d ON d.id = e.dim_a_id
		WHERE f.val IS NOT NULL AND d.id IS NULL
$$;
CREATE FUNCTION found_filled() RETURNS bigint LANGUAGE sql AS $$
	SELECT count(val) FROM sift._test_col_read('events_col', 3)
$$;
-- the journal of the parent starts after the first build: its segments say nothing
SELECT count(*) AS stamps, count(*) FILTER (WHERE lineage <> 0) AS stamped FROM sift._test_col_conn_stamps('events_col');
 stamps | stamped 
--------+---------
      1 |       0
(1 row)

SELECT found_filled() AS filled, found_wrong() AS wrong;
 filled | wrong 
--------+-------
      0 |     0
(1 row)

-- a promotion fills its segment under a stamp of the journal
INSERT INTO events SELECT i, 1 + i % 100, i FROM generate_series(2001, 2300) i;
SELECT sift.col_promote('events_col') >= 0 AS promoted;
 promoted 
----------
 t
(1 row)

CREATE TEMP TABLE st1 AS SELECT * FROM sift._test_col_conn_stamps('events_col') WHERE lineage <> 0;
SELECT count(*) AS stamped_segments, count(DISTINCT lineage) AS lineages FROM st1;
 stamped_segments | lineages 
------------------+----------
                1 |        1
(1 row)

SELECT found_filled() AS filled, found_wrong() AS wrong;
 filled | wrong 
--------+-------
    300 |     0
(1 row)

SELECT lineage = (SELECT lineage FROM sift._test_conn_gen('dim_a_col')) AS same_lineage FROM st1;
 same_lineage 
--------------
 t
(1 row)

-- the parent changes after the stamp: the segment keeps the old flags, the journal names the keys
UPDATE dim_a SET label = 'changed' WHERE id = 5;
UPDATE dim_a SET id = 1005 WHERE id = 6;
DELETE FROM dim_a WHERE id = 7;
INSERT INTO dim_a VALUES (999, 'late', 1);
SELECT found_wrong() > 0 AS stale_values;
 stale_values 
--------------
 t
(1 row)

SELECT why, array_agg(key ORDER BY key) AS keys
	FROM st1 s, LATERAL sift._test_conn_changed('dim_a_col', s.lineage, s.horizon) GROUP BY why;
 why |  keys   
-----+---------
 ok  | {5,6,7}
(1 row)

-- a segment written later has the new flags and a later stamp; the merge of segments with different stamps fills again under one
INSERT INTO events SELECT i, 1 + i % 100, i FROM generate_series(3001, 3200) i;
SELECT sift.col_promote('events_col') >= 0 AS promoted;
 promoted 
----------
 t
(1 row)

SELECT count(*) AS stamped_segments, count(DISTINCT horizon::text) AS horizons FROM sift._test_col_conn_stamps('events_col') WHERE lineage <> 0;
 stamped_segments | horizons 
------------------+----------
                2 |        2
(1 row)

SELECT sift.col_merge('events_col', true) >= 0 AS merged;
 merged 
--------
 t
(1 row)

SELECT count(*) AS segments FROM sift.col_segments('events_col') WHERE kind = 'immutable';
 segments 
----------
        1
(1 row)

CREATE TEMP TABLE st2 AS SELECT * FROM sift._test_col_conn_stamps('events_col');
SELECT count(*) AS stamps, count(*) FILTER (WHERE lineage <> 0) AS stamped FROM st2;
 stamps | stamped 
--------+---------
      1 |       1
(1 row)

SELECT (SELECT horizon::text::bigint FROM st2) > (SELECT max(horizon::text::bigint) FROM st1) AS newer_stamp;
 newer_stamp 
-------------
 t
(1 row)

SELECT found_filled() AS filled, found_wrong() AS wrong;
 filled | wrong 
--------+-------
   1471 |     0
(1 row)

SELECT sift.col_verify('events_col', true);
 col_verify 
------------
 
(1 row)

-- the rows of a parent that is gone, whose key changed or never was have no flag (the join reads them); the others have the flag of the merge
SELECT e.dim_a_id, count(*) AS n, count(f.val) AS found
	FROM events e JOIN sift._test_col_read('events_col', 3) f ON f.tid = e.ctid
	WHERE e.dim_a_id IN (4, 5, 6, 7, 999) OR e.dim_a_id IS NULL GROUP BY 1 ORDER BY 1;
 dim_a_id | n  | found 
----------+----+-------
        4 | 15 |    15
        5 | 15 |    15
        6 | 15 |     0
        7 | 15 |     0
      999 |  1 |     1
          |  1 |     0
(6 rows)

-- the answers do not change: the join is the same with and without the index
SET sift.enable_agg = on;
SELECT count(*), count(d.label), sum(d.price) FROM events e LEFT JOIN dim_a d ON d.id = e.dim_a_id;
 count | count |  sum   
-------+-------+--------
  1502 |  1471 | 755551
(1 row)

SET sift.enable_agg = off;
SELECT count(*), count(d.label), sum(d.price) FROM events e LEFT JOIN dim_a d ON d.id = e.dim_a_id;
 count | count |  sum   
-------+-------+--------
  1502 |  1471 | 755551
(1 row)

RESET sift.enable_agg;
-- a parent table whose keys do not fit maintenance_work_mem is looked up by its btree: the segments are stamped and the flags filled
CREATE TABLE dim_big (id int PRIMARY KEY, label text) WITH (autovacuum_enabled = false);
INSERT INTO dim_big SELECT i * 1000, repeat('x', 40) || i FROM generate_series(1, 60000) i;
CREATE INDEX dim_big_col ON dim_big USING sift_col (id, label) WITH (order_array_columns = 'off');
NOTICE:  sift_col index "dim_big_col": 1 automatic choice (track_changes)
DETAIL:  track_changes = id
HINT:  Reasons, costs and how to turn each off: SELECT * FROM sift.index_choices('dim_big_col');
CREATE TABLE big_events (id int PRIMARY KEY, dim_big_id int NOT NULL) WITH (autovacuum_enabled = false);
INSERT INTO big_events SELECT i, 1000 * (1 + i % 60000) FROM generate_series(1, 100) i;
CREATE INDEX big_events_col ON big_events USING sift_col (id, dim_big_id) WITH (connect = 'dim_big_id -> dim_big(id)', key_cut = 'off', order_array_columns = 'off');
NOTICE:  sift_col index "big_events_col": 1 automatic choice (track_changes)
DETAIL:  track_changes = id
HINT:  Reasons, costs and how to turn each off: SELECT * FROM sift.index_choices('big_events_col');
SET maintenance_work_mem = '1MB';
INSERT INTO big_events SELECT i, 1000 * (1 + i % 60000) FROM generate_series(101, 200) i;
SELECT sift.col_promote('big_events_col') >= 0 AS promoted;
 promoted 
----------
 t
(1 row)

SELECT count(*) AS stamps, count(*) FILTER (WHERE lineage <> 0) AS stamped FROM sift._test_col_conn_stamps('big_events_col');
 stamps | stamped 
--------+---------
      2 |       1
(1 row)

SELECT count(val) AS filled FROM sift._test_col_read('big_events_col', 2);
 filled 
--------
    100
(1 row)

RESET maintenance_work_mem;
INSERT INTO big_events SELECT i, 1000 * (1 + i % 60000) FROM generate_series(201, 300) i;
SELECT sift.col_promote('big_events_col') >= 0 AS promoted;
 promoted 
----------
 t
(1 row)

SELECT count(*) AS stamps, count(*) FILTER (WHERE lineage <> 0) AS stamped FROM sift._test_col_conn_stamps('big_events_col');
 stamps | stamped 
--------+---------
      3 |       2
(1 row)

SELECT count(val) AS filled FROM sift._test_col_read('big_events_col', 2);
 filled 
--------
    200
(1 row)

SELECT sift.col_verify('big_events_col', true);
 col_verify 
------------
 
(1 row)

-- a lock the fill takes (the table, the parent) can reload the relcache entry of the index and free the state it caches: the test setting overwrites it and invalidates the entry there
SET sift.test_col_rel_inval = on;
INSERT INTO events SELECT i, 1 + i % 100, i FROM generate_series(5001, 5100) i;
SELECT sift.col_promote('events_col') >= 0 AS promoted;
 promoted 
----------
 t
(1 row)

SELECT count(*) FILTER (WHERE lineage <> 0) > 0 AS stamped FROM sift._test_col_conn_stamps('events_col');
 stamped 
---------
 t
(1 row)

SELECT count(val) > 0 AS filled FROM sift._test_col_read('events_col', 3);
 filled 
--------
 t
(1 row)

RESET sift.test_col_rel_inval;
DROP TABLE big_events, dim_big, events, dim_a CASCADE;
DROP FUNCTION found_wrong(), found_filled();
DROP EXTENSION pg_sift_test;
DROP EXTENSION pg_sift;