enable local execution for DROP command on distributed & reference tables

src/backend/distributed/master/master_delete_protocol.c

@@ -396,6 +396,7 @@ DropShards(Oid relationId, char *schemaName, char *relationName,
 
 	List *dropTaskList = DropTaskList(relationId, schemaName, relationName,
 									  deletableShardIntervalList);
+	bool shouldExecuteTasksLocally = ShouldExecuteTasksLocally(dropTaskList);
 
 	Task *task = NULL;
 	foreach_ptr(task, dropTaskList)
@@ -423,6 +424,30 @@ DropShards(Oid relationId, char *schemaName, char *relationName,
 				continue;
 			}
 
+			/*
+			 * If it is a local placement of a distributed table or a reference table,
+			 * then execute the DROP command locally.
+			 */
+			if (isLocalShardPlacement && shouldExecuteTasksLocally)
+			{
+				List *singleTaskList = list_make1(task);
+
+				ExecuteLocalUtilityTaskList(singleTaskList);
+			}
+			else
+			{
+				/*
+				 * Either it was not a local placement or we could not use
+				 * local execution even if it was a local placement.
+				 * If it is the second case, then it is possibly because in
+				 * current transaction, some commands or queries connected
+				 * to local group as well.
+				 *
+				 * Regardless of the node is a remote node or the current node,
+				 * try to open a new connection (or use an existing one) to
+				 * connect to that node to drop the shard placement over that
+				 * remote connection.
+				 */
 				const char *dropShardPlacementCommand = TaskQueryString(task);
 				ExecuteDropShardPlacementCommandRemotely(shardPlacement,
 														 relationName,
@@ -432,6 +457,7 @@ DropShards(Oid relationId, char *schemaName, char *relationName,
 				{
 					TransactionConnectedToLocalGroup = true;
 				}
+			}
 
 			DeleteShardPlacementRow(shardPlacementId);
 		}