chore: check lock revision before watch delete

src/query/service/src/locks/lock_manager.rs

@@ -26,6 +26,7 @@ use databend_common_catalog::lock::Lock;
 use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_meta_api::kv_pb_api::KVPbApi;
 use databend_common_meta_app::schema::CreateLockRevReq;
 use databend_common_meta_app::schema::DeleteLockRevReq;
 use databend_common_meta_app::schema::ExtendLockRevReq;
@@ -162,14 +163,15 @@ impl LockManager {
                 break;
             }
 
+            let elapsed = start.elapsed();
             // if no need retry, return error directly.
-            if !should_retry {
+            if !should_retry || elapsed >= duration {
                 catalog
                     .delete_lock_revision(delete_table_lock_req.clone())
                     .await?;
                 return Err(ErrorCode::TableAlreadyLocked(format!(
                     "table is locked by other session, please retry later(elapsed: {:?})",
-                    start.elapsed()
+                    elapsed
                 )));
             }
 
@@ -182,8 +184,18 @@ impl LockManager {
                 filter_type: FilterType::Delete.into(),
             };
             let mut watch_stream = meta_api.watch(req).await?;
+
+            let lock_meta = meta_api.get_pb(&watch_delete_ident).await?;
+            if lock_meta.is_none() {
+                log::warn!(
+                    "Lock revision '{}' already does not exist, skipping",
+                    rev_list[position - 1]
+                );
+                continue;
+            }
+
             // Add a timeout period for watch.
-            match timeout(duration, async move {
+            match timeout(duration.abs_diff(elapsed), async move {
                 while let Some(Ok(resp)) = watch_stream.next().await {
                     if let Some(event) = resp.event {
                         if event.current.is_none() {