chore(query): use time base semaphore (#17879)

Author: zhang2014

src/meta/store/src/lib.rs

@@ -125,6 +125,44 @@ impl MetaStore {
             }
         }
     }
+
+    pub async fn new_acquired_by_time(
+        &self,
+        prefix: impl ToString,
+        capacity: u64,
+        id: impl ToString,
+        lease: Duration,
+    ) -> Result<Permit, AcquireError> {
+        match self {
+            MetaStore::L(v) => {
+                let mut local_lock_map = v.locks.lock().await;
+
+                let acquire_res = match local_lock_map.entry(prefix.to_string()) {
+                    Entry::Occupied(v) => v.get().clone(),
+                    Entry::Vacant(v) => v
+                        .insert(Arc::new(TokioSemaphore::new(capacity as usize)))
+                        .clone(),
+                };
+
+                match acquire_res.acquire_owned().await {
+                    Ok(guard) => Ok(Permit {
+                        stat: SharedAcquirerStat::new(),
+                        fu: Box::pin(async move {
+                            let _guard = guard;
+                            Ok(())
+                        }),
+                    }),
+                    Err(_e) => Err(AcquireError::ConnectionClosed(ConnectionClosed::new_str(
+                        "",
+                    ))),
+                }
+            }
+            MetaStore::R(grpc_client) => {
+                Semaphore::new_acquired_by_time(grpc_client.clone(), prefix, capacity, id, lease)
+                    .await
+            }
+        }
+    }
 }
 
 #[async_trait::async_trait]

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

@@ -175,7 +175,7 @@ impl<Data: QueueData> QueueManager<Data> {
             let start_time = SystemTime::now();
             let acquire_res = match self.global_statement_queue {
                 true => {
-                    let semaphore_acquire = self.meta_store.new_acquired(
+                    let semaphore_acquire = self.meta_store.new_acquired_by_time(
                         data.get_lock_key(),
                         self.permits as u64,
                         data.get_key(), // ID of this acquirer