chore: stop caching None in CachedMetaKvBackend (#1871)

* chore: dont cache none

* fix: test case

* chore: add comment

* chore: minor rewrite

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<Cache<Vec<u8>, Option<Kv>>>;
+pub type CacheBackendRef = Arc<Cache<Vec<u8>, 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<()> {

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

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<Output> {
-        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

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,

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<Instance>, key: &str) -> Option<Option<Kv>> {
         .unwrap();
     let cache = kvbackend.cache();
 
-    cache.get(key.as_bytes())
+    Some(cache.get(key.as_bytes()))
 }
 
 fn get_route_cache(instance: &Arc<Instance>, table_name: &TableName) -> Option<Arc<TableRoute>> {