Fix hard coded formatting strings for 64 bit numbers (#1831)

Postgres provides OS agnosting formatting macros for
formatting 64 bit numbers. Replaced %ld %lu with
INT64_FORMAT and UINT64_FORMAT respectively.

Also found some incorrect usages of formatting
flags and fixed them.

src/backend/distributed/commands/create_distributed_table.c

@@ -1261,13 +1261,13 @@ CopyLocalDataIntoShards(Oid distributedRelationId)
 
 		if (rowsCopied % 1000000 == 0)
 		{
-			ereport(DEBUG1, (errmsg("Copied %ld rows", rowsCopied)));
+			ereport(DEBUG1, (errmsg("Copied " UINT64_FORMAT " rows", rowsCopied)));
 		}
 	}
 
 	if (rowsCopied % 1000000 != 0)
 	{
-		ereport(DEBUG1, (errmsg("Copied %ld rows", rowsCopied)));
+		ereport(DEBUG1, (errmsg("Copied " UINT64_FORMAT " rows", rowsCopied)));
 	}
 
 	MemoryContextSwitchTo(oldContext);

src/backend/distributed/commands/multi_copy.c

@@ -1156,7 +1156,7 @@ SendCopyDataToPlacement(StringInfo dataBuffer, int64 shardId, MultiConnection *c
 	if (!PutRemoteCopyData(connection, dataBuffer->data, dataBuffer->len))
 	{
 		ereport(ERROR, (errcode(ERRCODE_IO_ERROR),
-						errmsg("failed to COPY to shard %ld on %s:%d",
+						errmsg("failed to COPY to shard " INT64_FORMAT " on %s:%d",
 							   shardId, connection->hostname, connection->port),
 						errdetail("failed to send %d bytes %s", dataBuffer->len,
 								  dataBuffer->data)));
@@ -1183,13 +1183,15 @@ EndRemoteCopy(int64 shardId, List *connectionList, bool stopOnFailure)
 		/* end the COPY input */
 		if (!PutRemoteCopyEnd(connection, NULL))
 		{
-			if (stopOnFailure)
+			if (!stopOnFailure)
 			{
-				ereport(ERROR, (errcode(ERRCODE_IO_ERROR),
+				continue;
-								errmsg("failed to COPY to shard %ld on %s:%d",
-									   shardId, connection->hostname, connection->port)));
 			}
 
+			ereport(ERROR, (errcode(ERRCODE_IO_ERROR),
+							errmsg("failed to COPY to shard " INT64_FORMAT " on %s:%d",
+								   shardId, connection->hostname, connection->port)));
+
 			continue;
 		}
 

src/backend/distributed/connection/placement_connection.c

@@ -511,9 +511,8 @@ FindPlacementListConnection(int flags, List *placementAccessList, const char *us
 
 			ereport(ERROR,
 					(errcode(ERRCODE_ACTIVE_SQL_TRANSACTION),
-					 errmsg(
+					 errmsg("cannot perform DDL on placement " UINT64_FORMAT
-						 "cannot perform DDL on placement %ld, which has been read over "
+							", which has been read over multiple connections",
-						 "multiple connections",
 							placement->placementId)));
 		}
 		else if (accessType == PLACEMENT_ACCESS_DDL && colocatedEntry != NULL &&
@@ -530,8 +529,8 @@ FindPlacementListConnection(int flags, List *placementAccessList, const char *us
 
 			ereport(ERROR,
 					(errcode(ERRCODE_ACTIVE_SQL_TRANSACTION),
-					 errmsg("cannot perform DDL on placement %ld since a co-located "
+					 errmsg("cannot perform DDL on placement " UINT64_FORMAT
-							"placement has been read over multiple connections",
+							" since a co-located placement has been read over multiple connections",
 							placement->placementId)));
 		}
 		else if (foundModifyingConnection)
@@ -584,8 +583,9 @@ FindPlacementListConnection(int flags, List *placementAccessList, const char *us
 
 			ereport(ERROR,
 					(errcode(ERRCODE_ACTIVE_SQL_TRANSACTION),
-					 errmsg("cannot establish a new connection for placement %ld, since "
+					 errmsg("cannot establish a new connection for "
-							"DDL has been executed on a connection that is in use",
+							"placement " UINT64_FORMAT
+							", since DDL has been executed on a connection that is in use",
 							placement->placementId)));
 		}
 		else if (placementConnection->hadDML)
@@ -606,8 +606,9 @@ FindPlacementListConnection(int flags, List *placementAccessList, const char *us
 
 			ereport(ERROR,
 					(errcode(ERRCODE_ACTIVE_SQL_TRANSACTION),
-					 errmsg("cannot establish a new connection for placement %ld, since "
+					 errmsg("cannot establish a new connection for "
-							"DML has been executed on a connection that is in use",
+							"placement " UINT64_FORMAT
+							", since DML has been executed on a connection that is in use",
 							placement->placementId)));
 		}
 		else if (accessType == PLACEMENT_ACCESS_DDL)

src/backend/distributed/executor/multi_client_executor.c

@@ -937,8 +937,8 @@ MultiClientWait(WaitInfo *waitInfo)
 		else if (rc == 0)
 		{
 			ereport(DEBUG5,
-					(errmsg("waiting for activity on tasks took longer than %ld ms",
+					(errmsg("waiting for activity on tasks took longer than %d ms",
-							(long) RemoteTaskCheckInterval * 10)));
+							(int) RemoteTaskCheckInterval * 10)));
 		}
 
 		/*

src/backend/distributed/executor/multi_task_tracker_executor.c

@@ -585,7 +585,7 @@ TaskHashEnter(HTAB *taskHash, Task *task)
 	/* if same node appears twice, we error-out */
 	if (handleFound)
 	{
-		ereport(ERROR, (errmsg("multiple entries for task: \"%d:%ld:%d\"",
+		ereport(ERROR, (errmsg("multiple entries for task: \"%d:" UINT64_FORMAT ":%u\"",
 							   task->taskType, task->jobId, task->taskId)));
 	}
 

src/backend/distributed/master/master_metadata_utility.c

@@ -311,8 +311,8 @@ ShardIntervalsOnWorkerGroup(WorkerNode *workerNode, Oid relationId)
 			if (metadataLock == false)
 			{
 				ereport(WARNING, (errcode(ERRCODE_LOCK_NOT_AVAILABLE),
-								  errmsg("lock is not acquired, size of shard %ld "
+								  errmsg("lock is not acquired, size of shard "
-										 "will be ignored", shardId)));
+										 UINT64_FORMAT " will be ignored", shardId)));
 				continue;
 			}
 
@@ -436,8 +436,9 @@ TableShardReplicationFactor(Oid relationId)
 			ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 							errmsg("cannot find the replication factor of the "
 								   "table %s", relationName),
-							errdetail("The shard %ld has different shards replication "
+							errdetail("The shard " UINT64_FORMAT
-									  "counts from other shards.", shardId)));
+									  " has different shards replication counts from "
+									  "other shards.", shardId)));
 		}
 	}
 

src/backend/distributed/metadata/metadata_sync.c

@@ -614,7 +614,8 @@ ShardListInsertCommand(List *shardIntervalList)
 			}
 
 			appendStringInfo(insertPlacementCommand,
-							 "(%lu, 1, %lu, %d, %lu)",
+							 "(" UINT64_FORMAT ", 1, " UINT64_FORMAT ", %d, "
+							 UINT64_FORMAT ")",
 							 shardId,
 							 placement->shardLength,
 							 placement->groupId,
@@ -664,7 +665,7 @@ ShardListInsertCommand(List *shardIntervalList)
 		}
 
 		appendStringInfo(insertShardCommand,
-						 "(%s::regclass, %lu, '%c', %s, %s)",
+						 "(%s::regclass, " UINT64_FORMAT ", '%c', %s, %s)",
 						 quote_literal_cstr(qualifiedRelationName),
 						 shardId,
 						 shardInterval->storageType,
@@ -700,7 +701,7 @@ ShardDeleteCommandList(ShardInterval *shardInterval)
 	/* create command to delete shard placements */
 	deletePlacementCommand = makeStringInfo();
 	appendStringInfo(deletePlacementCommand,
-					 "DELETE FROM pg_dist_placement WHERE shardid = %lu",
+					 "DELETE FROM pg_dist_placement WHERE shardid = " UINT64_FORMAT,
 					 shardId);
 
 	commandList = lappend(commandList, deletePlacementCommand->data);
@@ -708,7 +709,7 @@ ShardDeleteCommandList(ShardInterval *shardInterval)
 	/* create command to delete shard */
 	deleteShardCommand = makeStringInfo();
 	appendStringInfo(deleteShardCommand,
-					 "DELETE FROM pg_dist_shard WHERE shardid = %lu", shardId);
+					 "DELETE FROM pg_dist_shard WHERE shardid = " UINT64_FORMAT, shardId);
 
 	commandList = lappend(commandList, deleteShardCommand->data);
 

src/backend/distributed/planner/insert_select_planner.c

@@ -541,8 +541,9 @@ RouterModifyTaskForShardInterval(Query *originalQuery, ShardInterval *shardInter
 	/* ensure that we do not send queries where select is pruned away completely */
 	if (list_length(selectPlacementList) == 0)
 	{
-		ereport(DEBUG2, (errmsg("Skipping target shard interval %ld since "
+		ereport(DEBUG2, (errmsg("Skipping target shard interval " UINT64_FORMAT
-								"SELECT query for it pruned away", shardId)));
+								" since SELECT query for it pruned away",
+								shardId)));
 
 		return NULL;
 	}
@@ -562,7 +563,7 @@ RouterModifyTaskForShardInterval(Query *originalQuery, ShardInterval *shardInter
 						errmsg("cannot perform distributed planning for the given "
 							   "modification"),
 						errdetail("Insert query cannot be executed on all placements "
-								  "for shard %ld", shardId)));
+								  "for shard " UINT64_FORMAT "", shardId)));
 	}
 
 

src/backend/distributed/planner/multi_physical_planner.c

@@ -2398,8 +2398,8 @@ SubqueryTaskCreate(Query *originalQuery, ShardInterval *shardInterval,
 	/* ensure that we do not send queries where select is pruned away completely */
 	if (list_length(selectPlacementList) == 0)
 	{
-		ereport(DEBUG2, (errmsg("Skipping the target shard interval %ld because "
+		ereport(DEBUG2, (errmsg("Skipping the target shard interval " UINT64_FORMAT
-								"SELECT query is pruned away for the interval",
+								" because SELECT query is pruned away for the interval",
 								shardId)));
 
 		return NULL;

src/backend/distributed/transaction/distributed_deadlock_detection.c

@@ -599,7 +599,7 @@ LogCancellingBackend(TransactionNode *transactionNode)
 
 	appendStringInfo(logMessage, "Cancelling the following backend "
 								 "to resolve distributed deadlock "
-								 "(transaction numner = %ld, pid = %d)",
+								 "(transaction numner = " UINT64_FORMAT ", pid = %d)",
 					 transactionNode->transactionId.transactionNumber,
 					 transactionNode->initiatorProc->pid);
 
@@ -625,7 +625,8 @@ LogTransactionNode(TransactionNode *transactionNode)
 	logMessage = makeStringInfo();
 	transactionId = &(transactionNode->transactionId);
 
-	appendStringInfo(logMessage, "[DistributedTransactionId: (%d, %ld, %s)] = ",
+	appendStringInfo(logMessage,
+					 "[DistributedTransactionId: (%d, " UINT64_FORMAT ", %s)] = ",
 					 transactionId->initiatorNodeIdentifier,
 					 transactionId->transactionNumber,
 					 timestamptz_to_str(transactionId->timestamp));
@@ -685,7 +686,7 @@ WaitsForToString(List *waitsFor)
 			appendStringInfoString(transactionIdStr, ",");
 		}
 
-		appendStringInfo(transactionIdStr, "%ld",
+		appendStringInfo(transactionIdStr, UINT64_FORMAT,
 						 waitingNode->transactionId.transactionNumber);
 	}
 

src/backend/distributed/transaction/remote_transaction.c

@@ -85,7 +85,8 @@ StartRemoteTransactionBegin(struct MultiConnection *connection)
 	 */
 	distributedTransactionId = GetCurrentDistributedTransactionId();
 	appendStringInfo(beginAndSetDistributedTransactionId,
-					 "SELECT assign_distributed_transaction_id(%d, %ld, '%s');",
+					 "SELECT assign_distributed_transaction_id(%d, " UINT64_FORMAT
+					 ", '%s');",
 					 distributedTransactionId->initiatorNodeIdentifier,
 					 distributedTransactionId->transactionNumber,
 					 timestamptz_to_str(distributedTransactionId->timestamp));

src/backend/distributed/utils/citus_outfuncs.c

@@ -70,10 +70,6 @@
 #define WRITE_OID_FIELD(fldname) \
 	appendStringInfo(str, " :" CppAsString(fldname) " %u", node->fldname)
 
-/* Write a long-integer field */
-#define WRITE_LONG_FIELD(fldname) \
-	appendStringInfo(str, " :" CppAsString(fldname) " %ld", node->fldname)
-
 /* Write a char field (ie, one ascii character) */
 #define WRITE_CHAR_FIELD(fldname) \
 	appendStringInfo(str, " :" CppAsString(fldname) " %c", node->fldname)

src/backend/distributed/utils/colocation_utils.c

@@ -275,8 +275,9 @@ ErrorIfShardPlacementsNotColocated(Oid leftRelationId, Oid rightRelationId)
 		{
 			ereport(ERROR, (errmsg("cannot colocate tables %s and %s",
 								   leftRelationName, rightRelationName),
-							errdetail("Shard %ld of %s and shard %ld of %s "
+							errdetail("Shard " UINT64_FORMAT
-									  "have different number of shard placements.",
+									  " of %s and shard " UINT64_FORMAT
+									  " of %s have different number of shard placements.",
 									  leftShardId, leftRelationName,
 									  rightShardId, rightRelationName)));
 		}
@@ -307,8 +308,8 @@ ErrorIfShardPlacementsNotColocated(Oid leftRelationId, Oid rightRelationId)
 			{
 				ereport(ERROR, (errmsg("cannot colocate tables %s and %s",
 									   leftRelationName, rightRelationName),
-								errdetail("Shard %ld of %s and shard %ld of %s "
+								errdetail("Shard " UINT64_FORMAT " of %s and shard "
-										  "are not colocated.",
+										  UINT64_FORMAT " of %s are not colocated.",
 										  leftShardId, leftRelationName,
 										  rightShardId, rightRelationName)));
 			}

src/backend/distributed/utils/reference_table_utils.c

@@ -427,7 +427,8 @@ DeleteAllReferenceTablePlacementsFromNodeGroup(uint32 groupId)
 		DeleteShardPlacementRow(placement->placementId);
 
 		appendStringInfo(deletePlacementCommand,
-						 "DELETE FROM pg_dist_placement WHERE placementid=%lu",
+						 "DELETE FROM pg_dist_placement WHERE placementid = "
+						 UINT64_FORMAT,
 						 placement->placementId);
 		SendCommandToWorkers(WORKERS_WITH_METADATA, deletePlacementCommand->data);
 	}

src/backend/distributed/worker/worker_data_fetch_protocol.c

@@ -1380,10 +1380,10 @@ AlterSequenceMinMax(Oid sequenceId, char *schemaName, char *sequenceName)
 		 * DefElem->arg can only hold literal ints up to int4, in order to represent
 		 * larger numbers we need to construct a float represented as a string.
 		 */
-		appendStringInfo(startNumericString, "%lu", startValue);
+		appendStringInfo(startNumericString, INT64_FORMAT, startValue);
 		startFloatArg = (Node *) makeFloat(startNumericString->data);
 
-		appendStringInfo(maxNumericString, "%lu", maxValue);
+		appendStringInfo(maxNumericString, INT64_FORMAT, maxValue);
 		maxFloatArg = (Node *) makeFloat(maxNumericString->data);
 
 		SetDefElemArg(alterSequenceStatement, "start", startFloatArg);

src/include/distributed/master_protocol.h

@@ -72,9 +72,10 @@
 #define CREATE_SCHEMA_COMMAND "CREATE SCHEMA IF NOT EXISTS %s AUTHORIZATION %s"
 #define CREATE_EMPTY_SHARD_QUERY "SELECT master_create_empty_shard('%s')"
 #define FINALIZED_SHARD_PLACEMENTS_QUERY \
-	"SELECT placementid, nodename, nodeport FROM pg_dist_shard_placement WHERE shardstate = 1 AND shardid = %ld"
+	"SELECT placementid, nodename, nodeport FROM pg_dist_shard_placement WHERE shardstate = 1 AND shardid = " \
+	INT64_FORMAT
 #define UPDATE_SHARD_STATISTICS_QUERY \
-	"SELECT master_update_shard_statistics(%ld)"
+	"SELECT master_update_shard_statistics(" INT64_FORMAT ")"
 #define PARTITION_METHOD_QUERY "SELECT part_method FROM master_get_table_metadata('%s');"
 
 /* Enumeration that defines the shard placement policy to use while staging */

src/include/distributed/metadata_sync.h

@@ -43,10 +43,13 @@ extern void CreateTableMetadataOnWorkers(Oid relationId);
 	"SELECT worker_drop_distributed_table(logicalrelid) FROM pg_dist_partition"
 #define DISABLE_DDL_PROPAGATION "SET citus.enable_ddl_propagation TO 'off'"
 #define WORKER_APPLY_SEQUENCE_COMMAND "SELECT worker_apply_sequence_command (%s)"
-#define UPSERT_PLACEMENT "INSERT INTO pg_dist_placement " \
+#define UPSERT_PLACEMENT \
+	"INSERT INTO pg_dist_placement " \
 	"(shardid, shardstate, shardlength, " \
 	"groupid, placementid) " \
-						 "VALUES (%lu, %d, %lu, %d, %lu) " \
+	"VALUES (" UINT64_FORMAT ", %d, " UINT64_FORMAT \
+	", %d, " UINT64_FORMAT \
+	") " \
 	"ON CONFLICT (placementid) DO UPDATE SET " \
 	"shardid = EXCLUDED.shardid, " \
 	"shardstate = EXCLUDED.shardstate, " \