Copy function for logical replication slots

Hi,

I'd like to propose a copy function for logical replication slots.
Currently when we create a new logical replication slot it starts to
read WAL from an LSN of the current insert. This function copies a
existing logical replication slot while changing output plugin and
persistence. That is, the copied new replication slot starts from the
same LSN as the source one. Since a new copied slot starts from the
same LSN of existing one we don't need to care about WAL reservation.

A use case I imagined is for investigations for example. I mean that
when replication collision occurs on subscriber there is no way to see
what replicated data is conflicting (perhaps error log helps it but is
not detailed) and there is no way to advance a replication origin in
order to exactly skip to apply conflicting data. By creating a new
logical slot with a different output plugin at the same LSN, we can
see what data a replication slot will decode (and send) and those LSNs
as well. This function will help for that purpose.

Here is execution samples.

postgres(1:17715)=# select
pg_create_logical_replication_slot('orig_slot', 'test_decoding');
pg_create_logical_replication_slot
------------------------------------
(orig_slot,0/164A410)
(1 row)

Time: 17.759 ms
postgres(1:17715)=# select
pg_copy_logical_replication_slot('orig_slot', 'copy1_slot');
pg_copy_logical_replication_slot
----------------------------------
(copy1_slot,0/164A410)
(1 row)

Time: 6.074 ms
postgres(1:17715)=# select
pg_copy_logical_replication_slot('orig_slot', 'copy2_slot',
'wal2json');
pg_copy_logical_replication_slot
----------------------------------
(copy2_slot,0/164A410)
(1 row)

Time: 6.201 ms
postgres(1:17715)=# select
pg_copy_logical_replication_slot('orig_slot', 'copy3_slot',
'wal2json', true);
pg_copy_logical_replication_slot
----------------------------------
(copy3_slot,0/164A410)
(1 row)

Time: 5.071 ms
postgres(1:17715)=# select * from pg_replication_slots ;
slot_name | plugin | slot_type | datoid | database |
temporary | active | active_pid | xmin | catalog_xmin | restart_lsn |
confirmed_flush_lsn
------------+---------------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------
copy3_slot | wal2json | logical | 13237 | postgres | t
| t | 17715 | | 568 | 0/164A3D8 | 0/164A410
copy2_slot | wal2json | logical | 13237 | postgres | f
| f | | | 568 | 0/164A3D8 | 0/164A410
copy1_slot | orig_slot | logical | 13237 | postgres | f
| f | | | 568 | 0/164A3D8 | 0/164A410
orig_slot | test_decoding | logical | 13237 | postgres | f
| f | | | 568 | 0/164A3D8 | 0/164A410
(4 rows)

Feedback is very welcome.

Regards,

--
Masahiko Sawada
NIPPON TELEGRAPH AND TELEPHONE CORPORATION
NTT Open Source Software Center