From b5c0ca45f1f9695723899189426c5340eaa78a03 Mon Sep 17 00:00:00 2001
From: Hadi Moshayedi <hadi@moshayedi.net>
Date: Sat, 11 May 2019 06:13:52 -0700
Subject: [PATCH] Remove stopOnFailure flag from EndRemoteCopy()

---
 src/backend/distributed/commands/multi_copy.c | 21 ++++++-------------
 .../executor/intermediate_results.c           |  2 +-
 src/include/distributed/commands/multi_copy.h |  2 +-
 3 files changed, 8 insertions(+), 17 deletions(-)

diff --git a/src/backend/distributed/commands/multi_copy.c b/src/backend/distributed/commands/multi_copy.c
index 6cc0c5dca..d12ae4354 100644
--- a/src/backend/distributed/commands/multi_copy.c
+++ b/src/backend/distributed/commands/multi_copy.c
@@ -631,7 +631,7 @@ CopyToNewShards(CopyStmt *copyStatement, char *completionTag, Oid relationId)
 									  shardConnections->connectionList);
 			}
 
-			EndRemoteCopy(currentShardId, shardConnections->connectionList, true);
+			EndRemoteCopy(currentShardId, shardConnections->connectionList);
 			MasterUpdateShardStatistics(shardConnections->shardId);
 
 			copiedDataSizeInBytes = 0;
@@ -655,7 +655,7 @@ CopyToNewShards(CopyStmt *copyStatement, char *completionTag, Oid relationId)
 			SendCopyBinaryFooters(copyOutState, currentShardId,
 								  shardConnections->connectionList);
 		}
-		EndRemoteCopy(currentShardId, shardConnections->connectionList, true);
+		EndRemoteCopy(currentShardId, shardConnections->connectionList);
 		MasterUpdateShardStatistics(shardConnections->shardId);
 	}
 
@@ -1194,11 +1194,10 @@ SendCopyDataToPlacement(StringInfo dataBuffer, int64 shardId, MultiConnection *c
 
 /*
  * EndRemoteCopy ends the COPY input on all connections, and unclaims connections.
- * If stopOnFailure is true, then EndRemoteCopy reports an error on failure,
- * otherwise it reports a warning or continues.
+ * This reports an error on failure.
  */
 void
-EndRemoteCopy(int64 shardId, List *connectionList, bool stopOnFailure)
+EndRemoteCopy(int64 shardId, List *connectionList)
 {
 	ListCell *connectionCell = NULL;
 
@@ -1211,21 +1210,14 @@ EndRemoteCopy(int64 shardId, List *connectionList, bool stopOnFailure)
 		/* end the COPY input */
 		if (!PutRemoteCopyEnd(connection, NULL))
 		{
-			if (!stopOnFailure)
-			{
-				continue;
-			}
-
 			ereport(ERROR, (errcode(ERRCODE_IO_ERROR),
 							errmsg("failed to COPY to shard " INT64_FORMAT " on %s:%d",
 								   shardId, connection->hostname, connection->port)));
-
-			continue;
 		}
 
 		/* check whether there were any COPY errors */
 		result = GetRemoteCommandResult(connection, raiseInterrupts);
-		if (PQresultStatus(result) != PGRES_COMMAND_OK && stopOnFailure)
+		if (PQresultStatus(result) != PGRES_COMMAND_OK)
 		{
 			ReportCopyError(connection, result);
 		}
@@ -2456,8 +2448,7 @@ CitusCopyDestReceiverShutdown(DestReceiver *destReceiver)
 			}
 
 			/* close the COPY input on all shard placements */
-			EndRemoteCopy(shardConnections->shardId, shardConnections->connectionList,
-						  true);
+			EndRemoteCopy(shardConnections->shardId, shardConnections->connectionList);
 		}
 	}
 	PG_CATCH();
diff --git a/src/backend/distributed/executor/intermediate_results.c b/src/backend/distributed/executor/intermediate_results.c
index 8827fa79a..65ea09fb0 100644
--- a/src/backend/distributed/executor/intermediate_results.c
+++ b/src/backend/distributed/executor/intermediate_results.c
@@ -448,7 +448,7 @@ RemoteFileDestReceiverShutdown(DestReceiver *destReceiver)
 	}
 
 	/* close the COPY input */
-	EndRemoteCopy(0, connectionList, true);
+	EndRemoteCopy(0, connectionList);
 
 	if (resultDest->writeLocalFile)
 	{
diff --git a/src/include/distributed/commands/multi_copy.h b/src/include/distributed/commands/multi_copy.h
index 36a7cf978..907d19a10 100644
--- a/src/include/distributed/commands/multi_copy.h
+++ b/src/include/distributed/commands/multi_copy.h
@@ -131,7 +131,7 @@ extern void AppendCopyRowData(Datum *valueArray, bool *isNullArray,
 							  CopyCoercionData *columnCoercionPaths);
 extern void AppendCopyBinaryHeaders(CopyOutState headerOutputState);
 extern void AppendCopyBinaryFooters(CopyOutState footerOutputState);
-extern void EndRemoteCopy(int64 shardId, List *connectionList, bool stopOnFailure);
+extern void EndRemoteCopy(int64 shardId, List *connectionList);
 extern Node * ProcessCopyStmt(CopyStmt *copyStatement, char *completionTag,
 							  const char *queryString);
 extern void CheckCopyPermissions(CopyStmt *copyStatement);