Do not try to send query if it was canceled in RemoteSource

With optimize_distributed_group_by_sharding_key it is possible to get
enough rows even before sending query to another shard and in this case
it shouldn't send anything to others.

This reduces NETWORK_ERROR in the 01563_distributed_query_finish from 2 to 1.

src/Processors/Sources/RemoteSource.cpp

@@ -21,6 +21,10 @@ RemoteSource::~RemoteSource() = default;
 
 Chunk RemoteSource::generate()
 {
+    /// onCancel() will do the cancel if the query was sent.
+    if (was_query_canceled)
+        return {};
+
     if (!was_query_sent)
     {
         /// Progress method will be called on Progress packet.
@@ -62,6 +66,7 @@ Chunk RemoteSource::generate()
 
 void RemoteSource::onCancel()
 {
+    was_query_canceled = true;
     query_executor->cancel();
 }
 

src/Processors/Sources/RemoteSource.h

@@ -36,6 +36,7 @@ protected:
     void onCancel() override;
 
 private:
+    bool was_query_canceled = false;
     bool was_query_sent = false;
     bool add_aggregation_info = false;
     RemoteQueryExecutorPtr query_executor;