external
/
citus
mirror of https://github.com/citusdata/citus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove stopOnFailure flag from EndRemoteCopy()

pull/2706/head
Hadi Moshayedi 2019-05-11 06:13:52 -07:00
parent 39e806c276
commit b5c0ca45f1
3 changed files with 8 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
src/backend/distributed/commands/multi_copy.c
View File

@ -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();

2
src/backend/distributed/executor/intermediate_results.c
View File

@ -448,7 +448,7 @@ RemoteFileDestReceiverShutdown(DestReceiver *destReceiver)
}
/* close the COPY input */
EndRemoteCopy(0, connectionList, true);
EndRemoteCopy(0, connectionList);
if (resultDest->writeLocalFile)
{

2
src/include/distributed/commands/multi_copy.h
View File

@ -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);