Added information about thread number to trace log

dbms/src/Common/QueryProfiler.cpp

@@ -82,13 +82,16 @@ namespace
                                     8 * sizeof(char) + // maximum VarUInt length for string size
                                     QUERY_ID_MAX_LEN * sizeof(char) + // maximum query_id length
                                     sizeof(StackTrace) + // collected stack trace
-                                    sizeof(TimerType); // timer type
+                                    sizeof(TimerType) + // timer type
+                                    sizeof(UInt32); // thread_number
         char buffer[buf_size];
         WriteBufferDiscardOnFailure out(trace_pipe.fds_rw[1], buf_size, buffer);
 
         StringRef query_id = CurrentThread::getQueryId();
         query_id.size = std::min(query_id.size, QUERY_ID_MAX_LEN);
 
+        UInt32 thread_number = CurrentThread::get().thread_number;
+
         const auto signal_context = *reinterpret_cast<ucontext_t *>(context);
         const StackTrace stack_trace(signal_context);
 
@@ -96,6 +99,7 @@ namespace
         writeStringBinary(query_id, out);
         writePODBinary(stack_trace, out);
         writePODBinary(timer_type, out);
+        writePODBinary(thread_number, out);
         out.next();
     }
 

dbms/src/Common/TraceCollector.cpp

@@ -48,14 +48,15 @@ TraceCollector::TraceCollector(std::shared_ptr<TraceLog> & trace_log)
 
     /** Increase pipe size to avoid slowdown during fine-grained trace collection.
       */
+    constexpr int max_pipe_capacity_to_set = 1048576;
     int pipe_size = fcntl(trace_pipe.fds_rw[1], F_GETPIPE_SZ);
     if (-1 == pipe_size)
         throwFromErrno("Cannot get pipe capacity", ErrorCodes::CANNOT_FCNTL);
-    for (errno = 0; errno != EPERM && pipe_size <= 1048576; pipe_size *= 2)
+    for (errno = 0; errno != EPERM && pipe_size < max_pipe_capacity_to_set; pipe_size *= 2)
         if (-1 == fcntl(trace_pipe.fds_rw[1], F_SETPIPE_SZ, pipe_size * 2) && errno != EPERM)
             throwFromErrno("Cannot increase pipe capacity to " + toString(pipe_size * 2), ErrorCodes::CANNOT_FCNTL);
 
-    LOG_TRACE(log, "Pipe capacity is " << formatReadableSizeWithBinarySuffix(pipe_size));
+    LOG_TRACE(log, "Pipe capacity is " << formatReadableSizeWithBinarySuffix(std::min(pipe_size, max_pipe_capacity_to_set)));
 
     thread = ThreadFromGlobalPool(&TraceCollector::run, this);
 }
@@ -104,10 +105,12 @@ void TraceCollector::run()
         std::string query_id;
         StackTrace stack_trace(NoCapture{});
         TimerType timer_type;
+        UInt32 thread_number;
 
         readStringBinary(query_id, in);
         readPODBinary(stack_trace, in);
         readPODBinary(timer_type, in);
+        readPODBinary(thread_number, in);
 
         const auto size = stack_trace.getSize();
         const auto & frames = stack_trace.getFrames();
@@ -117,7 +120,7 @@ void TraceCollector::run()
         for (size_t i = 0; i < size; i++)
             trace.emplace_back(UInt64(reinterpret_cast<uintptr_t>(frames[i])));
 
-        TraceLogElement element{std::time(nullptr), timer_type, query_id, trace};
+        TraceLogElement element{std::time(nullptr), timer_type, thread_number, query_id, trace};
 
         trace_log->add(element);
     }

dbms/src/Interpreters/TraceLog.cpp

@@ -21,12 +21,13 @@ Block TraceLogElement::createBlock()
         {std::make_shared<DataTypeDate>(),                                    "event_date"},
         {std::make_shared<DataTypeDateTime>(),                                "event_time"},
         {std::make_shared<TimerDataType>(timer_values),                       "timer_type"},
+        {std::make_shared<DataTypeUInt32>(),                                  "thread_number"},
         {std::make_shared<DataTypeString>(),                                  "query_id"},
         {std::make_shared<DataTypeArray>(std::make_shared<DataTypeUInt64>()), "trace"}
     };
 }
 
-void TraceLogElement::appendToBlock(Block &block) const
+void TraceLogElement::appendToBlock(Block & block) const
 {
     MutableColumns columns = block.mutateColumns();
 
@@ -35,6 +36,7 @@ void TraceLogElement::appendToBlock(Block &block) const
     columns[i++]->insert(DateLUT::instance().toDayNum(event_time));
     columns[i++]->insert(event_time);
     columns[i++]->insert(static_cast<UInt8>(timer_type));
+    columns[i++]->insert(thread_number);
     columns[i++]->insertData(query_id.data(), query_id.size());
     columns[i++]->insert(trace);
 

dbms/src/Interpreters/TraceLog.h

@@ -16,6 +16,7 @@ struct TraceLogElement
 
     time_t event_time{};
     TimerType timer_type{};
+    UInt32 thread_number{};
     String query_id{};
     Array trace{};