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

Fix undefined connectionList bug and code style improvements

pull/366/head
Marco Slot 2016-03-10 21:23:01 +01:00 committed by Metin Doslu
parent ae7a42d957
commit f7dfc7df1b
1 changed files with 17 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -251,7 +251,7 @@ static ShardInterval * SearchCachedShardInterval(Datum partitionColumnValue,
ShardInterval **shardIntervalCache, ShardInterval **shardIntervalCache,
int shardCount, int shardCount,
FmgrInfo *compareFunction); FmgrInfo *compareFunction);
static void OpenShardConnections(CopyStmt *copyStatement, static void OpenCopyTransactions(CopyStmt *copyStatement,
ShardConnections *shardConnections, ShardConnections *shardConnections,
int64 shardId); int64 shardId);
static StringInfo ConstructCopyStatement(CopyStmt *copyStatement, int64 shardId); static StringInfo ConstructCopyStatement(CopyStmt *copyStatement, int64 shardId);
@ -481,7 +481,11 @@ CitusCopyFrom(CopyStmt *copyStatement, char *completionTag)
&found); &found);
if (!found) if (!found)
{ {
OpenShardConnections(copyStatement, shardConnections, shardId); /* intialize COPY transactions on shard placements */
shardConnections->shardId = shardId;
shardConnections->connectionList = NIL;
OpenCopyTransactions(copyStatement, shardConnections, shardId);
} }
/* get the (truncated) line buffer */ /* get the (truncated) line buffer */
@ -508,7 +512,6 @@ CitusCopyFrom(CopyStmt *copyStatement, char *completionTag)
PG_CATCH(); PG_CATCH();
{ {
EndCopyFrom(copyState); EndCopyFrom(copyState);
heap_close(rel, AccessShareLock);
/* roll back all transactions */ /* roll back all transactions */
connectionList = ConnectionList(shardConnectionHash); connectionList = ConnectionList(shardConnectionHash);
@ -521,21 +524,11 @@ CitusCopyFrom(CopyStmt *copyStatement, char *completionTag)
PG_END_TRY(); PG_END_TRY();
EndCopyFrom(copyState); EndCopyFrom(copyState);
heap_close(rel, AccessShareLock); heap_close(rel, NoLock);
error_context_stack = errorCallback.previous; error_context_stack = errorCallback.previous;
if (QueryCancelPending) CommitTransactions(connectionList);
{
AbortTransactions(connectionList);
ereport(ERROR, (errcode(ERRCODE_QUERY_CANCELED),
errmsg("canceling statement due to user request")));
}
else
{
CommitTransactions(connectionList);
}
CloseConnections(connectionList); CloseConnections(connectionList);
if (completionTag != NULL) if (completionTag != NULL)
@ -561,7 +554,7 @@ CreateShardConnectionHash(void)
info.entrysize = sizeof(ShardConnections); info.entrysize = sizeof(ShardConnections);
info.hash = tag_hash; info.hash = tag_hash;
shardConnectionsHash = hash_create("shardConnectionHash", shardConnectionsHash = hash_create("Shard Connections Hash",
INITIAL_CONNECTION_CACHE_SIZE, &info, INITIAL_CONNECTION_CACHE_SIZE, &info,
HASH_ELEM | HASH_FUNCTION); HASH_ELEM | HASH_FUNCTION);
@ -739,12 +732,13 @@ SearchCachedShardInterval(Datum partitionColumnValue, ShardInterval **shardInter
/* /*
* OpenShardConnections opens a connection for each placement of a shard. If a * OpenCopyTransactions opens a connection for each placement of a shard and
* connection cannot be opened, the shard placement is marked as inactive and * starts a COPY transaction. If a connection cannot be opened, then the shard
* the COPY continues with the remaining shard placements. * placement is marked as inactive and the COPY continues with the remaining
* shard placements.
*/ */
static void static void
OpenShardConnections(CopyStmt *copyStatement, ShardConnections *shardConnections, OpenCopyTransactions(CopyStmt *copyStatement, ShardConnections *shardConnections,
int64 shardId) int64 shardId)
{ {
List *finalizedPlacementList = NIL; List *finalizedPlacementList = NIL;
@ -816,7 +810,6 @@ OpenShardConnections(CopyStmt *copyStatement, ShardConnections *shardConnections
failedPlacement->nodeName, failedPlacement->nodePort); failedPlacement->nodeName, failedPlacement->nodePort);
} }
shardConnections->shardId = shardId;
shardConnections->connectionList = connectionList; shardConnections->connectionList = connectionList;
} }
@ -966,14 +959,8 @@ ConnectionList(HTAB *connectionHash)
shardConnections = (ShardConnections *) hash_seq_search(&status); shardConnections = (ShardConnections *) hash_seq_search(&status);
while (shardConnections != NULL) while (shardConnections != NULL)
{ {
ListCell *connectionCell = NULL; List *shardConnectionsList = list_copy(shardConnections->connectionList);
foreach(connectionCell, shardConnections->connectionList) connectionList = list_concat(connectionList, shardConnectionsList);
{
TransactionConnection *transactionConnection =
(TransactionConnection *) lfirst(connectionCell);
connectionList = lappend(connectionList, transactionConnection);
}
shardConnections = (ShardConnections *) hash_seq_search(&status); shardConnections = (ShardConnections *) hash_seq_search(&status);
} }
@ -985,7 +972,7 @@ ConnectionList(HTAB *connectionHash)
/* /*
* EndRemoteCopy ends the COPY input on all connections. If stopOnFailure * EndRemoteCopy ends the COPY input on all connections. If stopOnFailure
* is true, then EndRemoteCopy reports an error on failure, otherwise it * is true, then EndRemoteCopy reports an error on failure, otherwise it
* reports a warning. * reports a warning or continues.
*/ */
static void static void
EndRemoteCopy(List *connectionList, bool stopOnFailure) EndRemoteCopy(List *connectionList, bool stopOnFailure)