Improve LogWithPrefix of mpp

dbms/src/DataStreams/ParallelAggregatingBlockInputStream.cpp

@@ -31,7 +31,7 @@ ParallelAggregatingBlockInputStream::ParallelAggregatingBlockInputStream(
     , keys_size(params.keys_size)
     , aggregates_size(params.aggregates_size)
     , handler(*this)
-    , processor(inputs, additional_input_at_end, max_threads, handler)
+    , processor(inputs, additional_input_at_end, max_threads, handler, log)
 {
     children = inputs;
     if (additional_input_at_end)

dbms/src/DataStreams/ParallelInputsProcessor.h

@@ -6,6 +6,7 @@
 #include <Common/ThreadManager.h>
 #include <Common/setThreadName.h>
 #include <DataStreams/IProfilingBlockInputStream.h>
+#include <Flash/Mpp/getMPPTaskLog.h>
 #include <common/logger_useful.h>
 
 #include <atomic>
@@ -81,11 +82,12 @@ class ParallelInputsProcessor
       * - where you must first make JOIN in parallel, while noting which keys are not found,
       *   and only after the completion of this work, create blocks of keys that are not found.
       */
-    ParallelInputsProcessor(const BlockInputStreams & inputs_, const BlockInputStreamPtr & additional_input_at_end_, size_t max_threads_, Handler & handler_)
+    ParallelInputsProcessor(const BlockInputStreams & inputs_, const BlockInputStreamPtr & additional_input_at_end_, size_t max_threads_, Handler & handler_, const LogWithPrefixPtr & log_)
         : inputs(inputs_)
         , additional_input_at_end(additional_input_at_end_)
         , max_threads(std::min(inputs_.size(), max_threads_))
         , handler(handler_)
+        , log(getMPPTaskLog(log_, "ParallelInputsProcessor"))
     {
         for (size_t i = 0; i < inputs_.size(); ++i)
             unprepared_inputs.emplace(inputs_[i], i);
@@ -99,7 +101,7 @@ class ParallelInputsProcessor
         }
         catch (...)
         {
-            tryLogCurrentException(__PRETTY_FUNCTION__);
+            tryLogCurrentException(log, __PRETTY_FUNCTION__);
         }
     }
 
@@ -352,7 +354,7 @@ class ParallelInputsProcessor
     /// Wait for the completion of all threads.
     std::atomic<bool> joined_threads{false};
 
-    Poco::Logger * log = &Poco::Logger::get("ParallelInputsProcessor");
+    const LogWithPrefixPtr log;
 };
 
 

dbms/src/DataStreams/SharedQueryBlockInputStream.h

@@ -36,7 +36,7 @@ class SharedQueryBlockInputStream : public IProfilingBlockInputStream
         }
         catch (...)
         {
-            tryLogCurrentException(__PRETTY_FUNCTION__);
+            tryLogCurrentException(log, __PRETTY_FUNCTION__);
         }
     }
 

dbms/src/DataStreams/UnionBlockInputStream.h

@@ -86,10 +86,10 @@ class UnionBlockInputStream final : public IProfilingBlockInputStream
         const LogWithPrefixPtr & log_,
         ExceptionCallback exception_callback_ = ExceptionCallback())
         : output_queue(std::min(inputs.size(), max_threads))
+        , log(getMPPTaskLog(log_, NAME))
         , handler(*this)
-        , processor(inputs, additional_input_at_end, max_threads, handler)
+        , processor(inputs, additional_input_at_end, max_threads, handler, log)
         , exception_callback(exception_callback_)
-        , log(getMPPTaskLog(log_, NAME))
     {
         children = inputs;
         if (additional_input_at_end)
@@ -117,7 +117,7 @@ class UnionBlockInputStream final : public IProfilingBlockInputStream
         }
         catch (...)
         {
-            tryLogCurrentException(__PRETTY_FUNCTION__);
+            tryLogCurrentException(log, __PRETTY_FUNCTION__);
         }
     }
 
@@ -312,6 +312,8 @@ class UnionBlockInputStream final : public IProfilingBlockInputStream
         Self & parent;
     };
 
+    LogWithPrefixPtr log;
+
     Handler handler;
     ParallelInputsProcessor<Handler, mode> processor;
 
@@ -321,8 +323,6 @@ class UnionBlockInputStream final : public IProfilingBlockInputStream
 
     bool started = false;
     bool all_read = false;
-
-    LogWithPrefixPtr log;
 };
 
 } // namespace DB

dbms/src/Flash/Coprocessor/DAGQueryBlockInterpreter.cpp

@@ -932,9 +932,11 @@ void DAGQueryBlockInterpreter::handleExchangeReceiver(DAGPipeline & pipeline)
         throw Exception("Can not find exchange receiver for " + query_block.source_name, ErrorCodes::LOGICAL_ERROR);
     // todo choose a more reasonable stream number
     auto & exchange_receiver_io_input_streams = dagContext().getInBoundIOInputStreamsMap()[query_block.source_name];
+    // In order to distinguish different exchange receivers.
+    auto executor_id_prefix_log = getMPPTaskLog(taskLogger(), query_block.source_name);
     for (size_t i = 0; i < max_streams; ++i)
     {
-        BlockInputStreamPtr stream = std::make_shared<ExchangeReceiverInputStream>(it->second, taskLogger());
+        BlockInputStreamPtr stream = std::make_shared<ExchangeReceiverInputStream>(it->second, executor_id_prefix_log);
         exchange_receiver_io_input_streams.push_back(stream);
         stream = std::make_shared<SquashingBlockInputStream>(stream, 8192, 0, taskLogger());
         pipeline.streams.push_back(stream);

dbms/src/Interpreters/Aggregator.cpp

@@ -61,7 +61,7 @@ AggregatedDataVariants::~AggregatedDataVariants()
         }
         catch (...)
         {
-            tryLogCurrentException(__PRETTY_FUNCTION__);
+            tryLogCurrentException(aggregator->log, __PRETTY_FUNCTION__);
         }
     }
 }