add one more metric

src/Common/ProfileEvents.cpp

@@ -77,12 +77,14 @@
     M(GlobalThreadPoolThreadCreationMicroseconds, "Total time spent waiting for new threads to start.") \
     M(GlobalThreadPoolLockWaitMicroseconds,  "Total time threads have spent waiting for locks in the global thread pool.") \
     M(GlobalThreadPoolJobs, "Counts the number of jobs that have been pushed to the global thread pool.") \
+    M(GlobalThreadPoolJobWaitTimeMicroseconds, "Measures the elapsed time from when a job is scheduled in the thread pool to when it is picked up for execution by a worker thread. This metric helps identify delays in job processing, indicating the responsiveness of the thread pool to new tasks.") \
     M(LocalThreadPoolExpansions, "Counts the total number of times threads have been borrowed from the global thread pool to expand local thread pools.") \
     M(LocalThreadPoolShrinks,  "Counts the total number of times threads have been returned to the global thread pool from local thread pools.") \
     M(LocalThreadPoolThreadCreationMicroseconds, "Total time local thread pools have spent waiting to borrow a thread from the global pool.") \
     M(LocalThreadPoolLockWaitMicroseconds, "Total time threads have spent waiting for locks in the local thread pools.") \
     M(LocalThreadPoolJobs, "Counts the number of jobs that have been pushed to the local thread pools.") \
     M(LocalThreadPoolBusyMicroseconds, "Total time threads have spent executing the actual work.") \
+    M(LocalThreadPoolJobWaitTimeMicroseconds, "Measures the elapsed time from when a job is scheduled in the thread pool to when it is picked up for execution by a worker thread. This metric helps identify delays in job processing, indicating the responsiveness of the thread pool to new tasks.") \
     \
     M(DiskS3GetRequestThrottlerCount, "Number of DiskS3 GET and SELECT requests passed through throttler.") \
     M(DiskS3GetRequestThrottlerSleepMicroseconds, "Total time a query was sleeping to conform DiskS3 GET and SELECT request throttling.") \

src/Common/ThreadPool.cpp

@@ -38,13 +38,16 @@ namespace ProfileEvents
     extern const Event GlobalThreadPoolThreadCreationMicroseconds;
     extern const Event GlobalThreadPoolLockWaitMicroseconds;
     extern const Event GlobalThreadPoolJobs;
+    extern const Event GlobalThreadPoolJobWaitTimeMicroseconds;
 
     extern const Event LocalThreadPoolExpansions;
     extern const Event LocalThreadPoolShrinks;
     extern const Event LocalThreadPoolThreadCreationMicroseconds;
     extern const Event LocalThreadPoolLockWaitMicroseconds;
     extern const Event LocalThreadPoolJobs;
     extern const Event LocalThreadPoolBusyMicroseconds;
+    extern const Event LocalThreadPoolJobWaitTimeMicroseconds;
+
 }
 
 class JobWithPriority
@@ -60,6 +63,7 @@ class JobWithPriority
     /// Call stacks of all jobs' schedulings leading to this one
     std::vector<StackTrace::FramePointers> frame_pointers;
     bool enable_job_stack_trace = false;
+    Stopwatch job_create_time;
 
     JobWithPriority(
         Job job_, Priority priority_, CurrentMetrics::Metric metric,
@@ -79,6 +83,13 @@ class JobWithPriority
     {
         return priority > rhs.priority; // Reversed for `priority_queue` max-heap to yield minimum value (i.e. highest priority) first
     }
+
+    UInt64 elapsedMicroseconds() const
+    {
+        return job_create_time.elapsedMicroseconds();
+    }
+
+
 };
 
 
@@ -749,6 +760,10 @@ void ThreadPoolImpl<Thread>::worker(typename std::list<Thread>::iterator thread_
             job_data = std::move(const_cast<JobWithPriority &>(jobs.top()));
             jobs.pop();
 
+            ProfileEvents::increment(
+                std::is_same_v<Thread, std::thread> ? ProfileEvents::GlobalThreadPoolJobWaitTimeMicroseconds : ProfileEvents::LocalThreadPoolJobWaitTimeMicroseconds,
+                job_data->elapsedMicroseconds());
+
             /// We don't run jobs after `shutdown` is set, but we have to properly dequeue all jobs and finish them.
             if (shutdown)
             {