From 4e7d5191e3d846f4130029f9412ac6fdfa3a3b61 Mon Sep 17 00:00:00 2001
From: Nikolai Kochetov <nik-kochetov@yandex-team.ru>
Date: Mon, 20 Sep 2021 22:07:33 +0300
Subject: [PATCH] Fix more tests

---
 src/Interpreters/InterpreterDescribeQuery.cpp            | 1 -
 src/Interpreters/ProcessList.cpp                         | 9 ++++-----
 src/Processors/Executors/CompletedPipelineExecutor.cpp   | 2 +-
 .../Executors/PushingAsyncPipelineExecutor.cpp           | 2 +-
 src/Processors/Executors/PushingPipelineExecutor.cpp     | 2 +-
 src/Storages/StorageLog.cpp                              | 2 +-
 6 files changed, 8 insertions(+), 10 deletions(-)

diff --git a/src/Interpreters/InterpreterDescribeQuery.cpp b/src/Interpreters/InterpreterDescribeQuery.cpp
index b3267d6303f..2ebae17cd6b 100644
--- a/src/Interpreters/InterpreterDescribeQuery.cpp
+++ b/src/Interpreters/InterpreterDescribeQuery.cpp
@@ -16,7 +16,6 @@
 #include <Parsers/ASTTablesInSelectQuery.h>
 #include <Parsers/TablePropertiesQueriesASTs.h>
 #include <DataTypes/NestedUtils.h>
-#include <Processors/Sources/SourceFromSingleChunk.h>
 
 
 namespace DB
diff --git a/src/Interpreters/ProcessList.cpp b/src/Interpreters/ProcessList.cpp
index 27ecc534763..98dd67b2e12 100644
--- a/src/Interpreters/ProcessList.cpp
+++ b/src/Interpreters/ProcessList.cpp
@@ -305,13 +305,12 @@ CancellationCode QueryStatus::cancelQuery(bool)
     if (is_killed.load())
         return CancellationCode::CancelSent;
 
-    SCOPE_EXIT({
-        std::lock_guard lock(executors_mutex);
-        for (auto * e : executors)
-            e->cancel();
-    });
+    std::lock_guard lock(executors_mutex);
+    for (auto * e : executors)
+        e->cancel();
 
     is_killed.store(true);
+
     return CancellationCode::CancelSent;
 }
 
diff --git a/src/Processors/Executors/CompletedPipelineExecutor.cpp b/src/Processors/Executors/CompletedPipelineExecutor.cpp
index e8f55bea936..ff7901c815d 100644
--- a/src/Processors/Executors/CompletedPipelineExecutor.cpp
+++ b/src/Processors/Executors/CompletedPipelineExecutor.cpp
@@ -18,7 +18,7 @@ CompletedPipelineExecutor::CompletedPipelineExecutor(QueryPipeline & pipeline_)
 
 void CompletedPipelineExecutor::execute()
 {
-    PipelineExecutor executor(pipeline.processors);
+    PipelineExecutor executor(pipeline.processors, pipeline.process_list_element);
     executor.execute(pipeline.getNumThreads());
 }
 
diff --git a/src/Processors/Executors/PushingAsyncPipelineExecutor.cpp b/src/Processors/Executors/PushingAsyncPipelineExecutor.cpp
index 00a5892151c..ee4b73c3222 100644
--- a/src/Processors/Executors/PushingAsyncPipelineExecutor.cpp
+++ b/src/Processors/Executors/PushingAsyncPipelineExecutor.cpp
@@ -163,7 +163,7 @@ void PushingAsyncPipelineExecutor::start()
     started = true;
 
     data = std::make_unique<Data>();
-    data->executor = std::make_shared<PipelineExecutor>(pipeline.processors);
+    data->executor = std::make_shared<PipelineExecutor>(pipeline.processors, pipeline.process_list_element);
     data->source = pushing_source.get();
 
     auto func = [&, thread_group = CurrentThread::getGroup()]()
diff --git a/src/Processors/Executors/PushingPipelineExecutor.cpp b/src/Processors/Executors/PushingPipelineExecutor.cpp
index 61aae02c068..2e2b5e9ca1e 100644
--- a/src/Processors/Executors/PushingPipelineExecutor.cpp
+++ b/src/Processors/Executors/PushingPipelineExecutor.cpp
@@ -85,7 +85,7 @@ void PushingPipelineExecutor::start()
         return;
 
     started = true;
-    executor = std::make_shared<PipelineExecutor>(pipeline.processors);
+    executor = std::make_shared<PipelineExecutor>(pipeline.processors, pipeline.process_list_element);
 
     if (!executor->executeStep(&need_data_flag))
         throw Exception(ErrorCodes::LOGICAL_ERROR,
diff --git a/src/Storages/StorageLog.cpp b/src/Storages/StorageLog.cpp
index 9e37acbdb0f..61dacf9b2ee 100644
--- a/src/Storages/StorageLog.cpp
+++ b/src/Storages/StorageLog.cpp
@@ -465,7 +465,7 @@ void LogSink::writeMarks(MarksForColumns && marks)
 
 StorageLog::~StorageLog()
 {
-    std::cerr << "======================\n" << StackTrace().toString() << std::endl;
+    // std::cerr << "======================\n" << StackTrace().toString() << std::endl;
 }
 
 StorageLog::StorageLog(