diff --git a/src/catalog/src/remote/client.rs b/src/catalog/src/remote/client.rs index 9d87fa37a9a..02d16926fc5 100644 --- a/src/catalog/src/remote/client.rs +++ b/src/catalog/src/remote/client.rs @@ -19,7 +19,8 @@ use std::time::Duration; use async_stream::stream; use common_error::prelude::BoxedError; -use common_meta::error::{Error, GetKvCacheSnafu, MetaSrvSnafu, Result}; +use common_meta::error::Error::{CacheNotGet, GetKvCache}; +use common_meta::error::{CacheNotGetSnafu, Error, MetaSrvSnafu, Result}; use common_meta::kv_backend::{Kv, KvBackend, KvBackendRef, ValueIter}; use common_meta::rpc::store::{ CompareAndPutRequest, DeleteRangeRequest, MoveValueRequest, PutRequest, RangeRequest, @@ -27,7 +28,7 @@ use common_meta::rpc::store::{ use common_telemetry::{info, timer}; use meta_client::client::MetaClient; use moka::future::{Cache, CacheBuilder}; -use snafu::ResultExt; +use snafu::{OptionExt, ResultExt}; use super::KvCacheInvalidator; use crate::metrics::{METRIC_CATALOG_KV_GET, METRIC_CATALOG_KV_REMOTE_GET}; @@ -36,7 +37,7 @@ const CACHE_MAX_CAPACITY: u64 = 10000; const CACHE_TTL_SECOND: u64 = 10 * 60; const CACHE_TTI_SECOND: u64 = 5 * 60; -pub type CacheBackendRef = Arc, Option>>; +pub type CacheBackendRef = Arc, Kv>>; pub struct CachedMetaKvBackend { kv_backend: KvBackendRef, cache: CacheBackendRef, @@ -58,19 +59,26 @@ impl KvBackend for CachedMetaKvBackend { let init = async { let _timer = timer!(METRIC_CATALOG_KV_REMOTE_GET); - - self.kv_backend.get(key).await + self.kv_backend.get(key).await.map(|val| { + val.with_context(|| CacheNotGetSnafu { + key: String::from_utf8_lossy(key), + }) + })? }; - self.cache - .try_get_with_by_ref(key, init) - .await - .map_err(|e| { - GetKvCacheSnafu { - err_msg: e.to_string(), - } - .build() - }) + // currently moka doesn't have `optionally_try_get_with_by_ref` + // TODO(fys): change to moka method when available + // https://github.com/moka-rs/moka/issues/254 + match self.cache.try_get_with_by_ref(key, init).await { + Ok(val) => Ok(Some(val)), + Err(e) => match e.as_ref() { + CacheNotGet { .. } => Ok(None), + _ => Err(e), + }, + } + .map_err(|e| GetKvCache { + err_msg: e.to_string(), + }) } async fn set(&self, key: &[u8], val: &[u8]) -> Result<()> { diff --git a/src/common/meta/src/error.rs b/src/common/meta/src/error.rs index 804b69dd6e4..90bc6c03e1a 100644 --- a/src/common/meta/src/error.rs +++ b/src/common/meta/src/error.rs @@ -60,7 +60,10 @@ pub enum Error { InvalidTableMetadata { err_msg: String, location: Location }, #[snafu(display("Failed to get kv cache, err: {}", err_msg))] - GetKvCache { err_msg: String, location: Location }, + GetKvCache { err_msg: String }, + + #[snafu(display("Get null from cache, key: {}", key))] + CacheNotGet { key: String, location: Location }, #[snafu(display("Failed to request MetaSrv, source: {}", source))] MetaSrv { @@ -82,7 +85,7 @@ impl ErrorExt for Error { | InvalidProtoMsg { .. } | InvalidTableMetadata { .. } => StatusCode::Unexpected, - SendMessage { .. } | GetKvCache { .. } => StatusCode::Internal, + SendMessage { .. } | GetKvCache { .. } | CacheNotGet { .. } => StatusCode::Internal, EncodeJson { .. } | DecodeJson { .. } | PayloadNotExist { .. } => { StatusCode::Unexpected diff --git a/src/frontend/src/instance.rs b/src/frontend/src/instance.rs index db885e65c13..c235f2f185a 100644 --- a/src/frontend/src/instance.rs +++ b/src/frontend/src/instance.rs @@ -47,7 +47,6 @@ use datanode::instance::sql::table_idents_to_full_name; use datanode::instance::InstanceRef as DnInstanceRef; use datatypes::schema::Schema; use distributed::DistInstance; -use futures::future; use meta_client::client::{MetaClient, MetaClientBuilder}; use meta_client::MetaClientOptions; use partition::manager::PartitionRuleManager; @@ -289,13 +288,10 @@ impl Instance { requests: InsertRequests, ctx: QueryContextRef, ) -> Result { - let _ = future::join_all( - requests - .inserts - .iter() - .map(|x| self.create_or_alter_table_on_demand(ctx.clone(), x)), - ) - .await; + for req in requests.inserts.iter() { + self.create_or_alter_table_on_demand(ctx.clone(), req) + .await?; + } let query = Request::Inserts(requests); GrpcQueryHandler::do_query(&*self.grpc_query_handler, query, ctx).await diff --git a/src/frontend/src/instance/distributed.rs b/src/frontend/src/instance/distributed.rs index f0c96170493..b4fe21e3b5a 100644 --- a/src/frontend/src/instance/distributed.rs +++ b/src/frontend/src/instance/distributed.rs @@ -658,7 +658,7 @@ impl GrpcQueryHandler for DistInstance { match expr { DdlExpr::CreateDatabase(expr) => self.handle_create_database(expr, ctx).await, DdlExpr::CreateTable(mut expr) => { - let _ = self.create_table(&mut expr, None).await; + let _ = self.create_table(&mut expr, None).await?; Ok(Output::AffectedRows(0)) } DdlExpr::Alter(expr) => self.handle_alter_table(expr).await, diff --git a/tests-integration/tests/region_failover.rs b/tests-integration/tests/region_failover.rs index 621fbb25dc1..e4a1eb64632 100644 --- a/tests-integration/tests/region_failover.rs +++ b/tests-integration/tests/region_failover.rs @@ -152,7 +152,7 @@ pub async fn test_region_failover(store_type: StorageType) { time::sleep(Duration::from_millis(100)).await; let cache = get_table_cache(&frontend, &cache_key); - assert!(cache.is_none()); + assert!(cache.unwrap().is_none()); let route_cache = get_route_cache(&frontend, &table_name); assert!(route_cache.is_none()); @@ -190,7 +190,7 @@ fn get_table_cache(instance: &Arc, key: &str) -> Option> { .unwrap(); let cache = kvbackend.cache(); - cache.get(key.as_bytes()) + Some(cache.get(key.as_bytes())) } fn get_route_cache(instance: &Arc, table_name: &TableName) -> Option> {