changes_amit_2.patch

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index df3aad43e1..13513f2e57 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -7,43 +7,98 @@
  * IDENTIFICATION
  *	  src/backend/replication/logical/applyparallelworker.c
  *
- * This file contains routines that are intended to support setting up, using,
- * and tearing down a ParallelApplyWorkerInfo.
+ * This file contains the code to launch, set up, and teardown parallel apply
+ * worker which receives the changes from the leader worker and invokes routines
+ * to apply those on the subscriber database.
  *
- * Refer to the comments in the file header of logical/worker.c to see more
- * information about parallel apply workers.
+ * This file contains routines that are intended to support setting up, using
+ * and tearing down a ParallelApplyWorkerInfo which is required to communicate
+ * among leader and parallel apply workers.
+ *
+ * The parallel apply workers are assigned (if available) as soon as xact's
+ * first stream is received for subscriptions that have set their 'streaming'
+ * option as parallel. The leader apply worker will send changes to this new
+ * worker via shared memory. We keep this worker assigned till the transaction
+ * commit is received and also wait for the worker to finish at commit. This
+ * preserves commit ordering and avoid file I/O in most cases, although we
+ * still need to spill to a file if there is no worker available. See comments
+ * atop logical/worker to know more about streamed xacts whose changes are
+ * spilled to disk. It is important to maintain commit order to avoid failures
+ * due to (a) transaction dependencies, say if we insert a row in the first
+ * transaction and update it in the second transaction on publisher then
+ * allowing the subscriber to apply both in parallel can lead to failure in the
+ * update. (b) deadlocks, allowing transactions that update the same set of
+ * rows/tables in the opposite order to be applied in parallel can lead to
+ * deadlocks.
+ *
+ * We maintain a worker pool to avoid restarting workers for each streaming
+ * transaction. We maintain each worker's information in the
+ * ParallelApplyWorkersList. After successfully launching a new worker, its
+ * information is added to the ParallelApplyWorkersList. Once the worker
+ * finishes applying the transaction, we mark it available for re-use. Now,
+ * before starting a new worker to apply the streaming transaction, we check
+ * the list for any available worker. Note that we maintain a maximum of half
+ * the max_parallel_apply_workers_per_subscription workers in the pool and
+ * after that, we simply exit the worker after applying the transaction.
+ *
+ * XXX This worker pool threshold is a bit arbitrary and we can provide a GUC
+ * variable for this in the future if required.
+ *
+ * The leader apply worker will create a separate dynamic shared memory segment
+ * when each parallel apply worker starts. The reason for this design is that
+ * we cannot count how many workers will be started. It may be possible to
+ * allocate enough shared memory in one segment based on the maximum number of
+ * parallel apply workers (max_parallel_apply_workers_per_subscription), but
+ * this would waste memory if no process is actually started.
+ *
+ * The dynamic shared memory segment will contain (a) a shm_mq that is used to
+ * send changes in the transaction from leader apply worker to parallel apply
+ * worker (b) another shm_mq that is used to send errors (and other messages
+ * reported via elog/ereport) from the parallel apply worker to leader apply
+ * worker (c) necessary information to be shared among parallel apply workers
+ * and leader apply worker (i.e. members of ParallelApplyWorkerShared).
  *
  * Locking Considerations
- * ------------------
- * Since the database structure (schema of subscription tables, etc.) of
- * publisher and subscriber may be different, applying transactions in parallel
- * mode on the subscriber side may cause some deadlock problems that do not
- * occur on the publisher side.
+ * ----------------------
+ * Since the database structure (schema of subscription tables, constraints,
+ * etc.) of the publisher and subscriber could be different, applying
+ * transactions in parallel mode on the subscriber side can cause some
+ * deadlocks that do not occur on the publisher side which is expected and can
+ * happen even without parallel mode. In order to detect the deadlocks among
+ * leader and parallel apply workers, we need to ensure that we wait using lmgr
+ * locks, otherwise, such deadlocks won't be detected. The other approach was
+ * to not allow parallelism when the schema of tables is different between the
+ * publisher and subscriber but that would be too restrictive and would require
+ * the publisher to send much more information than it is currently sending.
  *
- * Suppose a subscribed table does not have a unique key on the publisher and
- * has a unique key on the subscriber.
+ * Consider a case where the subscribed table does not have a unique key on the
+ * publisher and has a unique key on the subscriber.
  *
  * 1) Deadlock between the leader apply worker and a parallel apply worker
  *
  * The parallel apply worker (PA) is executing TX-1 and the leader apply worker
  * (LA) is executing TX-2 concurrently on the subscriber. Now, LA is waiting
  * for PA because of the unique key of the subscribed table while PA is waiting
- * for LA to send further messages.
+ * for LA to send the next stream of changes or transaction finish command
+ * message.
  *
  * In order for lmgr to detect this, we have LA acquire a session lock on the
  * remote transaction (by pa_lock_stream()) and have PA wait on the lock before
- * trying to receive messages. In other words, LA acquires the lock before
- * sending STREAM_STOP and releases it if already acquired before sending
- * STREAM_START, STREAM_ABORT (for toplevel transaction), STREAM_PREPARE and
- * STREAM_COMMIT. For PA, it always needs to acquire the lock after processing
- * STREAM_STOP and STREAM_ABORT (for subtransaction) and then release
- * immediately after acquiring it. That way, when PA is waiting for LA, we can
- * have a wait-edge from PA to LA in lmgr, which will make a deadlock in lmgr
- * like:
+ * trying to receive the next stream of changes. Specifically, LA will acquire
+ * the lock before sending the STREAM_STOP and will release it if already
+ * acquired before sending the STREAM_START, STREAM_ABORT (for toplevel
+ * transaction), STREAM_PREPARE, and STREAM_COMMIT. The PA will acquire the
+ * lock after processing STREAM_STOP and STREAM_ABORT (for subtransaction) and
+ * then release the lock immediately after acquiring it.
  *
+ * The lock graph for the above example will look as follows:
  * LA (waiting to acquire the lock on the unique index) -> PA (waiting to
  * acquire the lock on the remote transaction) -> LA
  *
+ * This way, when PA is waiting for LA for the next stream of changes, we can
+ * have a wait-edge from PA to LA in lmgr, which will make us detect the
+ * deadlock between LA and PA.
+ *
  * 2) Deadlock between the leader apply worker and parallel apply workers
  *
  * This scenario is similar to the first case but TX-1 and TX-2 are executed by
@@ -51,38 +106,46 @@
  * PA-2 is waiting for PA-1 to complete its transaction while PA-1 is waiting
  * for subsequent input from LA. Also, LA is waiting for PA-2 to complete its
  * transaction in order to preserve the commit order. There is a deadlock among
- * three processes.
+ * the three processes.
+ *
+ * In order for lmgr to detect this, we have PA acquire a session lock (this is
+ * a different lock than referred in previous case, see
+ * pa_lock_transaction()) on the transaction being applied and have LA wait on
+ * the lock before proceeding in the transaction finish commands. Specifically,
+ * PA will acquire this lock in Share mode before executing the first message
+ * of the transaction and release it at the xact end. LA will acquire this lock
+ * in AccessExclusive mode at transaction finish commands (STREAM_COMMIT and
+ * STREAM_PREAPRE) and release it immediately.
  *
- * In order for lmgr to detect this, we have PA acquire a session lock on the
- * transaction being applied and have LA wait on the lock before proceeding in
- * transaction finish commands. That way, we can have a wait-edge from LA to PA
- * in lmgr, which will make a deadlock in lmgr like:
+ * The lock graph for the above example will look as follows:
+ * LA (waiting to acquire the transaction lock) -> PA-2 (waiting to acquire the
+ * lock due to unique index constraint) -> PA-1 (waiting to acquire the stream
+ * lock) -> LA
  *
- * LA (waiting to acquire the local transaction lock) -> PA-2 (waiting to
- * acquire the lock on the unique index) -> PA-1 (waiting to acquire the lock
- * on the remote transaction) -> LA
+ * This way when LA is waiting to finish the transaction end command to preserve
+ * the commit order, we will be able to detect deadlock, if any.
  *
  * One might think we can use XactLockTableWait(), but XactLockTableWait()
  * considers PREPARED TRANSACTION as still in progress which means the lock
- * won't be released even if the parallel apply worker prepared the transaction.
+ * won't be released even after the parallel apply worker has prepared the
+ * transaction.
  *
  * 3) Deadlock when the shm_mq buffer is full
  *
- * The scenario where LA has to wait is when the shm_mq buffer is full. In the
- * above scenario (ie. PA-1 and PA-2 are executing transactions concurrently),
- * if the shm_mq buffer between LA and PA-2 is full, LA has to wait to send
- * messages, and this wait doesn't appear in lmgr.
+ * In the previous scenario (ie. PA-1 and PA-2 are executing transactions
+ * concurrently), if the shm_mq buffer between LA and PA-2 is full, LA has to
+ * wait to send messages, and this wait doesn't appear in lmgr.
  *
  * To resolve this issue, we use non-blocking write and wait with a timeout. If
- * timeout is exceeded, the LA reports an error and restarts logical
+ * the timeout is exceeded, the LA reports an error and restarts logical
  * replication.
  *
- * 4) Lock type
+ * 4) Lock types
  *
  * Both the stream lock and the transaction lock mentioned above are
- * session-level locks, because both locks could be acquired outside the
+ * session-level locks because both locks could be acquired outside the
  * transaction, and the stream lock in the leader need to persist across
- * transaction boundaries until the end of the streaming transaction.
+ * transaction boundaries i.e until the end of the streaming transaction.
  *-------------------------------------------------------------------------
  */
 
@@ -158,7 +221,7 @@ static HTAB *ParallelApplyWorkersHash = NULL;
  * list entry is removed if there are already enough workers in the worker
  * pool either at the end of the transaction or while trying to find a free
  * worker for applying the transaction. For more information about the worker
- * pool, see comments atop worker.c.
+ * pool, see comments atop this file.
  */
 static List *ParallelApplyWorkersList = NIL;
 
@@ -365,7 +428,7 @@ pa_find_worker(TransactionId xid)
  * and frees the corresponding info. Otherwise it just marks the worker as
  * available for reuse.
  *
- * For more information about the worker pool, see comments atop worker.c.
+ * For more information about the worker pool, see comments atop this file.
  *
  * Returns true if the worker is stopped, false otherwise.
  */
@@ -726,9 +789,11 @@ ParallelApplyWorkerMain(Datum main_arg)
 	LogicalParallelApplyLoop(mqh);
 
 	/*
-	 * The parallel apply worker should not get here, because the parallel
-	 * apply worker should only stop when it receives a SIGTERM or SIGINT from
-	 * the leader or detects a change in subscription information.
+	 * The parallel apply worker must not get here because the parallel
+	 * apply worker will only stop when it receives a SIGTERM or SIGINT from
+	 * the leader, detects a change in subscription information, or when there
+	 * there is an error. None of these cases will allow the code to reach
+	 * here.
 	 */
 	Assert(false);
 }
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 56e3c6c1a2..d9ef50864e 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -24,52 +24,10 @@
  * Streamed transactions (large transactions exceeding a memory limit on the
  * upstream) are applied using one of two approaches:
  *
- * 1) Parallel apply workers
+ * 1) Write to temporary files and apply when the final commit arrives
  *
- * If streaming = parallel, we assign a new parallel apply worker (if
- * available) as soon as the xact's first stream is received. The leader apply
- * worker will send changes to this new worker via shared memory. We keep this
- * worker assigned till the transaction commit is received and also wait for
- * the worker to finish at commit. This preserves commit ordering and avoids
- * file I/O in most cases, although we still need to spill to a file if there
- * is no worker available. It is important to maintain commit order to avoid
- * failures due to (a) transaction dependencies, say if we insert a row in the
- * first transaction and update it in the second transaction then allowing to
- * apply both in parallel can lead to failure in the update. (b) deadlocks,
- * allowing transactions that update the same set of rows/tables in opposite
- * order to be applied in parallel can lead to deadlocks.
- *
- * We maintain a worker pool to avoid restarting workers for each streaming
- * transaction. We maintain each worker's information in the
- * ParallelApplyWorkersList. After successfully launching a new worker, its
- * information is added to the ParallelApplyWorkersList. Once the worker
- * finishes applying the transaction, we mark it available for re-use. Now,
- * before starting a new worker to apply the streaming transaction, we check
- * the list for any available worker. Note that we maintain a maximum of half
- * the max_parallel_apply_workers_per_subscription workers in the pool and
- * after that, we simply exit the worker after applying the transaction.
- *
- * XXX This worker pool threshold is a bit arbitrary and we can provide a GUC
- * variable for this in the future if required.
- *
- * The leader apply worker will create separate dynamic shared memory segment
- * when each parallel apply worker starts. The reason for this design is that
- * we cannot count how many workers will be started. It may be possible to
- * allocate enough shared memory in one segment based on the maximum number of
- * parallel apply workers (max_parallel_apply_workers_per_subscription), but this
- * would waste memory if no process is actually started.
- *
- * The dynamic shared memory segment will contain (a) a shm_mq that is used to
- * send changes in the transaction from leader apply worker to parallel apply
- * worker (b) another shm_mq that is used to send errors (and other messages
- * reported via elog/ereport) from the parallel apply worker to leader apply
- * worker (c) necessary information to be shared among parallel apply workers
- * and leader apply worker (i.e. members of ParallelApplyWorkerShared).
- *
- * If no parallel apply worker is available to handle the streamed transaction
- * we follow approach 2.
- *
- * 2) Write to temporary files and apply when the final commit arrives
+ * This approach is used when user has set subscription's streaming option as
+ * on.
  *
  * Unlike the regular (non-streamed) case, handling streamed transactions has
  * to handle aborts of both the toplevel transaction and subtransactions. This
@@ -96,6 +54,12 @@
  * the file we desired across multiple stream-open calls for the same
  * transaction.
  *
+ * 2) Parallel apply workers.
+ *
+ * This approach is used when user has set subscription's streaming option as
+ * parallel. See logical/applyparallelworker.c for information about this
+ * approach.
+ *
  * TWO_PHASE TRANSACTIONS
  * ----------------------
  * Two phase transactions are replayed at prepare and then committed or