diff --git a/src/common/function/src/admin/gc.rs b/src/common/function/src/admin/gc.rs index ad2e3d3890..b5e067ce72 100644 --- a/src/common/function/src/admin/gc.rs +++ b/src/common/function/src/admin/gc.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::time::Duration; - use common_error::ext::BoxedError; use common_macro::admin_fn; use common_meta::rpc::procedure::{GcRegionsRequest, GcTableRequest}; @@ -30,7 +28,6 @@ use snafu::{ResultExt, ensure}; use crate::handlers::ProcedureServiceHandlerRef; use crate::helper::cast_u64; -const DEFAULT_GC_TIMEOUT: Duration = Duration::from_secs(60); const DEFAULT_FULL_FILE_LISTING: bool = false; #[admin_fn( @@ -50,7 +47,7 @@ pub(crate) async fn gc_regions( .gc_regions(GcRegionsRequest { region_ids, full_file_listing, - timeout: DEFAULT_GC_TIMEOUT, + timeout: None, }) .await?; @@ -77,7 +74,7 @@ pub(crate) async fn gc_table( schema_name, table_name, full_file_listing, - timeout: DEFAULT_GC_TIMEOUT, + timeout: None, }) .await?; @@ -180,9 +177,18 @@ fn gc_table_signature() -> Signature { #[cfg(test)] mod tests { + use std::sync::{Arc, Mutex}; + + use api::v1::meta::ReconcileRequest; + use async_trait::async_trait; + use catalog::CatalogManagerRef; + use common_meta::rpc::procedure::{ + GcResponse, ManageRegionFollowerRequest, MigrateRegionRequest, ProcedureStateResponse, + }; use session::context::QueryContext; use super::*; + use crate::handlers::ProcedureServiceHandler; #[test] fn test_parse_gc_regions_params_with_full_file_listing() { @@ -217,4 +223,80 @@ mod tests { assert_eq!(table, "t"); assert!(full_file_listing); } + + #[tokio::test] + async fn test_gc_regions_uses_meta_gc_timeout_config() { + let handler = Arc::new(MockProcedureServiceHandler::default()); + let handler_ref: ProcedureServiceHandlerRef = handler.clone(); + let params = vec![ValueRef::UInt64(1), ValueRef::Boolean(true)]; + + super::gc_regions(&handler_ref, &QueryContext::arc(), ¶ms) + .await + .unwrap(); + + let request = handler.gc_regions_request.lock().unwrap().clone().unwrap(); + assert_eq!(request.region_ids, vec![1]); + assert!(request.full_file_listing); + assert_eq!(request.timeout, None); + } + + #[tokio::test] + async fn test_gc_table_uses_meta_gc_timeout_config() { + let handler = Arc::new(MockProcedureServiceHandler::default()); + let handler_ref: ProcedureServiceHandlerRef = handler.clone(); + let params = vec![ValueRef::String("public.t"), ValueRef::Boolean(true)]; + + super::gc_table(&handler_ref, &QueryContext::arc(), ¶ms) + .await + .unwrap(); + + let request = handler.gc_table_request.lock().unwrap().clone().unwrap(); + assert_eq!(request.catalog_name, "greptime"); + assert_eq!(request.schema_name, "public"); + assert_eq!(request.table_name, "t"); + assert!(request.full_file_listing); + assert_eq!(request.timeout, None); + } + + #[derive(Default)] + struct MockProcedureServiceHandler { + gc_regions_request: Mutex>, + gc_table_request: Mutex>, + } + + #[async_trait] + impl ProcedureServiceHandler for MockProcedureServiceHandler { + async fn migrate_region(&self, _request: MigrateRegionRequest) -> Result> { + unreachable!() + } + + async fn reconcile(&self, _request: ReconcileRequest) -> Result> { + unreachable!() + } + + async fn query_procedure_state(&self, _pid: &str) -> Result { + unreachable!() + } + + async fn manage_region_follower( + &self, + _request: ManageRegionFollowerRequest, + ) -> Result<()> { + unreachable!() + } + + fn catalog_manager(&self) -> &CatalogManagerRef { + unreachable!() + } + + async fn gc_regions(&self, request: GcRegionsRequest) -> Result { + *self.gc_regions_request.lock().unwrap() = Some(request); + Ok(GcResponse::default()) + } + + async fn gc_table(&self, request: GcTableRequest) -> Result { + *self.gc_table_request.lock().unwrap() = Some(request); + Ok(GcResponse::default()) + } + } } diff --git a/src/common/meta/src/rpc/procedure.rs b/src/common/meta/src/rpc/procedure.rs index baab254fef..fcdea07df5 100644 --- a/src/common/meta/src/rpc/procedure.rs +++ b/src/common/meta/src/rpc/procedure.rs @@ -82,7 +82,7 @@ pub struct RemoveRegionFollowerRequest { pub struct GcRegionsRequest { pub region_ids: Vec, pub full_file_listing: bool, - pub timeout: Duration, + pub timeout: Option, } #[derive(Debug, Clone)] @@ -91,7 +91,7 @@ pub struct GcTableRequest { pub schema_name: String, pub table_name: String, pub full_file_listing: bool, - pub timeout: Duration, + pub timeout: Option, } #[derive(Debug, Clone, Default, PartialEq, Eq)] diff --git a/src/meta-client/src/client/procedure.rs b/src/meta-client/src/client/procedure.rs index 63f45c28fb..7658653e13 100644 --- a/src/meta-client/src/client/procedure.rs +++ b/src/meta-client/src/client/procedure.rs @@ -286,7 +286,7 @@ impl Inner { }), region_ids: request.region_ids, full_file_listing: request.full_file_listing, - timeout_secs: timeout.as_secs() as u32, + timeout_secs: gc_timeout_secs(timeout), }; let resp: GcRegionsResponse = self @@ -294,7 +294,9 @@ impl Inner { "gc_regions", move |mut client| { let mut req = Request::new(req.clone()); - req.set_timeout(timeout); + if let Some(timeout) = timeout { + req.set_timeout(timeout); + } async move { client.gc_regions(req).await.map(|res| res.into_inner()) } }, |resp: &GcRegionsResponse| &resp.header, @@ -323,7 +325,7 @@ impl Inner { schema_name: request.schema_name, table_name: request.table_name, full_file_listing: request.full_file_listing, - timeout_secs: timeout.as_secs() as u32, + timeout_secs: gc_timeout_secs(timeout), }; let resp: GcTableResponse = self @@ -331,7 +333,9 @@ impl Inner { "gc_table", move |mut client| { let mut req = Request::new(req.clone()); - req.set_timeout(timeout); + if let Some(timeout) = timeout { + req.set_timeout(timeout); + } async move { client.gc_table(req).await.map(|res| res.into_inner()) } }, |resp: &GcTableResponse| &resp.header, @@ -413,6 +417,12 @@ impl Inner { } } +fn gc_timeout_secs(timeout: Option) -> u32 { + timeout + .map(|timeout| timeout.as_secs().max(1).try_into().unwrap_or(u32::MAX)) + .unwrap_or(0) +} + #[cfg(test)] mod tests { use std::time::{Duration, Instant}; @@ -438,8 +448,18 @@ mod tests { use tonic::codec::CompressionEncoding; use tonic::{Request, Response, Status}; + use super::gc_timeout_secs; use crate::client::MetaClientBuilder; + #[test] + fn test_gc_timeout_secs() { + assert_eq!(gc_timeout_secs(None), 0); + assert_eq!(gc_timeout_secs(Some(Duration::from_millis(1))), 1); + assert_eq!(gc_timeout_secs(Some(Duration::from_millis(999))), 1); + assert_eq!(gc_timeout_secs(Some(Duration::from_secs(1))), 1); + assert_eq!(gc_timeout_secs(Some(Duration::from_secs(10))), 10); + } + #[derive(Clone)] struct MockHeartbeat { leader_addr: String, diff --git a/src/meta-srv/src/service/procedure.rs b/src/meta-srv/src/service/procedure.rs index 25add392a6..adfd8c9bac 100644 --- a/src/meta-srv/src/service/procedure.rs +++ b/src/meta-srv/src/service/procedure.rs @@ -268,7 +268,7 @@ impl procedure_service_server::ProcedureService for Metasrv { .handle_gc_regions(MetaGcRegionsRequest { region_ids, full_file_listing, - timeout: Duration::from_secs(timeout_secs as u64), + timeout: Self::normalize_gc_timeout(Duration::from_secs(timeout_secs as u64)), }) .await?; @@ -295,7 +295,7 @@ impl procedure_service_server::ProcedureService for Metasrv { schema_name, table_name, full_file_listing, - timeout: Duration::from_secs(timeout_secs as u64), + timeout: Self::normalize_gc_timeout(Duration::from_secs(timeout_secs as u64)), }) .await?; @@ -356,9 +356,8 @@ impl Metasrv { &self, region_ids: Vec, full_file_listing: bool, - timeout: Duration, + timeout: Option, ) -> error::Result { - let timeout = Self::normalize_gc_timeout(timeout); let gc_ticker = self.gc_ticker().context(error::UnexpectedSnafu { violated: "GC ticker not available".to_string(), })?;