The main use-case for the sentinel table is to communicate progress made by
the three independent worker processes for stream, transform, and replay.
It is also needed to communicate new startpos, endpos and apply values with
those processes from another pgcopydb (interactive) command, a process
that's not in the process tree.

This was previously solved by creating and using a pgcopydb.sentinel table
on the source database. Now that we have a SQLite database internal to the
pgcopydb run, we can use that instead.

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

* Remove summary files, use our SQLite database instead.

* Get rid of summary files for the indexing and constraints too.

This allows also to entirely remove the need for semaphores to handle
concurrency between worker processes, using SQLite insert-or-ignore for
concurrency control instead.

Also the idea of TablePaths and IndexPaths elements are gone, with the lock
files and the done files mechanisms.

* Protect SQLite concurrent write access with a semaphore.

SQLite has not been designed for write-write concurrency, so implement a
critical section in our code so that SQLite only sees a single writer at all
times. It may still see multiple readers though.

Because we have re-entrant queries (an SQLite iterator function may then
choose to run another SQLite query), this commit also adds support for
re-entrant semaphore in a way that doesn't call into semop(2) again when we
already are in the critical section.

* Assorted fixes.

When the COPY table-data workers fail before the iteration over the list of
tables is done, we have a race condition where the COPY supervisor would be
busy and is not using waitpid(), thus not reacting to worker process
termination.

To solve, limit the responsibilities of the COPY supervisor process even
more and create a new process that's responsible for iterating over the
table list and sending tables/parts to the queue, etc.

The code to call TRUNCATE ONLY was only active for tables setup with COPY
partitioning (--split-tables-larger-than), and was forced to "false" for the
other tables.

The code also evolved in a way that it would build a sub-SELECT query with
the table attribute list as the "srcQname" argument to the pgsql.c internal
bits for the copy, making it impossible to then re-use that bit in the sql
command for the TRUNCATE operation.

To fix, implement a new way to prepare the COPY query bits and pieces so
that we can re-use the qualified table name in the TRUNCATE command and also
build a full COPY (SELECT a, b, c FROM ONLY t WHERE ...) TO STDOUT; query.

In passing, because of the way we are changing the summary files to use our
internal SQLite database instead, also push the TRUNCATE command of COPY
partitionned table to the COPY supervisor process. This is a better way to
solve the concurrency issues and make sure that TRUNCATE is done only once,
and also finished before any of the copy-data worker processes get started.

* add quotes identifiers

* test changes

* Fix escaping of identifiers while transforming for wal2json plugin.

wal2json doesn't escape identifiers as test_decoding do. To fix this issue
we are using PQescapeIdentifier function. Even though this function doesn't
require PGconn to be connected, but creating an empty PGconn object was
unfeasible. Some fields of PGconn that are required by new versions of
libpq for PQescapeIdentifier are not available in older PGconn. So decided
to do this the right way, but creating PGconn with an actual server
connection. It turned out more simpler than managing a dummy PGconn object.

* Fix review comments.

* Fix style.

---------

Co-authored-by: VaibhaveS <vaibhavedavey@gmail.com>

test_decoding transform relies on source catalog to decode the update message.

Prior to this commit, update while using test_decoding would work only while doing `pgcopydb clone --follow`. This commit enables support for update message decoding
while using test_decoding plugin in `pgcopydb follow` too.

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

We close the FILE which is used to log into file before calling
the `copydb_cleanup_sysv_resources` which still uses log functions leads
to crash due to invalid FILE* access.

This commit fixes the crash by closing the FILE after calling the
cleanup function.

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

We track resources likes semaphore and queue into a global array and
deallocate the non freed objects using exit handler. The array tracking
these resources uses pointers, which could inadvertently track resource
handles allocated from the stack.

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

The filter clause is not supported by my local SQLite for some reason,
so I needed to use a different approach to count the number of
constraints on a table.

This fixes the following error:

    ERROR  Failed to prepare SQLite statement: select count(1) as indexes, count(1) filter(where c.oid is not null) OVER () as constraints from s_index i left join s_constraint c on c.indexoid = i.oid where tableoid = $1
    ERROR  [SQLite] misuse of aggregate: count()
    ERROR  Failed to count indexes attached to table dist_tables.t1
    ERROR  Failed to copy data for table with oid 17226 and part number 0, see above for details

Relevant documentation at https://www.sqlite.org/lang_aggfunc.html says:

The count(X) function returns a count of the number of times that X is not NULL in a group

There were few issues with --skip-extensions flag

1) Excluded extensions were not part of the filter
2) Invalid memory access from pg_dump_db_extension_namespace_hook
3) Filter table populated without checking whether it is already done or
   not.

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

Currently, pgcopydb dump & restore doesn't support skipping extensions or filtering because we don't populate the context with extension details & skip list information.

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

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

Catalog queries may return a very large number of rows when the source
database contain a very large number of objects (tables, indexes, etc). To
best cope with that use Single Row Mode to fetch one row at a time in-memory
before inserting it to our internal SQLite catalogs.

This allows to implement an in-memory array with the capabilities to
spill-to-disk and also hash-table like lookup operations, and with good
multi-process concurrency handling.

Using SQLite internally will allow reducing the memory usage of pgcopydb and
make it possible to address databases with more than a million of tables
defined.

That allows C-c to work in more situations, which is always good.

We would see “WARNING: transaction already in progress” in the logs.

Postgres application_name is limited to NAMEDATALEN, sending more bytes
than that gives us a NOTICE and we can easily avoid it. In passing, avoid
redundancies in the application_name when the ps_buffer already starts with
"pgcopydb: ".

Search page shows 404, whereas genindex and modindex are empty pages.

* Add option to output numeric as string on wal2json

This change adds a new option `--wal2json-numeric-as-string` that
changes wal2json plugin output format to print numeric data types as
strings. This is accomplished by passing the
`--numeric-data-types-as-string` option to wal2json plugin.

This is useful to prevent precision loss when using wal2json
plugin to stream changes from a database that uses numeric data types.

wal2json plugin version that supports `--numeric-data-types-as-string`
option is required to use this pgcopydb option. As of today there is no
official wal2json release that supports this option, but it is available
on master branch of the project.

Relevant changes in wal2json plugin is at
https://github.com/eulerto/wal2json/pull/255

* Add env to output numeric as string on wal2json

PGCOPYDB_WAL2JSON_NUMERIC_AS_STRING can be set to a boolean value that
will be used to determine if pgcopydb should set the wal2json option
`--numeric-data-types-as-string`.

In passing, also add the PGCOPYDB_OUTPUT_PLUGIN env variable to all
relevant pages of our documentation.

* Run make indent target

This commit will allow developers to run `make indent` to indent their
code in their development branches by solving all the indentation issues
in the codebase.

* Add citus_indent to CI

Endpos can be set to 0/0 which is an invalid LSN.

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

This commit adds a new option --restore-jobs to pgcopydb that allows
specifying how many jobs can be used to run pg_restore operations in
parallel. This option can also be set using the PGCOPYDB_RESTORE_JOBS
environment variable.

When this option is set to 1, pgcopydb will run pg_restore with the
--single-transaction option, and pgcopydb will behave the same as it
used to. Otherwise, pg_restore will be run with the --jobs option set
to the number of jobs specified by the user.

If the user does not supply the --restore-jobs option, or set the
environment variable, pgcopydb will use the index-jobs value as default.

In some cases, the replay process doesn't reach the endpos due to a problem
in the stream process. The problem arises when stream process receives a
'k' message from wal sender and reaches the endpos. But instead of writing
the newly received LSN, we write previous LSN with the KEEPALIVE message.
The following filtered logs from the streaming process illustrate this
issue.

```
endpos is now set to 3/AF01C408
Reported write_lsn 3/AF01C408, flush_lsn 3/AF01BF38, replay_lsn 3/AF01BF38
streamFlush: 3/AF01C408 3/AF01C0A0
Inserted action KEEPALIVE for lsn 3/AF01C0A0 @2023-11-27 11:43:55.946327+0000
Flushed up to 3/AF01C0A0 in file "cdc/0000000100000003000000AF.json.partial"
pgsql_stream_logical: endpos reached on keepalive: 3/AF01C5A0
streamFlush: 3/AF01C5A0 3/AF01C0A0
Inserted action KEEPALIVE for lsn 3/AF01C0A0 @2023-11-27 11:43:59.754050+0000
```

This commit addresses and resolves the issue of duplicate key errors when resuming partially executed transactions (continuedTxn) in pgcopydb. We have reintroduced the transaction metadata file, which is essential for identifying the commitLSN of a partial transaction.

Unlike our previous approach, which led to a deadlock between the transform and apply phases, this update brings a more efficient process. Now, the apply phase creates metadata for any partial (continued) transactions during the commit. This metadata is then used to accurately skip the already applied partial transaction if a resume is needed.

This fix is crucial, particularly for tables with unique constraints, where executing the same continued transaction twice previously resulted in duplicate key errors. With this update, pgcopydb ensures smooth and error-free handling of transaction resumes.

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

In the follow process, we create two Unix pipes for bidirectional communication between child processes (prefetch, transform, apply) during replay mode. Each child process closes its unused pipe descriptors upon startup. For example, the prefetch process, which only writes to the transform process, closes the read end of its pipe descriptors immediately.

However, the follow process (parent) itself, which initializes these pipes, does not utilize them for communication. Leaving these pipe descriptors open in the follow process leads to an issue where EOF is not sent, even after the child processes close their ends upon completion. This situation can result in a deadlock, especially when an interruption occurs through Ctrl+C or a SIGTERM signal.

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

In .rst files, the first line of a code block must be empty.  This
was causing a malformed code block that is hard to read.

* Update options for pgcopydb snapshot command

* Update options for pgcopydb compare

* Update options for pgcopydb dump subcommands

* Rename create_snapshot_command to snapshot_command

* Capitalize first letters of print and version command descriptions.

This change makes it easier to identify the connections of pgcopydb in
pg_stat_activity. The process title contains information on the current
operation, and the pid of the process is also included in the
application name. This way we can easily identify the connections of
pgcopydb in pg_stat_activity even if the process title is truncated.

Currently for each table we are creating and closing a new connection to
the target. Using a same connection would save us some time and resource,
both at pgcopydb and target side.

Also, seting GUCs only once for a given target connection session would
again save us resources. This would also reduce 8 extra log lines that are
currently produced for each table which otherwise would produce noise if we
have say, 10K tables.

The issue was that we weren't setting the snapshot for the query that
fetches the list of large object ids. This inconsistency led to errors when
trying to access large objects that weren't restored on target.

Error:
```
15:30:11 108 ERROR  [SOURCE 85] ERROR:  large object 16390 does not exist
15:30:11 108 ERROR  [SOURCE 85] Context: Failed to open large object 16390
15:30:11 107 ERROR  [SOURCE 83] ERROR:  large object 16392 does not exist
15:30:11 107 ERROR  [SOURCE 83] Context: Failed to open large object 16392
```

Here is a snippet of man page for msgsnd & msgrcv,

```
      int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

       ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp,
                      int msgflg);

DESCRIPTION
       The  msgsnd() and msgrcv() system calls are used to send messages to, and receive messages from, a System V message queue.  The calling process must have write
       permission on the message queue in order to send a message, and read permission to receive a message.

       The msgp argument is a pointer to a caller-defined structure of the following general form:

           struct msgbuf {
               long mtype;       /* message type, must be > 0 */
               char mtext[1];    /* message data */
           };

       The mtext field is an array (or other structure) whose size is specified by msgsz, a nonnegative integer value.  Messages of zero length (i.e., no mtext field)
       are  permitted.   The  mtype field must have a strictly positive integer value.  This value can be used by the receiving process for message selection (see the
       description of msgrcv() below).
```

According to it, the 3rd arg of both msgsnd and msgrcv must pass the
size of mtext i.e. excluding the mtype.

If we apply this logic into our context, we must pass the sizeof(QMessage->data), not sizeof(QMessage).

We can also find the corruption using valgrind like below,

```
valgrind --trace-children=yes --tool=memcheck --leak-check=no pgcopydb
copy table-data
```

It gives the following summary,

```
==640986== Syscall param msgsnd(msgp->mtext) points to uninitialised byte(s)
==640986==    at 0x4AC92AA: msgsnd (msgsnd.c:25)
==640986==    by 0x14E178: queue_send (queue_utils.c:100)
==640986==    by 0x162643: UnknownInlinedFun (table-data.c:403)
==640986==    by 0x162643: UnknownInlinedFun (table-data.c:295)
==640986==    by 0x162643: UnknownInlinedFun (table-data.c:232)
==640986==    by 0x162643: UnknownInlinedFun (table-data.c:209)
==640986==    by 0x162643: UnknownInlinedFun (table-data.c:117)
==640986==    by 0x162643: UnknownInlinedFun (table-data.c:89)
==640986==    by 0x162643: copydb_copy_all_table_data (table-data.c:57)
==640986==    by 0x11406E: cli_copy_table_data (cli_copy.c:395)
==640986==    by 0x1118B2: UnknownInlinedFun (commandline.c:71)
==640986==    by 0x1118B2: main (main.c:142)
==640986==  Address 0x1ffeeec5f0 is on thread 1's stack
==640986==  in frame #2, created by copydb_copy_all_table_data (table-data.c:38)
==640986==
==640986== Syscall param msgsnd(msgp->mtext) points to uninitialised byte(s)
==640986==    at 0x4AC92AA: msgsnd (msgsnd.c:25)
==640986==    by 0x14E178: queue_send (queue_utils.c:100)
==640986==    by 0x162AD8: UnknownInlinedFun (table-data.c:380)
==640986==    by 0x162AD8: UnknownInlinedFun (table-data.c:307)
==640986==    by 0x162AD8: UnknownInlinedFun (table-data.c:232)
==640986==    by 0x162AD8: UnknownInlinedFun (table-data.c:209)
==640986==    by 0x162AD8: UnknownInlinedFun (table-data.c:117)
==640986==    by 0x162AD8: UnknownInlinedFun (table-data.c:89)
==640986==    by 0x162AD8: copydb_copy_all_table_data (table-data.c:57)
==640986==    by 0x11406E: cli_copy_table_data (cli_copy.c:395)
==640986==    by 0x1118B2: UnknownInlinedFun (commandline.c:71)
==640986==    by 0x1118B2: main (main.c:142)
==640986==  Address 0x1ffeeec5f0 is on thread 1's stack
==640986==  in frame #2, created by copydb_copy_all_table_data (table-data.c:38)
==640986==

...
```

The above error disappers with the fix.

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

* Add byte level details to list progress command

This patch allows reporting the total number of bytes transferred during
a copy operation. However, it has a limitation that the reported value
is only updated after each table is copied. This is because we do not
update table summary files while the copy is in progress.

Changes include:
- 3 new fields in table summary structures:
  - network.bytes: total number of bytes transmitted
  - network.bytes-pretty: pretty printed form of network.bytes
  - network.transmit-rate: pretty printed bytes transmitted per second
- Some new fields on the `pgcopydb list progress --json --summary`
  command:
  - steps[].network is a new field that exists for COPY step and has
    the following fields:
	- bytes: total number of bytes transmitted for all tables
	- bytes-pretty: pretty printed form of bytes
  - tables[].network is a new json value with the following fields:
	- bytes: total number of bytes transmitted for table
	- bytes-pretty: pretty printed form of bytes
	- transmit-rate: pretty printed bytes transmitted per second
- A new column on top level summary that shows total number of bytes
  copied.

* Address reviews

This commit addresses 2 issues raised in the review:
1. Using a SI standards when reporting data-rate units
2. Updating the relevant documentation with reproducible examples

The documentation update is done in a way that allowed me to run the
commands in a docker environment. Once I polish my changes in this
docker environment, I will update all the documentation with it and
share it in a separate PR.

* Use bit/s instead of byte/s for data-rate units

Before this update, exiting the process (either reaching ENDPOS or user-initiated abort like Ctrl-C) could leave us in the midst of an unfinished transaction. Upon resumption, PostgreSQL begins streaming from a consistent point, which might be the start of this incomplete transaction or from an earlier one. Our previous method tried to bypass all earlier messages until hitting the final message of the incomplete transaction, but this approach had its shortcomings, especially when the messages originated from several transactions back.

This commit introduces a more straightforward solution. Now, if the process is aborted and the last transaction is not complete, we'll issue a rollback command for that transaction. This ensures a clean and consistent state for a graceful exit and subsequent resumption.

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