-- predictability
SET synchronous_commit = on;
SELECT 'init' FROM
    pg_create_logical_replication_slot('regression_slot_stats1', 'test_decoding') s1,
    pg_create_logical_replication_slot('regression_slot_stats2', 'test_decoding') s2,
    pg_create_logical_replication_slot('regression_slot_stats3', 'test_decoding') s3;
 ?column? 
----------
 init
(1 row)

CREATE TABLE stats_test(data text);
-- non-spilled xact
SET logical_decoding_work_mem to '64MB';
INSERT INTO stats_test values(1);
SELECT count(*) FROM pg_logical_slot_get_changes('regression_slot_stats1', NULL, NULL, 'skip-empty-xacts', '1');
 count 
-------
     3
(1 row)

SELECT count(*) FROM pg_logical_slot_get_changes('regression_slot_stats2', NULL, NULL, 'skip-empty-xacts', '1');
 count 
-------
     3
(1 row)

SELECT count(*) FROM pg_logical_slot_get_changes('regression_slot_stats3', NULL, NULL, 'skip-empty-xacts', '1');
 count 
-------
     3
(1 row)

SELECT pg_stat_force_next_flush();
 pg_stat_force_next_flush 
--------------------------
 
(1 row)

SELECT slot_name, spill_txns = 0 AS spill_txns, spill_count = 0 AS spill_count, total_txns > 0 AS total_txns, total_bytes > 0 AS total_bytes FROM pg_stat_replication_slots ORDER BY slot_name;
       slot_name        | spill_txns | spill_count | total_txns | total_bytes 
------------------------+------------+-------------+------------+-------------
 regression_slot_stats1 | t          | t           | t          | t
 regression_slot_stats2 | t          | t           | t          | t
 regression_slot_stats3 | t          | t           | t          | t
(3 rows)

RESET logical_decoding_work_mem;
-- reset stats for one slot, others should be unaffected
SELECT pg_stat_reset_replication_slot('regression_slot_stats1');
 pg_stat_reset_replication_slot 
--------------------------------
 
(1 row)

SELECT slot_name, spill_txns = 0 AS spill_txns, spill_count = 0 AS spill_count, total_txns > 0 AS total_txns, total_bytes > 0 AS total_bytes FROM pg_stat_replication_slots ORDER BY slot_name;
       slot_name        | spill_txns | spill_count | total_txns | total_bytes 
------------------------+------------+-------------+------------+-------------
 regression_slot_stats1 | t          | t           | f          | f
 regression_slot_stats2 | t          | t           | t          | t
 regression_slot_stats3 | t          | t           | t          | t
(3 rows)

-- reset stats for all slots
SELECT pg_stat_reset_replication_slot(NULL);
 pg_stat_reset_replication_slot 
--------------------------------
 
(1 row)

SELECT slot_name, spill_txns = 0 AS spill_txns, spill_count = 0 AS spill_count, total_txns > 0 AS total_txns, total_bytes > 0 AS total_bytes FROM pg_stat_replication_slots ORDER BY slot_name;
       slot_name        | spill_txns | spill_count | total_txns | total_bytes 
------------------------+------------+-------------+------------+-------------
 regression_slot_stats1 | t          | t           | f          | f
 regression_slot_stats2 | t          | t           | f          | f
 regression_slot_stats3 | t          | t           | f          | f
(3 rows)

-- verify accessing/resetting stats for non-existent slot does something reasonable
SELECT * FROM pg_stat_get_replication_slot('do-not-exist');
  slot_name   | spill_txns | spill_count | spill_bytes | stream_txns | stream_count | stream_bytes | total_txns | total_bytes | stats_reset 
--------------+------------+-------------+-------------+-------------+--------------+--------------+------------+-------------+-------------
 do-not-exist |          0 |           0 |           0 |           0 |            0 |            0 |          0 |           0 | 
(1 row)

SELECT pg_stat_reset_replication_slot('do-not-exist');
ERROR:  replication slot "do-not-exist" does not exist
SELECT * FROM pg_stat_get_replication_slot('do-not-exist');
  slot_name   | spill_txns | spill_count | spill_bytes | stream_txns | stream_count | stream_bytes | total_txns | total_bytes | stats_reset 
--------------+------------+-------------+-------------+-------------+--------------+--------------+------------+-------------+-------------
 do-not-exist |          0 |           0 |           0 |           0 |            0 |            0 |          0 |           0 | 
(1 row)

-- spilling the xact
BEGIN;
INSERT INTO stats_test SELECT 'serialize-topbig--1:'||g.i FROM generate_series(1, 5000) g(i);
COMMIT;
SELECT count(*) FROM pg_logical_slot_peek_changes('regression_slot_stats1', NULL, NULL, 'skip-empty-xacts', '1');
 count 
-------
  5002
(1 row)

-- Check stats. We can't test the exact stats count as that can vary if any
-- background transaction (say by autovacuum) happens in parallel to the main
-- transaction.
SELECT pg_stat_force_next_flush();
 pg_stat_force_next_flush 
--------------------------
 
(1 row)

SELECT slot_name, spill_txns > 0 AS spill_txns, spill_count > 0 AS spill_count FROM pg_stat_replication_slots;
       slot_name        | spill_txns | spill_count 
------------------------+------------+-------------
 regression_slot_stats1 | t          | t
 regression_slot_stats2 | f          | f
 regression_slot_stats3 | f          | f
(3 rows)

-- Ensure stats can be repeatedly accessed using the same stats snapshot. See
-- https://postgr.es/m/20210317230447.c7uc4g3vbs4wi32i%40alap3.anarazel.de
BEGIN;
SELECT slot_name FROM pg_stat_replication_slots;
       slot_name        
------------------------
 regression_slot_stats1
 regression_slot_stats2
 regression_slot_stats3
(3 rows)

SELECT slot_name FROM pg_stat_replication_slots;
       slot_name        
------------------------
 regression_slot_stats1
 regression_slot_stats2
 regression_slot_stats3
(3 rows)

COMMIT;
SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot_stats4_twophase', 'test_decoding', false, true) s4;
 ?column? 
----------
 init
(1 row)

-- The INSERT changes are large enough to be spilled but will not be, because
-- the transaction is aborted. The logical decoding skips collecting further
-- changes too. The transaction is prepared to make sure the decoding processes
-- the aborted transaction.
BEGIN;
INSERT INTO stats_test SELECT 'serialize-toobig--1:'||g.i FROM generate_series(1, 5000) g(i);
PREPARE TRANSACTION 'test1_abort';
ROLLBACK PREPARED 'test1_abort';
SELECT count(*) FROM pg_logical_slot_get_changes('regression_slot_stats4_twophase', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
 count 
-------
     1
(1 row)

-- Verify that the decoding doesn't spill already-aborted transaction's changes.
SELECT pg_stat_force_next_flush();
 pg_stat_force_next_flush 
--------------------------
 
(1 row)

SELECT slot_name, spill_txns, spill_count FROM pg_stat_replication_slots WHERE slot_name = 'regression_slot_stats4_twophase';
            slot_name            | spill_txns | spill_count 
---------------------------------+------------+-------------
 regression_slot_stats4_twophase |          0 |           0
(1 row)

DROP TABLE stats_test;
SELECT pg_drop_replication_slot('regression_slot_stats1'),
    pg_drop_replication_slot('regression_slot_stats2'),
    pg_drop_replication_slot('regression_slot_stats3'),
    pg_drop_replication_slot('regression_slot_stats4_twophase');
 pg_drop_replication_slot | pg_drop_replication_slot | pg_drop_replication_slot | pg_drop_replication_slot 
--------------------------+--------------------------+--------------------------+--------------------------
                          |                          |                          | 
(1 row)