Problem: CDC replication/apply is slow (#704)

* Problem: CDC replication is slow

Our existing implementation is basic and it will execute a statement and waits for its result to come before issuing next statement. We already improved DML statements targetting same table with in a transaction by coalscing them into multi-value insert statement, but it will be beneficial only for logical decoding transaction which has multiple DML statements with in a txn.

However, we have encountered few cases where the transaction had only one DML statement. This kind of txn will be slower because statements are executed sequentially from the client's perspective.

Here is an example of how a single statement transaction will be executed now,

1. Execute ["BEGIN"]()
2. Execute [bunch of "SET" statements]() which are related to replication session setup
3. Prepare [DML statement]()
4. Execute [prepared statement with values]()
5. Execute procedure to update [replication origin progress]()
6. Execute [COMMIT]()
7. Map [target current insert lsn to commit lsn]() for feedback reporting (i.e. sentinel replay_lsn)

**Solution**: Use [pipeline
API](https://www.postgresql.org/docs/current/libpq-pipeline-mode.html

) from libpq client library

The proposed implementation would enter into pipeline mode as soon as a new libpq connection is created for the ld_apply/ld_replay process. All the statements would be executed on a pipeline by default except few for statements which needs response immediately(e.g. step 7). A dedicated connection would be used to serve (step 7), because it needs synchronous response.

The following functions are being called on the target PGSQL handle from ld_apply & ld_replay.

* pgsql_begin
* pgsql_set_gucs
* pgsql_execute
* pgsql_replication_origin_xact_setup
* pgsql_prepare
* pgsql_execute_prepared
* pgsql_current_wal_insert_lsn
* pgsql_current_wal_flush_lsn

Among all of the above function, only pgsql_current_wal_insert_lsn and pgsql_current_wal_flush_lsn returns values and other functions are write only.

The idea is to have 2 PGSQL connection handles, 1 for all write activity which can go through pipeline and another one could be used for reading.

Pipeline connection has to be synced/drained at some point to avoid accumulating results on the server & client which would end up eating lots of heap memory. The current implementation syncs based on the time interval(i.e. for every 1s). There are other methods like statement/txn count based sync, which may or may not be efficient.

The following command can be used to generate loads to understand the performance improvement made by this commit,

```
CREATE TABLE metrics (
    time TIMESTAMP NOT NULL,
    name TEXT,
    id NUMERIC,
    value FLOAT
);
```

```
-- insert_metric.sql
\set id random(1, 1000000)
\set value random(0,100)
INSERT INTO metrics (time, name, value) VALUES (NOW(), 'metric_' || :id, :value);
```
```
pgbench -n -c 40 -j 1 -t 10000 -f insert_metric.sql $SOURCE
```

- `-c` number of database connection to utlize (i.e. server side concurrency)
- `-j` number of threads to create on the client machine (i.e. client side concurrency)

synchronous_commit=off)
```
pgbench -n -c 1 -j 1 -t 10000 -f insert_metric.sql $TARGET
```
```
tps = 1177 (without initial connection time)
```

```
tps = 175
```

```
tps = 1652
```

**This commit improves the single statement txn throughput by 10x**

Ideally, we should aim to get performance number close to direct ingestion(i.e. 1200 txn/s). We are 40% performing better than the baseline in this iteration. However, we can aim more as in the real system there will be more than 1 connection will be utilized. We can't really race against multiple connection doing steady ingestion around 1000 txn/s per connection, but lets optimize the single connection throughput to the max!

This change will be a foundation to the future improvements which steers towards that.

1. Optimize step 2 - Instead of executing bunch of SET statements for every txn, run once in the beginning for the session
2. Optimize step 9 - Probably we can simply use pg_replication_origin_progress as replay_lsn?

Signed-off-by: Arunprasad Rajkumar <ar.arunprasad@gmail.com>