v1-allow-subxacts-in-parallel-workers.patch

diff --git a/doc/src/sgml/parallel.sgml b/doc/src/sgml/parallel.sgml
index 5acc9537d6..7237be40bd 100644
--- a/doc/src/sgml/parallel.sgml
+++ b/doc/src/sgml/parallel.sgml
@@ -546,9 +546,8 @@ EXPLAIN SELECT * FROM pgbench_accounts WHERE filler LIKE '%x%';
 
   <para>
     Functions and aggregates must be marked <literal>PARALLEL UNSAFE</literal> if
-    they write to the database, access sequences, change the transaction state
-    even temporarily (e.g., a PL/pgSQL function that establishes an
-    <literal>EXCEPTION</literal> block to catch errors), or make persistent changes to
+    they write to the database, access sequences, change the transaction state,
+    or make persistent changes to
     settings.  Similarly, functions must be marked <literal>PARALLEL
     RESTRICTED</literal> if they access temporary tables, client connection state,
     cursors, prepared statements, or miscellaneous backend-local state that
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 8442c5e6a7..2aa218638c 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -638,7 +638,9 @@ AssignTransactionId(TransactionState s)
 	 * operation, so we can't account for new XIDs at this point.
 	 */
 	if (IsInParallelMode() || IsParallelWorker())
-		elog(ERROR, "cannot assign XIDs during a parallel operation");
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_TRANSACTION_STATE),
+				 errmsg("cannot assign XIDs during a parallel operation")));
 
 	/*
 	 * Ensure parent(s) have XIDs, so that a child always has an XID later
@@ -826,7 +828,11 @@ GetCurrentCommandId(bool used)
 		 * could relax this restriction when currentCommandIdUsed was already
 		 * true at the start of the parallel operation.
 		 */
-		Assert(!IsParallelWorker());
+		if (IsParallelWorker())
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_TRANSACTION_STATE),
+					 errmsg("cannot modify data in a parallel worker")));
+
 		currentCommandIdUsed = true;
 	}
 	return currentCommandId;
@@ -1091,7 +1097,9 @@ CommandCounterIncrement(void)
 		 * point.
 		 */
 		if (IsInParallelMode() || IsParallelWorker())
-			elog(ERROR, "cannot start commands during a parallel operation");
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_TRANSACTION_STATE),
+					 errmsg("cannot start commands during a parallel operation")));
 
 		currentCommandId += 1;
 		if (currentCommandId == InvalidCommandId)
@@ -4227,7 +4235,7 @@ DefineSavepoint(const char *name)
 	 * is TBLOCK_PARALLEL_INPROGRESS, so we can treat that as an invalid case
 	 * below.)
 	 */
-	if (IsInParallelMode())
+	if (IsInParallelMode() || IsParallelWorker())
 		ereport(ERROR,
 				(errcode(ERRCODE_INVALID_TRANSACTION_STATE),
 				 errmsg("cannot define savepoints during a parallel operation")));
@@ -4314,7 +4322,7 @@ ReleaseSavepoint(const char *name)
 	 * is TBLOCK_PARALLEL_INPROGRESS, so we can treat that as an invalid case
 	 * below.)
 	 */
-	if (IsInParallelMode())
+	if (IsInParallelMode() || IsParallelWorker())
 		ereport(ERROR,
 				(errcode(ERRCODE_INVALID_TRANSACTION_STATE),
 				 errmsg("cannot release savepoints during a parallel operation")));
@@ -4423,7 +4431,7 @@ RollbackToSavepoint(const char *name)
 	 * is TBLOCK_PARALLEL_INPROGRESS, so we can treat that as an invalid case
 	 * below.)
 	 */
-	if (IsInParallelMode())
+	if (IsInParallelMode() || IsParallelWorker())
 		ereport(ERROR,
 				(errcode(ERRCODE_INVALID_TRANSACTION_STATE),
 				 errmsg("cannot rollback to savepoints during a parallel operation")));
@@ -4529,38 +4537,39 @@ RollbackToSavepoint(const char *name)
 /*
  * BeginInternalSubTransaction
  *		This is the same as DefineSavepoint except it allows TBLOCK_STARTED,
- *		TBLOCK_IMPLICIT_INPROGRESS, TBLOCK_END, and TBLOCK_PREPARE states,
- *		and therefore it can safely be used in functions that might be called
- *		when not inside a BEGIN block or when running deferred triggers at
- *		COMMIT/PREPARE time.  Also, it automatically does
- *		CommitTransactionCommand/StartTransactionCommand instead of expecting
- *		the caller to do it.
+ *		TBLOCK_IMPLICIT_INPROGRESS, TBLOCK_PARALLEL_INPROGRESS, TBLOCK_END,
+ *		and TBLOCK_PREPARE states, and therefore it can safely be used in
+ *		functions that might be called when not inside a BEGIN block or when
+ *		running deferred triggers at COMMIT/PREPARE time.  Also, it
+ *		automatically does CommitTransactionCommand/StartTransactionCommand
+ *		instead of expecting the caller to do it.
  */
 void
 BeginInternalSubTransaction(const char *name)
 {
 	TransactionState s = CurrentTransactionState;
+	bool		save_ExitOnAnyError = ExitOnAnyError;
 
 	/*
-	 * Workers synchronize transaction state at the beginning of each parallel
-	 * operation, so we can't account for new subtransactions after that
-	 * point. We might be able to make an exception for the type of
-	 * subtransaction established by this function, which is typically used in
-	 * contexts where we're going to release or roll back the subtransaction
-	 * before proceeding further, so that no enduring change to the
-	 * transaction state occurs. For now, however, we prohibit this case along
-	 * with all the others.
+	 * Errors within this function are improbable, but if one does happen we
+	 * force a FATAL exit.  Callers generally aren't prepared to handle losing
+	 * control, and moreover our transaction state is probably corrupted if we
+	 * fail partway through; so an ordinary ERROR longjmp isn't okay.
+	 */
+	ExitOnAnyError = true;
+
+	/*
+	 * We allow "internal" subtransactions inside parallel workers as long as
+	 * they don't require their own XIDs.  So, no need to check here;
+	 * enforcement occurs in AssignTransactionId().
 	 */
-	if (IsInParallelMode())
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_TRANSACTION_STATE),
-				 errmsg("cannot start subtransactions during a parallel operation")));
 
 	switch (s->blockState)
 	{
 		case TBLOCK_STARTED:
 		case TBLOCK_INPROGRESS:
 		case TBLOCK_IMPLICIT_INPROGRESS:
+		case TBLOCK_PARALLEL_INPROGRESS:
 		case TBLOCK_END:
 		case TBLOCK_PREPARE:
 		case TBLOCK_SUBINPROGRESS:
@@ -4579,7 +4588,6 @@ BeginInternalSubTransaction(const char *name)
 			/* These cases are invalid. */
 		case TBLOCK_DEFAULT:
 		case TBLOCK_BEGIN:
-		case TBLOCK_PARALLEL_INPROGRESS:
 		case TBLOCK_SUBBEGIN:
 		case TBLOCK_SUBRELEASE:
 		case TBLOCK_SUBCOMMIT:
@@ -4598,6 +4606,8 @@ BeginInternalSubTransaction(const char *name)
 
 	CommitTransactionCommand();
 	StartTransactionCommand();
+
+	ExitOnAnyError = save_ExitOnAnyError;
 }
 
 /*
@@ -4613,16 +4623,10 @@ ReleaseCurrentSubTransaction(void)
 	TransactionState s = CurrentTransactionState;
 
 	/*
-	 * Workers synchronize transaction state at the beginning of each parallel
-	 * operation, so we can't account for commit of subtransactions after that
-	 * point.  This should not happen anyway.  Code calling this would
-	 * typically have called BeginInternalSubTransaction() first, failing
-	 * there.
+	 * We do not check for parallel mode here.  If the subtransaction tried to
+	 * do anything that's forbidden in parallel mode, we'd have errored out
+	 * already.
 	 */
-	if (IsInParallelMode())
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_TRANSACTION_STATE),
-				 errmsg("cannot commit subtransactions during a parallel operation")));
 
 	if (s->blockState != TBLOCK_SUBINPROGRESS)
 		elog(ERROR, "ReleaseCurrentSubTransaction: unexpected state %s",
@@ -4647,11 +4651,9 @@ RollbackAndReleaseCurrentSubTransaction(void)
 	TransactionState s = CurrentTransactionState;
 
 	/*
-	 * Unlike ReleaseCurrentSubTransaction(), this is nominally permitted
-	 * during parallel operations.  That's because we may be in the leader,
-	 * recovering from an error thrown while we were in parallel mode.  We
-	 * won't reach here in a worker, because BeginInternalSubTransaction()
-	 * will have failed.
+	 * We do not check for parallel mode here.  If the subtransaction tried to
+	 * do anything that's forbidden in parallel mode, we'd have errored out
+	 * already.
 	 */
 
 	switch (s->blockState)
@@ -4698,6 +4700,7 @@ RollbackAndReleaseCurrentSubTransaction(void)
 	Assert(s->blockState == TBLOCK_SUBINPROGRESS ||
 		   s->blockState == TBLOCK_INPROGRESS ||
 		   s->blockState == TBLOCK_IMPLICIT_INPROGRESS ||
+		   s->blockState == TBLOCK_PARALLEL_INPROGRESS ||
 		   s->blockState == TBLOCK_STARTED);
 }