Cancel other processors on query exception (#15578)

* Add a test for canceling distributed queries on error

* Cancel other processors on exception in PipelineExecutor

Fixes: 01514_distributed_cancel_query_on_error

src/Processors/Executors/PipelineExecutor.cpp

@@ -566,7 +566,7 @@ void PipelineExecutor::executeStepImpl(size_t thread_num, size_t num_threads, st
             }
 
             if (node->exception)
-                finish();
+                cancel();
 
             if (finished)
                 break;

tests/queries/0_stateless/01514_distributed_cancel_query_on_error.sh

@@ -0,0 +1,21 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+. "$CURDIR"/../shell_config.sh
+
+# _shard_num:
+#   1 on 127.2
+#   2 on 127.3
+# max_block_size to fail faster
+# max_memory_usage/_shard_num/repeat() will allow failure on the first shard earlier.
+opts=(
+    --max_memory_usage=3G
+    --max_block_size=50
+    --max_threads=1
+    --max_distributed_connections=2
+)
+${CLICKHOUSE_CLIENT} "${opts[@]}" -q "SELECT groupArray(repeat('a', 1000*_shard_num)), number%100000 k from remote('127.{2,3}', system.numbers) GROUP BY k LIMIT 10e6" |& {
+    # the query should fail earlier on 127.3 and 127.2 should not even go to the memory limit exceeded error.
+    fgrep -q 'DB::Exception: Received from 127.3:9000. DB::Exception: Memory limit (for query) exceeded:'
+    # while if this will not correctly then it will got the exception from the 127.2:9000 and fail
+}