extract(epoch from clock_timestamp() - start_time);

END

$$ LANGUAGE plpgsql;

-- 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', NULL, NULL, 'skip-empty-xacts', '1');

-- non-spilled xact

INSERT INTO stats_test values(1);

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

count

-------

(1 row)

-- Check stats, wait for the stats collector to update. 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 wait_for_decode_stats(false, true);

SELECT wait_for_decode_stats(false, false);

wait_for_decode_stats

-----------------------

(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;

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;

slot_name | spill_txns | spill_count | total_txns | total_bytes

-----------------+------------+-------------+------------+-------------

regression_slot | t | t | t | t

(1 row)

-- reset the slot stats, and wait for stats collector to reset

-- reset the slot stats, and wait for stats collector's total txn to reset

SELECT pg_stat_reset_replication_slot('regression_slot');

pg_stat_reset_replication_slot

--------------------------------

(1 row)

SELECT wait_for_decode_stats(true,true);

SELECT wait_for_decode_stats(true,false);

wait_for_decode_stats

-----------------------

regression_slot | 0 | 0 | 0 | 0

(1 row)

-- decode and check stats again.

-- 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', NULL, NULL, 'skip-empty-xacts', '1');

count

-------

5002

(1 row)

-- Check stats, wait for the stats collector to update. 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 wait_for_decode_stats(false, true);

wait_for_decode_stats

-----------------------

regression_slot | t | t | t | t

(1 row)

-- reset the slot stats, and wait for stats collector to reset

SELECT pg_stat_reset_replication_slot('regression_slot');

pg_stat_reset_replication_slot

--------------------------------

(1 row)

-- non-spilled xact

INSERT INTO stats_test values(generate_series(1, 10));

SELECT wait_for_decode_stats(false, false);

SELECT wait_for_decode_stats(true, true);

wait_for_decode_stats

-----------------------

(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;

SELECT slot_name, spill_txns, spill_count, total_txns, total_bytes FROM pg_stat_replication_slots;

slot_name | spill_txns | spill_count | total_txns | total_bytes

-----------------+------------+-------------+------------+-------------

regression_slot | f | f | t | t

regression_slot | 0 | 0 | 0 | 0

(1 row)

-- decode and check stats again.

SELECT count(*) FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'skip-empty-xacts', '1');

count

-------

5002

(1 row)

SELECT wait_for_decode_stats(false, true);

wait_for_decode_stats

-----------------------

(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;

slot_name | spill_txns | spill_count | total_txns | total_bytes

-----------------+------------+-------------+------------+-------------

regression_slot | t | t | t | t

(1 row)

-- Ensure stats can be repeatedly accessed using the same stats snapshot. See

END

$$ LANGUAGE plpgsql;

INSERT INTO stats_testvalues(1);

SELECTcount(*)FROM pg_logical_slot_get_changes('regression_slot',NULL,NULL,'skip-empty-xacts','1');

SELECT wait_for_decode_stats(false, false);

SELECT slot_name, spill_txns=0AS spill_txns, spill_count=0AS spill_count, total_txns>0AS total_txns, total_bytes>0AS total_bytesFROM pg_stat_replication_slots;

SELECT pg_stat_reset_replication_slot('regression_slot');

SELECT wait_for_decode_stats(true, false);

SELECT slot_name, spill_txns, spill_count, total_txns, total_bytesFROM pg_stat_replication_slots;

BEGIN;

INSERT INTO stats_testSELECT'serialize-topbig--1:'||g.iFROM generate_series(1,5000) g(i);

SELECT wait_for_decode_stats(false, true);

SELECT slot_name, spill_txns>0AS spill_txns, spill_count>0AS spill_count, total_txns>0AS total_txns, total_bytes>0AS total_bytesFROM pg_stat_replication_slots;

SELECT pg_stat_reset_replication_slot('regression_slot');

INSERT INTO stats_testvalues(generate_series(1,10));

SELECT wait_for_decode_stats(false, false);

SELECT slot_name, spill_txns=0AS spill_txns, spill_count=0AS spill_count, total_txns>0AS total_txns, total_bytes>0AS total_bytesFROM pg_stat_replication_slots;

BEGIN;

</para>

<para>

Number of decoded transactions sent to the decoding output plugin for

this slot. Thiscounter is used to maintain the top level transactions,

so the counter is not incrementedfor subtransactions. Note that this

includes the transactions that arestreamed and/or spilled.

this slot. Thiscounts toplevel transactions only, and is not incremented

for subtransactions. Note that this includes the transactions that are

streamed and/or spilled.

</para></entry>

</row>

<structfield>total_bytes</structfield><type>bigint</type>

</para>

<para>

Amount of decodedtransactions data sent to the decoding output plugin

Amount of decodedtransaction data sent to the decoding output plugin

while decoding the changes from WAL for this slot. This can be used to

gauge the total amount of data sent during logical decoding. Note that

this includesthedata that is streamed and/or spilled.

this includes data that is streamed and/or spilled.

</para>

</entry>

</row>