changes_atop_v33_1_amit.patch

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 4f3c4efeff..e1d3dc99ca 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -716,8 +716,8 @@ HandleParallelApplyMessages(void)
 /*
  * Set up a dynamic shared memory segment.
  *
- * We set up a control region that contains a ParallelApplyWorkerShared, plus
- * one region to each of message queue and error queue.
+ * We set up a control region that contains a fixed worker info
+ * (ParallelApplyWorkerShared), a message queue, and an error queue.
  *
  * Returns true on success, false on failure.
  */
@@ -916,9 +916,9 @@ parallel_apply_get_in_xact(ParallelApplyWorkerShared *wshared)
  * Define a savepoint for a subxact in parallel apply worker if needed.
  *
  * The parallel apply worker can figure out if a new subtransaction was
- * started by checking if the new change arrived with different xid. In that
+ * started by checking if the new change arrived with a different xid. In that
  * case define a named savepoint, so that we are able to rollback to it
- * separately later.
+ * if required.
  */
 void
 parallel_apply_subxact_info_add(TransactionId current_xid, TransactionId top_xid)
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index d34de7f5ae..67ac2be818 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1346,6 +1346,11 @@ apply_handle_origin(StringInfo s)
 
 /*
  * Handle STREAM START message.
+ *
+ * XXX We can avoid sending pair of the START/STOP messages to the parallel
+ * worker because unlike apply worker it will process only one
+ * transaction-at-a-time. However, it is not clear whether that is worth the
+ * effort because it is sent after logical_decoding_work_mem changes.
  */
 static void
 apply_handle_stream_start(StringInfo s)
@@ -1442,7 +1447,7 @@ apply_handle_stream_start(StringInfo s)
 		case TRANS_PARALLEL_APPLY:
 
 			/*
-			 * Make sure the handle apply_dispatch methods are aware we're in
+			 * Make sure the apply_dispatch methods are aware that we're in
 			 * a remote transaction.
 			 */
 			in_remote_transaction = true;
@@ -4556,16 +4561,29 @@ get_transaction_apply_action(TransactionId xid, ParallelApplyWorkerInfo **winfo)
 	*winfo = NULL;
 
 	if (am_parallel_apply_worker())
+	{
 		return TRANS_PARALLEL_APPLY;
+	}
 	else if (in_remote_transaction)
+	{
+		/*
+		 * The in_remote_transaction can be set both for leader and parallel
+		 * apply worker but because of a prior check for the parallel worker,
+		 * we will reach here only for leader apply worker.
+		 */
 		return TRANS_LEADER_APPLY;
+	}
 
 	/*
 	 * Check if we are processing this transaction using a parallel apply
 	 * worker and if so, send the changes to that worker.
 	 */
 	else if ((*winfo = parallel_apply_find_worker(xid)))
+	{
 		return TRANS_LEADER_SEND_TO_PARALLEL;
+	}
 	else
+	{
 		return TRANS_LEADER_SERIALIZE;
+	}
 }