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

remote shard update

m3hm3t/add-freeze-copy
Mehmet YILMAZ 2024-09-10 13:37:07 +00:00
parent d13ecedcfa
commit 812f9b182f
1 changed files with 47 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
src/backend/distributed/operations/worker_shard_copy.c
View File

@ -111,27 +111,35 @@ ConnectToRemoteAndStartCopy(ShardCopyDestReceiver *copyDest)
NULL /* database (current) */);
ClaimConnectionExclusively(copyDest->connection);
StartRemoteTransactionBegin(copyDest->connection);
/* Begin the remote transaction */
RemoteTransactionBegin(copyDest->connection);
SetupReplicationOriginRemoteSession(copyDest->connection);
/* Construct and send the TRUNCATE statement to the remote node */
/* Handle TRUNCATE or any setup commands */
StringInfo truncateStatement = ConstructShardTruncateStatement(
copyDest->destinationShardFullyQualifiedName);
if (!SendRemoteCommand(copyDest->connection, truncateStatement->data))
{
ReportConnectionError(copyDest->connection, ERROR);
RemoteTransactionAbort(copyDest->connection);
ResetRemoteTransaction(copyDest->connection);
}
PGresult *truncateResult = GetRemoteCommandResult(copyDest->connection, true);
if (PQresultStatus(truncateResult) != PGRES_COMMAND_OK)
if (!IsResponseOK(truncateResult))
{
ReportResultError(copyDest->connection, truncateResult, ERROR);
PQclear(truncateResult);
ForgetResults(copyDest->connection);
RemoteTransactionAbort(copyDest->connection);
ResetRemoteTransaction(copyDest->connection);
}
PQclear(truncateResult);
ForgetResults(copyDest->connection);
/* Construct the COPY command and send it to the remote node */
/* Construct and send the COPY statement with FREEZE */
StringInfo copyStatement = ConstructShardCopyStatement(
copyDest->destinationShardFullyQualifiedName,
copyDest->copyOutState->binary,
@ -140,16 +148,18 @@ ConnectToRemoteAndStartCopy(ShardCopyDestReceiver *copyDest)
if (!SendRemoteCommand(copyDest->connection, copyStatement->data))
{
ReportConnectionError(copyDest->connection, ERROR);
RemoteTransactionAbort(copyDest->connection);
ResetRemoteTransaction(copyDest->connection);
}
PGresult *result = GetRemoteCommandResult(copyDest->connection,
PGresult *copyResult = GetRemoteCommandResult(copyDest->connection,
true /* raiseInterrupts */);
if (PQresultStatus(result) != PGRES_COPY_IN)
if (PQresultStatus(copyResult ) != PGRES_COPY_IN)
{
ReportResultError(copyDest->connection, result, ERROR);
ReportResultError(copyDest->connection, copyResult, ERROR);
}
PQclear(result);
PQclear(copyResult);
}