Merge branch 'main' into enlarge-sleep-limit

Author: mergify[bot]

Cargo.lock

@@ -22,7 +22,6 @@ common-proto-conv = { path = "../proto-conv" }
 common-protos = { path = "../protos" }
 common-tracing = { path = "../../common/tracing" }
 
-anyerror = "=0.1.6"
 anyhow = "1.0.58"
 async-trait = "0.1.56"
 enumflags2 = { version = "0.7.5", features = ["serde"] }

src/meta/api/Cargo.toml

@@ -14,13 +14,13 @@
 
 use std::fmt::Display;
 
-use anyerror::AnyError;
 use common_meta_app::schema::DatabaseId;
 use common_meta_app::schema::DatabaseIdToName;
 use common_meta_app::schema::DatabaseMeta;
 use common_meta_app::schema::DatabaseNameIdent;
 use common_meta_app::schema::TableNameIdent;
 use common_meta_app::share::*;
+use common_meta_types::anyerror::AnyError;
 use common_meta_types::app_error::AppError;
 use common_meta_types::app_error::ShareHasNoGrantedDatabase;
 use common_meta_types::app_error::UnknownDatabase;
@@ -32,6 +32,7 @@ use common_meta_types::txn_condition::Target;
 use common_meta_types::txn_op::Request;
 use common_meta_types::ConditionResult;
 use common_meta_types::MatchSeq;
+use common_meta_types::MetaBytesError;
 use common_meta_types::MetaError;
 use common_meta_types::Operation;
 use common_meta_types::TxnCondition;
@@ -107,7 +108,7 @@ pub async fn list_keys<K: KVApiKey>(
     let mut structured_keys = Vec::with_capacity(n);
 
     for (str_key, _seq_id) in res.iter() {
-        let struct_key = K::from_key(str_key).map_err(meta_encode_err)?;
+        let struct_key = K::from_key(str_key).map_err(to_bytes_err)?;
         structured_keys.push(struct_key);
     }
 
@@ -133,24 +134,24 @@ pub async fn list_u64_value<K: KVApiKey>(
     let mut values = Vec::with_capacity(n);
 
     for (str_key, seqv) in res.iter() {
-        let id = *deserialize_u64(&seqv.data).map_err(meta_encode_err)?;
+        let id = *deserialize_u64(&seqv.data)?;
         values.push(id);
 
         // Parse key
-        let struct_key = K::from_key(str_key).map_err(meta_encode_err)?;
+        let struct_key = K::from_key(str_key).map_err(to_bytes_err)?;
         structured_keys.push(struct_key);
     }
 
     Ok((structured_keys, values))
 }
 
-pub fn serialize_u64(value: impl Into<Id>) -> Result<Vec<u8>, MetaError> {
-    let v = serde_json::to_vec(&*value.into()).map_err(meta_encode_err)?;
+pub fn serialize_u64(value: impl Into<Id>) -> Result<Vec<u8>, MetaBytesError> {
+    let v = serde_json::to_vec(&*value.into())?;
     Ok(v)
 }
 
-pub fn deserialize_u64(v: &[u8]) -> Result<Id, MetaError> {
-    let id = serde_json::from_slice(v).map_err(meta_encode_err)?;
+pub fn deserialize_u64(v: &[u8]) -> Result<Id, MetaBytesError> {
+    let id = serde_json::from_slice(v)?;
     Ok(Id::new(id))
 }
 
@@ -173,30 +174,30 @@ pub async fn fetch_id<T: KVApiKey>(kv_api: &impl KVApi, generator: T) -> Result<
     Ok(seq_v.seq)
 }
 
-pub fn serialize_struct<T>(value: &T) -> Result<Vec<u8>, MetaError>
+pub fn serialize_struct<T>(value: &T) -> Result<Vec<u8>, MetaBytesError>
 where
     T: FromToProto + 'static,
     T::PB: common_protos::prost::Message,
 {
-    let p = value.to_pb().map_err(meta_encode_err)?;
+    let p = value.to_pb().map_err(to_bytes_err)?;
     let mut buf = vec![];
-    common_protos::prost::Message::encode(&p, &mut buf).map_err(meta_encode_err)?;
+    common_protos::prost::Message::encode(&p, &mut buf)?;
     Ok(buf)
 }
 
-pub fn deserialize_struct<T>(buf: &[u8]) -> Result<T, MetaError>
+pub fn deserialize_struct<T>(buf: &[u8]) -> Result<T, MetaBytesError>
 where
     T: FromToProto,
     T::PB: common_protos::prost::Message + Default,
 {
-    let p: T::PB = common_protos::prost::Message::decode(buf).map_err(meta_encode_err)?;
-    let v: T = FromToProto::from_pb(p).map_err(meta_encode_err)?;
+    let p: T::PB = common_protos::prost::Message::decode(buf)?;
+    let v: T = FromToProto::from_pb(p).map_err(to_bytes_err)?;
 
     Ok(v)
 }
 
-pub fn meta_encode_err<E: std::error::Error + 'static>(e: E) -> MetaError {
-    MetaError::EncodeError(AnyError::new(&e))
+pub fn to_bytes_err<E: std::error::Error + 'static>(e: E) -> MetaBytesError {
+    MetaBytesError::new(&e)
 }
 
 pub async fn send_txn(
@@ -485,6 +486,8 @@ where
     Ok((false, None))
 }
 
+/// Get existing value by key. Panic if key is absent.
+/// This function is only used for testing.
 pub async fn get_kv_data<T>(
     kv_api: &(impl KVApi + ?Sized),
     key: &impl KVApiKey,
@@ -495,11 +498,12 @@ where
 {
     let res = kv_api.get_kv(&key.to_key()).await?;
     if let Some(res) = res {
-        return deserialize_struct(&res.data);
+        let s = deserialize_struct(&res.data)?;
+        return Ok(s);
     };
 
-    Err(MetaError::SerdeError(AnyError::error(format!(
-        "get_kv {:?} fail",
+    Err(MetaError::Fatal(AnyError::error(format!(
+        "failed to get {}",
         key.to_key()
     ))))
 }

src/meta/api/src/kv_api_utils.rs

@@ -57,7 +57,6 @@ pub use kv_api_utils::is_all_db_data_removed;
 pub use kv_api_utils::is_db_need_to_be_remove;
 pub use kv_api_utils::list_keys;
 pub use kv_api_utils::list_u64_value;
-pub use kv_api_utils::meta_encode_err;
 pub use kv_api_utils::send_txn;
 pub use kv_api_utils::serialize_struct;
 pub use kv_api_utils::serialize_u64;

src/meta/api/src/lib.rs

@@ -107,7 +107,6 @@ use crate::get_u64_value;
 use crate::is_db_need_to_be_remove;
 use crate::list_keys;
 use crate::list_u64_value;
-use crate::meta_encode_err;
 use crate::send_txn;
 use crate::serialize_struct;
 use crate::serialize_u64;
@@ -818,8 +817,7 @@ impl<KV: KVApi> SchemaApi for KV {
 
         for (i, seq_meta_opt) in seq_metas.iter().enumerate() {
             if let Some(seq_meta) = seq_meta_opt {
-                let db_meta: DatabaseMeta =
-                    deserialize_struct(&seq_meta.data).map_err(meta_encode_err)?;
+                let db_meta: DatabaseMeta = deserialize_struct(&seq_meta.data)?;
 
                 let db_info = DatabaseInfo {
                     ident: DatabaseIdent {
@@ -1665,8 +1663,7 @@ impl<KV: KVApi> SchemaApi for KV {
 
         for (i, seq_meta_opt) in seq_tb_metas.iter().enumerate() {
             if let Some(seq_meta) = seq_meta_opt {
-                let tb_meta: TableMeta =
-                    deserialize_struct(&seq_meta.data).map_err(meta_encode_err)?;
+                let tb_meta: TableMeta = deserialize_struct(&seq_meta.data)?;
 
                 let tb_info = TableInfo {
                     ident: TableIdent {

src/meta/api/src/schema_api_impl.rs

@@ -2374,11 +2374,13 @@ impl SchemaApiTestSuite {
         for db_id in old_id_list.iter() {
             let id_key = DatabaseId { db_id: *db_id };
             let id_mapping = DatabaseIdToName { db_id: *db_id };
+
             let meta_res: Result<DatabaseMeta, MetaError> =
                 get_kv_data(mt.as_kv_api(), &id_key).await;
+            assert!(meta_res.is_err());
+
             let mapping_res: Result<DatabaseNameIdent, MetaError> =
                 get_kv_data(mt.as_kv_api(), &id_mapping).await;
-            assert!(meta_res.is_err());
             assert!(mapping_res.is_err());
         }
 

src/meta/api/src/schema_api_test_suite.rs

@@ -802,11 +802,11 @@ impl MetaGrpcClient {
         T: Into<MetaGrpcReadReq>,
         R: DeserializeOwned,
     {
-        let act: MetaGrpcReadReq = v.into();
+        let read_req: MetaGrpcReadReq = v.into();
 
-        debug!(req = debug(&act), "MetaGrpcClient::do_read request");
+        debug!(req = debug(&read_req), "MetaGrpcClient::do_read request");
 
-        let req: Request<RaftRequest> = act.clone().try_into()?;
+        let req: Request<RaftRequest> = read_req.clone().try_into()?;
 
         debug!(
             req = debug(&req),
@@ -826,7 +826,7 @@ impl MetaGrpcClient {
                 if status_is_retryable(&s) {
                     self.mark_as_unhealthy().await;
                     let mut client = self.make_client().await?;
-                    let req: Request<RaftRequest> = act.try_into()?;
+                    let req: Request<RaftRequest> = read_req.try_into()?;
                     let req = common_tracing::inject_span_to_tonic_request(req);
                     Ok(client.read_msg(req).await?.into_inner())
                 } else {

src/meta/grpc/src/grpc_client.rs

@@ -27,6 +27,7 @@ mod kv_message;
 mod log_entry;
 mod match_seq;
 mod message;
+mod meta_bytes_error;
 mod meta_errors;
 mod meta_errors_into;
 mod meta_management_error;
@@ -112,6 +113,7 @@ pub use message::ForwardRequestBody;
 pub use message::ForwardResponse;
 pub use message::JoinRequest;
 pub use message::LeaveRequest;
+pub use meta_bytes_error::MetaBytesError;
 pub use meta_errors::MetaError;
 pub use meta_errors::MetaResult;
 pub use meta_errors_into::ToMetaError;

src/meta/types/src/lib.rs

@@ -0,0 +1,55 @@
+// Copyright 2022 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use anyerror::AnyError;
+
+/// Errors that occur when encode/decode
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, thiserror::Error)]
+#[error("MetaBytesError: {source}")]
+pub struct MetaBytesError {
+    #[source]
+    pub source: AnyError,
+}
+
+impl MetaBytesError {
+    pub fn new(error: &(impl std::error::Error + 'static)) -> Self {
+        Self {
+            source: AnyError::new(error),
+        }
+    }
+}
+
+impl From<serde_json::Error> for MetaBytesError {
+    fn from(e: serde_json::Error) -> Self {
+        Self::new(&e)
+    }
+}
+
+impl From<std::string::FromUtf8Error> for MetaBytesError {
+    fn from(e: std::string::FromUtf8Error) -> Self {
+        Self::new(&e)
+    }
+}
+
+impl From<prost::EncodeError> for MetaBytesError {
+    fn from(e: prost::EncodeError) -> Self {
+        Self::new(&e)
+    }
+}
+
+impl From<prost::DecodeError> for MetaBytesError {
+    fn from(e: prost::DecodeError) -> Self {
+        Self::new(&e)
+    }
+}

src/meta/types/src/meta_bytes_error.rs

@@ -18,6 +18,7 @@ use serde::Serialize;
 use thiserror::Error;
 
 use crate::app_error::AppError;
+use crate::MetaBytesError;
 use crate::MetaNetworkError;
 use crate::MetaRaftError;
 use crate::MetaStorageError;
@@ -46,13 +47,8 @@ pub enum MetaError {
     #[error("{0}")]
     MetaServiceError(String),
 
-    /// type to represent serialize/deserialize errors
     #[error(transparent)]
-    SerdeError(AnyError),
-
-    /// Error when encoding auth
-    #[error(transparent)]
-    EncodeError(AnyError),
+    BytesError(#[from] MetaBytesError),
 
     #[error(transparent)]
     AppError(#[from] AppError),