feat: enhance resource scheduling logs with clear status and configuration details

- Add detailed status information for workload group and warehouse limits
- Display configuration parameter names (max_concurrency, max_running_queries)
- Show real-time resource usage (X/Y slots used, queue length)
- Improve user experience by clearly identifying bottlenecks and stages
- Enable faster troubleshooting and optimization of resource limits

Author: BohuTANG

src/query/service/src/sessions/queue_mgr.rs

@@ -229,25 +229,35 @@ impl<Data: QueueData> QueueManager<Data> {
                     workload_group_timeout = std::cmp::min(queue_timeout, workload_group_timeout);
                 }
 
-                data.set_status("[QUERY-QUEUE] Waiting for local workload semaphore");
-
                 let semaphore = workload_group.semaphore.clone();
-                let acquire = tokio::time::timeout(timeout, semaphore.acquire_owned());
+                let used_slots = permits - semaphore.available_permits();
+                data.set_status(&format!("[BLOCKED] Workload group '{}' local limit (max_concurrency={}): {}/{} slots used", 
+                    workload_group.meta.name, permits, used_slots, permits));
+
+                let acquire = tokio::time::timeout(timeout, semaphore.clone().acquire_owned());
                 let queue_future = AcquireQueueFuture::create(data.clone(), acquire, self.clone());
 
                 guards.push(queue_future.await?);
 
+                let used_slots_after = permits - semaphore.available_permits();
                 info!(
-                    "[QUERY-QUEUE] Successfully acquired from local workload semaphore. elapsed: {:?}",
-                    instant.elapsed()
+                    "[ACQUIRED] Workload group '{}' local (max_concurrency={}): {}/{} slots used (waited {:.2}s)",
+                    workload_group.meta.name,
+                    permits,
+                    used_slots_after,
+                    permits,
+                    instant.elapsed().as_secs_f64()
                 );
 
                 timeout -= instant.elapsed();
 
                 // Prevent concurrent access to meta and serialize the submission of acquire requests.
                 // This ensures that at most permits + nodes acquirers will be in the queue at any given time.
                 let _guard = workload_group.mutex.clone().lock_owned().await;
-                data.set_status("[QUERY-QUEUE] Waiting for global workload semaphore");
+                data.set_status(&format!(
+                    "[BLOCKED] Workload group '{}' global limit: acquiring distributed lock",
+                    workload_group.meta.name
+                ));
 
                 let workload_queue_guard = self
                     .acquire_workload_queue(
@@ -259,15 +269,21 @@ impl<Data: QueueData> QueueManager<Data> {
                     .await?;
 
                 info!(
-                    "[QUERY-QUEUE] Successfully acquired from global workload semaphore. elapsed: {:?}",
-                    instant.elapsed()
+                    "[ACQUIRED] Workload group '{}' global: distributed lock acquired (waited {:.2}s)",
+                    workload_group.meta.name,
+                    instant.elapsed().as_secs_f64()
                 );
                 timeout -= instant.elapsed();
                 guards.push(workload_queue_guard);
             }
         }
 
-        data.set_status("[QUERY-QUEUE] Waiting for warehouse resource scheduling");
+        let available_permits = self.semaphore.available_permits();
+        let used_slots = self.permits - available_permits;
+        let queue_length = self.length();
+
+        data.set_status(&format!("[BLOCKED] Warehouse limit (max_running_queries={}): {}/{} slots used, {} queries waiting", 
+            self.permits, used_slots, self.permits, queue_length));
 
         guards.extend(self.acquire_warehouse_queue(data, timeout).await?);
 
@@ -339,6 +355,8 @@ impl<Data: QueueData> QueueManager<Data> {
 
                 // Prevent concurrent access to meta and serialize the submission of acquire requests.
                 // This ensures that at most permits + nodes acquirers will be in the queue at any given time.
+                data.set_status(&format!("[BLOCKED] Global statement queue (max_running_queries={}): {}/{} slots used, acquiring cluster lock", 
+                    self.permits, self.permits - self.semaphore.available_permits(), self.permits));
                 let _guard = self.tokio_mutex.clone().lock_owned().await;
                 let acquire = tokio::time::timeout(timeout, semaphore_acquire);
                 let queue_future = AcquireQueueFuture::create(data, acquire, self.clone());
@@ -351,9 +369,17 @@ impl<Data: QueueData> QueueManager<Data> {
         match acquire_res {
             Ok(v) => {
                 guards.push(v);
+                let available_permits_after = self.semaphore.available_permits();
+                let used_slots_after = self.permits - available_permits_after;
+                let queue_length_after = self.length();
+
+                if self.global_statement_queue && guards.len() > 1 {
+                    info!("[ACQUIRED] Global statement queue: cluster lock acquired");
+                }
+
                 info!(
-                    "[QUERY-QUEUE] Successfully acquired from queue, current length: {}",
-                    self.length()
+                    "[ACQUIRED] Warehouse (max_running_queries={}): {}/{} slots used, {} queries waiting",
+                    self.permits, used_slots_after, self.permits, queue_length_after
                 );
 
                 Ok(guards)