build: bump rust edition to 2024 (#6920)

* bump edition

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* format

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* gen keyword

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* lifetime and env var

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* one more gen fix

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* lifetime of temporaries in tail expressions

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* format again

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* clippy nested if

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* clippy let and return

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

---------

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

Cargo.toml

@@ -74,7 +74,7 @@ resolver = "2"
 
 [workspace.package]
 version = "0.18.0"
-edition = "2021"
+edition = "2024"
 license = "Apache-2.0"
 
 [workspace.lints]

src/api/src/error.rs

@@ -19,8 +19,8 @@ use common_error::status_code::StatusCode;
 use common_macro::stack_trace_debug;
 use common_time::timestamp::TimeUnit;
 use datatypes::prelude::ConcreteDataType;
-use snafu::prelude::*;
 use snafu::Location;
+use snafu::prelude::*;
 
 pub type Result<T> = std::result::Result<T, Error>;
 

src/api/src/helper.rs

@@ -16,15 +16,15 @@ use std::collections::HashSet;
 use std::sync::Arc;
 
 use common_base::BitVec;
-use common_decimal::decimal128::{DECIMAL128_DEFAULT_SCALE, DECIMAL128_MAX_PRECISION};
 use common_decimal::Decimal128;
+use common_decimal::decimal128::{DECIMAL128_DEFAULT_SCALE, DECIMAL128_MAX_PRECISION};
 use common_time::time::Time;
 use common_time::timestamp::TimeUnit;
 use common_time::{Date, IntervalDayTime, IntervalMonthDayNano, IntervalYearMonth, Timestamp};
 use datatypes::prelude::{ConcreteDataType, ValueRef};
 use datatypes::scalars::ScalarVector;
 use datatypes::types::{
-    Int16Type, Int8Type, IntervalType, TimeType, TimestampType, UInt16Type, UInt8Type,
+    Int8Type, Int16Type, IntervalType, TimeType, TimestampType, UInt8Type, UInt16Type,
 };
 use datatypes::value::{OrderedF32, OrderedF64, Value};
 use datatypes::vectors::{
@@ -295,7 +295,7 @@ impl TryFrom<ConcreteDataType> for ColumnDataTypeWrapper {
             | ConcreteDataType::Struct(_)
             | ConcreteDataType::Dictionary(_)
             | ConcreteDataType::Duration(_) => {
-                return error::IntoColumnDataTypeSnafu { from: datatype }.fail()
+                return error::IntoColumnDataTypeSnafu { from: datatype }.fail();
             }
         };
         let datatype_extension = match column_datatype {

src/api/src/v1/column_def.rs

@@ -15,9 +15,9 @@
 use std::collections::HashMap;
 
 use datatypes::schema::{
-    ColumnDefaultConstraint, ColumnSchema, FulltextAnalyzer, FulltextBackend, FulltextOptions,
-    SkippingIndexOptions, SkippingIndexType, COMMENT_KEY, FULLTEXT_KEY, INVERTED_INDEX_KEY,
-    SKIPPING_INDEX_KEY,
+    COMMENT_KEY, ColumnDefaultConstraint, ColumnSchema, FULLTEXT_KEY, FulltextAnalyzer,
+    FulltextBackend, FulltextOptions, INVERTED_INDEX_KEY, SKIPPING_INDEX_KEY, SkippingIndexOptions,
+    SkippingIndexType,
 };
 use greptime_proto::v1::{
     Analyzer, FulltextBackend as PbFulltextBackend, SkippingIndexType as PbSkippingIndexType,

src/auth/src/common.rs

@@ -17,13 +17,13 @@ use std::sync::Arc;
 use common_base::secrets::SecretString;
 use digest::Digest;
 use sha1::Sha1;
-use snafu::{ensure, OptionExt};
+use snafu::{OptionExt, ensure};
 
 use crate::error::{IllegalParamSnafu, InvalidConfigSnafu, Result, UserPasswordMismatchSnafu};
 use crate::user_info::DefaultUserInfo;
-use crate::user_provider::static_user_provider::{StaticUserProvider, STATIC_USER_PROVIDER};
+use crate::user_provider::static_user_provider::{STATIC_USER_PROVIDER, StaticUserProvider};
 use crate::user_provider::watch_file_user_provider::{
-    WatchFileUserProvider, WATCH_FILE_USER_PROVIDER,
+    WATCH_FILE_USER_PROVIDER, WatchFileUserProvider,
 };
 use crate::{UserInfoRef, UserProviderRef};
 

src/auth/src/lib.rs

@@ -22,13 +22,13 @@ mod user_provider;
 pub mod tests;
 
 pub use common::{
-    auth_mysql, static_user_provider_from_option, user_provider_from_option, userinfo_by_name,
-    HashedPassword, Identity, Password,
+    HashedPassword, Identity, Password, auth_mysql, static_user_provider_from_option,
+    user_provider_from_option, userinfo_by_name,
 };
 pub use permission::{PermissionChecker, PermissionReq, PermissionResp};
 pub use user_info::UserInfo;
-pub use user_provider::static_user_provider::StaticUserProvider;
 pub use user_provider::UserProvider;
+pub use user_provider::static_user_provider::StaticUserProvider;
 
 /// pub type alias
 pub type UserInfoRef = std::sync::Arc<dyn UserInfo>;

src/auth/src/tests.rs

@@ -21,7 +21,7 @@ use crate::error::{
     UserPasswordMismatchSnafu,
 };
 use crate::user_info::DefaultUserInfo;
-use crate::{auth_mysql, Identity, Password, UserInfoRef, UserProvider};
+use crate::{Identity, Password, UserInfoRef, UserProvider, auth_mysql};
 
 pub struct DatabaseAuthInfo<'a> {
     pub catalog: &'a str,

src/auth/src/user_provider.rs

@@ -22,7 +22,7 @@ use std::io::BufRead;
 use std::path::Path;
 
 use common_base::secrets::ExposeSecret;
-use snafu::{ensure, OptionExt, ResultExt};
+use snafu::{OptionExt, ResultExt, ensure};
 
 use crate::common::{Identity, Password};
 use crate::error::{
@@ -30,7 +30,7 @@ use crate::error::{
     UserNotFoundSnafu, UserPasswordMismatchSnafu,
 };
 use crate::user_info::DefaultUserInfo;
-use crate::{auth_mysql, UserInfoRef};
+use crate::{UserInfoRef, auth_mysql};
 
 #[async_trait::async_trait]
 pub trait UserProvider: Send + Sync {

src/auth/src/user_provider/static_user_provider.rs

@@ -102,10 +102,10 @@ pub mod test {
 
     use common_test_util::temp_dir::create_temp_dir;
 
+    use crate::UserProvider;
     use crate::user_info::DefaultUserInfo;
     use crate::user_provider::static_user_provider::StaticUserProvider;
     use crate::user_provider::{Identity, Password};
-    use crate::UserProvider;
 
     async fn test_authenticate(provider: &dyn UserProvider, username: &str, password: &str) {
         let re = provider
@@ -143,12 +143,13 @@ pub mod test {
             let file = File::create(&file_path);
             let file = file.unwrap();
             let mut lw = LineWriter::new(file);
-            assert!(lw
-                .write_all(
+            assert!(
+                lw.write_all(
                     b"root=123456
 admin=654321",
                 )
-                .is_ok());
+                .is_ok()
+            );
             lw.flush().unwrap();
         }
 

src/auth/src/user_provider/watch_file_user_provider.rs

@@ -20,7 +20,7 @@ use std::sync::{Arc, Mutex};
 use async_trait::async_trait;
 use common_telemetry::{info, warn};
 use notify::{EventKind, RecursiveMode, Watcher};
-use snafu::{ensure, ResultExt};
+use snafu::{ResultExt, ensure};
 
 use crate::error::{FileWatchSnafu, InvalidConfigSnafu, Result};
 use crate::user_info::DefaultUserInfo;
@@ -133,9 +133,9 @@ pub mod test {
     use common_test_util::temp_dir::create_temp_dir;
     use tokio::time::sleep;
 
+    use crate::UserProvider;
     use crate::user_provider::watch_file_user_provider::WatchFileUserProvider;
     use crate::user_provider::{Identity, Password};
-    use crate::UserProvider;
 
     async fn test_authenticate(
         provider: &dyn UserProvider,

src/cache/src/lib.rs

@@ -19,9 +19,9 @@ use std::time::Duration;
 
 use catalog::kvbackend::new_table_cache;
 use common_meta::cache::{
-    new_schema_cache, new_table_flownode_set_cache, new_table_info_cache, new_table_name_cache,
-    new_table_route_cache, new_table_schema_cache, new_view_info_cache, CacheRegistry,
-    CacheRegistryBuilder, LayeredCacheRegistryBuilder,
+    CacheRegistry, CacheRegistryBuilder, LayeredCacheRegistryBuilder, new_schema_cache,
+    new_table_flownode_set_cache, new_table_info_cache, new_table_name_cache,
+    new_table_route_cache, new_table_schema_cache, new_view_info_cache,
 };
 use common_meta::kv_backend::KvBackendRef;
 use moka::future::CacheBuilder;

src/catalog/src/kvbackend.rs

@@ -21,4 +21,4 @@ mod table_cache;
 
 pub use builder::KvBackendCatalogManagerBuilder;
 pub use manager::KvBackendCatalogManager;
-pub use table_cache::{new_table_cache, TableCache, TableCacheRef};
+pub use table_cache::{TableCache, TableCacheRef, new_table_cache};

src/catalog/src/kvbackend/builder.rs

@@ -16,8 +16,8 @@ use std::sync::Arc;
 
 use common_catalog::consts::DEFAULT_CATALOG_NAME;
 use common_meta::cache::LayeredCacheRegistryRef;
-use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::TableMetadataManager;
+use common_meta::key::flow::FlowMetadataManager;
 use common_meta::kv_backend::KvBackendRef;
 use common_procedure::ProcedureManagerRef;
 use moka::sync::Cache;
@@ -26,8 +26,8 @@ use partition::manager::PartitionRuleManager;
 #[cfg(feature = "enterprise")]
 use crate::information_schema::InformationSchemaTableFactoryRef;
 use crate::information_schema::{InformationExtensionRef, InformationSchemaProvider};
-use crate::kvbackend::manager::{SystemCatalog, CATALOG_CACHE_MAX_CAPACITY};
 use crate::kvbackend::KvBackendCatalogManager;
+use crate::kvbackend::manager::{CATALOG_CACHE_MAX_CAPACITY, SystemCatalog};
 use crate::process_manager::ProcessManagerRef;
 use crate::system_schema::pg_catalog::PGCatalogProvider;
 

src/catalog/src/kvbackend/client.rs

@@ -24,12 +24,12 @@ use common_meta::error::Error::CacheNotGet;
 use common_meta::error::{CacheNotGetSnafu, Error, ExternalSnafu, GetKvCacheSnafu, Result};
 use common_meta::kv_backend::txn::{Txn, TxnResponse};
 use common_meta::kv_backend::{KvBackend, KvBackendRef, TxnService};
+use common_meta::rpc::KeyValue;
 use common_meta::rpc::store::{
     BatchDeleteRequest, BatchDeleteResponse, BatchGetRequest, BatchGetResponse, BatchPutRequest,
     BatchPutResponse, CompareAndPutRequest, CompareAndPutResponse, DeleteRangeRequest,
     DeleteRangeResponse, PutRequest, PutResponse, RangeRequest, RangeResponse,
 };
-use common_meta::rpc::KeyValue;
 use common_telemetry::debug;
 use meta_client::client::MetaClient;
 use moka::future::{Cache, CacheBuilder};
@@ -461,17 +461,17 @@ impl KvBackend for MetaKvBackend {
 #[cfg(test)]
 mod tests {
     use std::any::Any;
-    use std::sync::atomic::{AtomicU32, Ordering};
     use std::sync::Arc;
+    use std::sync::atomic::{AtomicU32, Ordering};
 
     use async_trait::async_trait;
     use common_meta::kv_backend::{KvBackend, TxnService};
+    use common_meta::rpc::KeyValue;
     use common_meta::rpc::store::{
         BatchDeleteRequest, BatchDeleteResponse, BatchGetRequest, BatchGetResponse,
         BatchPutRequest, BatchPutResponse, DeleteRangeRequest, DeleteRangeResponse, PutRequest,
         PutResponse, RangeRequest, RangeResponse,
     };
-    use common_meta::rpc::KeyValue;
     use dashmap::DashMap;
 
     use super::CachedKvBackend;

src/catalog/src/kvbackend/manager.rs

@@ -26,12 +26,12 @@ use common_meta::cache::{
     LayeredCacheRegistryRef, TableInfoCacheRef, TableNameCacheRef, TableRoute, TableRouteCacheRef,
     ViewInfoCacheRef,
 };
+use common_meta::key::TableMetadataManagerRef;
 use common_meta::key::catalog_name::CatalogNameKey;
 use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::schema_name::SchemaNameKey;
 use common_meta::key::table_info::{TableInfoManager, TableInfoValue};
 use common_meta::key::table_name::TableNameKey;
-use common_meta::key::TableMetadataManagerRef;
 use common_meta::kv_backend::KvBackendRef;
 use common_procedure::ProcedureManagerRef;
 use futures_util::stream::BoxStream;
@@ -41,15 +41,16 @@ use partition::manager::PartitionRuleManagerRef;
 use session::context::{Channel, QueryContext};
 use snafu::prelude::*;
 use store_api::metric_engine_consts::METRIC_ENGINE_NAME;
+use table::TableRef;
 use table::dist_table::DistTable;
 use table::metadata::{TableId, TableInfoRef};
-use table::table::numbers::{NumbersTable, NUMBERS_TABLE_NAME};
 use table::table::PartitionRules;
+use table::table::numbers::{NUMBERS_TABLE_NAME, NumbersTable};
 use table::table_name::TableName;
-use table::TableRef;
 use tokio::sync::Semaphore;
 use tokio_stream::wrappers::ReceiverStream;
 
+use crate::CatalogManager;
 use crate::error::{
     CacheNotFoundSnafu, GetTableCacheSnafu, InvalidTableInfoInCatalogSnafu, ListCatalogsSnafu,
     ListSchemasSnafu, ListTablesSnafu, Result, TableMetadataManagerSnafu,
@@ -59,9 +60,8 @@ use crate::information_schema::InformationSchemaTableFactoryRef;
 use crate::information_schema::{InformationExtensionRef, InformationSchemaProvider};
 use crate::kvbackend::TableCacheRef;
 use crate::process_manager::ProcessManagerRef;
-use crate::system_schema::pg_catalog::PGCatalogProvider;
 use crate::system_schema::SystemSchemaProvider;
-use crate::CatalogManager;
+use crate::system_schema::pg_catalog::PGCatalogProvider;
 
 /// Access all existing catalog, schema and tables.
 ///

src/catalog/src/kvbackend/table_cache.rs

@@ -20,9 +20,9 @@ use common_meta::instruction::CacheIdent;
 use futures::future::BoxFuture;
 use moka::future::Cache;
 use snafu::OptionExt;
+use table::TableRef;
 use table::dist_table::DistTable;
 use table::table_name::TableName;
-use table::TableRef;
 
 pub type TableCacheRef = Arc<TableCache>;
 

src/catalog/src/lib.rs

@@ -25,8 +25,8 @@ use common_catalog::consts::{INFORMATION_SCHEMA_NAME, PG_CATALOG_NAME};
 use futures::future::BoxFuture;
 use futures_util::stream::BoxStream;
 use session::context::QueryContext;
-use table::metadata::{TableId, TableInfoRef};
 use table::TableRef;
+use table::metadata::{TableId, TableInfoRef};
 
 use crate::error::Result;
 

src/catalog/src/memory.rs

@@ -14,4 +14,4 @@
 
 pub mod manager;
 
-pub use manager::{new_memory_catalog_manager, MemoryCatalogManager};
+pub use manager::{MemoryCatalogManager, new_memory_catalog_manager};

src/catalog/src/memory/manager.rs

@@ -28,8 +28,8 @@ use common_meta::kv_backend::memory::MemoryKvBackend;
 use futures_util::stream::BoxStream;
 use session::context::QueryContext;
 use snafu::OptionExt;
-use table::metadata::{TableId, TableInfoRef};
 use table::TableRef;
+use table::metadata::{TableId, TableInfoRef};
 
 use crate::error::{CatalogNotFoundSnafu, Result, SchemaNotFoundSnafu, TableExistsSnafu};
 use crate::information_schema::InformationSchemaProvider;
@@ -419,7 +419,7 @@ pub fn new_memory_catalog_manager() -> Result<Arc<MemoryCatalogManager>> {
 mod tests {
     use common_catalog::consts::*;
     use futures_util::TryStreamExt;
-    use table::table::numbers::{NumbersTable, NUMBERS_TABLE_NAME};
+    use table::table::numbers::{NUMBERS_TABLE_NAME, NumbersTable};
 
     use super::*;
 
@@ -454,7 +454,8 @@ mod tests {
             tables[0].table_info().table_id()
         );
 
-        assert!(catalog_list
+        assert!(
+            catalog_list
                 .table(
                     DEFAULT_CATALOG_NAME,
                     DEFAULT_SCHEMA_NAME,
@@ -463,7 +464,8 @@ mod tests {
                 )
                 .await
                 .unwrap()
-            .is_none());
+                .is_none()
+        );
     }
 
     #[test]
@@ -486,11 +488,13 @@ mod tests {
             table: NumbersTable::table(2333),
         };
         catalog.register_table_sync(register_table_req).unwrap();
-        assert!(catalog
+        assert!(
+            catalog
                 .table(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, table_name, None)
                 .await
                 .unwrap()
-            .is_some());
+                .is_some()
+        );
 
         let deregister_table_req = DeregisterTableRequest {
             catalog: DEFAULT_CATALOG_NAME.to_string(),
@@ -498,10 +502,12 @@ mod tests {
             table_name: table_name.to_string(),
         };
         catalog.deregister_table_sync(deregister_table_req).unwrap();
-        assert!(catalog
+        assert!(
+            catalog
                 .table(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, table_name, None)
                 .await
                 .unwrap()
-            .is_none());
+                .is_none()
+        );
     }
 }

src/catalog/src/process_manager.rs

@@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::hash_map::Entry;
 use std::collections::HashMap;
+use std::collections::hash_map::Entry;
 use std::fmt::{Debug, Display, Formatter};
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::{Arc, RwLock};
@@ -30,7 +30,7 @@ use common_time::util::current_time_millis;
 use meta_client::MetaClientRef;
 use promql_parser::parser::EvalStmt;
 use rand::random;
-use snafu::{ensure, OptionExt, ResultExt};
+use snafu::{OptionExt, ResultExt, ensure};
 use sql::statements::statement::Statement;
 
 use crate::error;

src/catalog/src/system_schema/information_schema.rs

@@ -36,8 +36,8 @@ use common_catalog::consts::{self, DEFAULT_CATALOG_NAME, INFORMATION_SCHEMA_NAME
 use common_error::ext::ErrorExt;
 use common_meta::cluster::NodeInfo;
 use common_meta::datanode::RegionStat;
-use common_meta::key::flow::flow_state::FlowStat;
 use common_meta::key::flow::FlowMetadataManager;
+use common_meta::key::flow::flow_state::FlowStat;
 use common_meta::kv_backend::KvBackendRef;
 use common_procedure::ProcedureInfo;
 use common_recordbatch::SendableRecordBatchStream;
@@ -46,12 +46,13 @@ use lazy_static::lazy_static;
 use paste::paste;
 use process_list::InformationSchemaProcessList;
 use store_api::storage::{ScanRequest, TableId};
-use table::metadata::TableType;
 use table::TableRef;
+use table::metadata::TableType;
 pub use table_names::*;
 use views::InformationSchemaViews;
 
 use self::columns::InformationSchemaColumns;
+use crate::CatalogManager;
 use crate::error::{Error, Result};
 use crate::process_manager::ProcessManagerRef;
 use crate::system_schema::information_schema::cluster_info::InformationSchemaClusterInfo;
@@ -69,7 +70,6 @@ pub(crate) use crate::system_schema::predicate::Predicates;
 use crate::system_schema::{
     SystemSchemaProvider, SystemSchemaProviderInner, SystemTable, SystemTableRef,
 };
-use crate::CatalogManager;
 
 lazy_static! {
     // Memory tables in `information_schema`.

src/catalog/src/system_schema/information_schema/cluster_info.rs

@@ -23,9 +23,9 @@ use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
 use common_time::timestamp::Timestamp;
 use datafusion::execution::TaskContext;
+use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
 use datatypes::timestamp::TimestampMillisecond;
@@ -36,10 +36,10 @@ use datatypes::vectors::{
 use snafu::ResultExt;
 use store_api::storage::{ScanRequest, TableId};
 
-use crate::error::{CreateRecordBatchSnafu, InternalSnafu, Result};
-use crate::system_schema::information_schema::{InformationTable, Predicates, CLUSTER_INFO};
-use crate::system_schema::utils;
 use crate::CatalogManager;
+use crate::error::{CreateRecordBatchSnafu, InternalSnafu, Result};
+use crate::system_schema::information_schema::{CLUSTER_INFO, InformationTable, Predicates};
+use crate::system_schema::utils;
 
 const PEER_ID: &str = "peer_id";
 const PEER_TYPE: &str = "peer_type";

src/catalog/src/system_schema/information_schema/columns.rs

@@ -23,9 +23,9 @@ use common_error::ext::BoxedError;
 use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
 use datafusion::execution::TaskContext;
+use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datatypes::prelude::{ConcreteDataType, DataType, MutableVector};
 use datatypes::scalars::ScalarVectorBuilder;
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
@@ -38,12 +38,12 @@ use snafu::{OptionExt, ResultExt};
 use sql::statements;
 use store_api::storage::{ScanRequest, TableId};
 
+use crate::CatalogManager;
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::information_schema::Predicates;
-use crate::system_schema::information_schema::{InformationTable, COLUMNS};
-use crate::CatalogManager;
+use crate::system_schema::information_schema::{COLUMNS, InformationTable};
 
 #[derive(Debug)]
 pub(super) struct InformationSchemaColumns {

src/catalog/src/system_schema/information_schema/flows.rs

@@ -16,10 +16,10 @@ use std::sync::{Arc, Weak};
 
 use common_catalog::consts::INFORMATION_SCHEMA_FLOW_TABLE_ID;
 use common_error::ext::BoxedError;
+use common_meta::key::FlowId;
+use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::flow::flow_info::FlowInfoValue;
 use common_meta::key::flow::flow_state::FlowStat;
-use common_meta::key::flow::FlowMetadataManager;
-use common_meta::key::FlowId;
 use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{DfSendableRecordBatchStream, RecordBatch, SendableRecordBatchStream};
 use datafusion::execution::TaskContext;
@@ -38,14 +38,14 @@ use futures::TryStreamExt;
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::{ScanRequest, TableId};
 
+use crate::CatalogManager;
 use crate::error::{
     CreateRecordBatchSnafu, FlowInfoNotFoundSnafu, InternalSnafu, JsonSnafu, ListFlowsSnafu,
     Result, UpgradeWeakCatalogManagerRefSnafu,
 };
-use crate::information_schema::{Predicates, FLOWS};
+use crate::information_schema::{FLOWS, Predicates};
 use crate::system_schema::information_schema::InformationTable;
 use crate::system_schema::utils;
-use crate::CatalogManager;
 
 const INIT_CAPACITY: usize = 42;
 

src/catalog/src/system_schema/information_schema/information_memory_table.rs

@@ -89,9 +89,9 @@ pub(super) fn get_schema_columns(table_name: &str) -> (SchemaRef, Vec<VectorRef>
                 vec![
                     Arc::new(StringVector::from(vec![build_info.branch.to_string()])),
                     Arc::new(StringVector::from(vec![build_info.commit.to_string()])),
-                    Arc::new(StringVector::from(vec![build_info
-                        .commit_short
-                        .to_string()])),
+                    Arc::new(StringVector::from(vec![
+                        build_info.commit_short.to_string(),
+                    ])),
                     Arc::new(StringVector::from(vec![build_info.clean.to_string()])),
                     Arc::new(StringVector::from(vec![build_info.version.to_string()])),
                 ],
@@ -369,17 +369,9 @@ pub(super) fn get_schema_columns(table_name: &str) -> (SchemaRef, Vec<VectorRef>
         TRIGGERS => (
             vec![
                 string_column("TRIGGER_NAME"),
-                ColumnSchema::new(
-                    "trigger_id",
-                    ConcreteDataType::uint64_datatype(),
-                    false,
-                ),
+                ColumnSchema::new("trigger_id", ConcreteDataType::uint64_datatype(), false),
                 string_column("TRIGGER_DEFINITION"),
-                ColumnSchema::new(
-                    "flownode_id",
-                    ConcreteDataType::uint64_datatype(),
-                    true,
-                ),
+                ColumnSchema::new("flownode_id", ConcreteDataType::uint64_datatype(), true),
             ],
             vec![],
         ),

src/catalog/src/system_schema/information_schema/key_column_usage.rs

@@ -20,9 +20,9 @@ use common_error::ext::BoxedError;
 use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
 use datafusion::execution::TaskContext;
+use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datatypes::prelude::{ConcreteDataType, MutableVector, ScalarVectorBuilder, VectorRef};
 use datatypes::schema::{ColumnSchema, FulltextBackend, Schema, SchemaRef};
 use datatypes::value::Value;
@@ -31,11 +31,11 @@ use futures_util::TryStreamExt;
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::{ScanRequest, TableId};
 
+use crate::CatalogManager;
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
-use crate::system_schema::information_schema::{InformationTable, Predicates, KEY_COLUMN_USAGE};
-use crate::CatalogManager;
+use crate::system_schema::information_schema::{InformationTable, KEY_COLUMN_USAGE, Predicates};
 
 pub const CONSTRAINT_SCHEMA: &str = "constraint_schema";
 pub const CONSTRAINT_NAME: &str = "constraint_name";
@@ -277,8 +277,9 @@ impl InformationSchemaKeyColumnUsageBuilder {
                         constraints.push(CONSTRAINT_NAME_INVERTED_INDEX);
                         greptime_index_type.push(INDEX_TYPE_INVERTED_INDEX);
                     }
-                    if let Ok(Some(options)) = column.fulltext_options() {
-                        if options.enable {
+                    if let Ok(Some(options)) = column.fulltext_options()
+                        && options.enable
+                    {
                         constraints.push(CONSTRAINT_NAME_FULLTEXT_INDEX);
                         let index_type = match options.backend {
                             FulltextBackend::Bloom => INDEX_TYPE_FULLTEXT_BLOOM,
@@ -286,7 +287,6 @@ impl InformationSchemaKeyColumnUsageBuilder {
                         };
                         greptime_index_type.push(index_type);
                     }
-                    }
                     if column.is_skipping_indexed() {
                         constraints.push(CONSTRAINT_NAME_SKIPPING_INDEX);
                         greptime_index_type.push(INDEX_TYPE_SKIPPING_INDEX);

src/catalog/src/system_schema/information_schema/partitions.rs

@@ -21,9 +21,9 @@ use common_error::ext::BoxedError;
 use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
 use datafusion::execution::TaskContext;
+use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
 use datatypes::timestamp::TimestampMicrosecond;
@@ -39,13 +39,13 @@ use snafu::{OptionExt, ResultExt};
 use store_api::storage::{ScanRequest, TableId};
 use table::metadata::{TableInfo, TableType};
 
+use crate::CatalogManager;
 use crate::error::{
     CreateRecordBatchSnafu, FindPartitionsSnafu, InternalSnafu, PartitionManagerNotFoundSnafu,
     Result, UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::kvbackend::KvBackendCatalogManager;
-use crate::system_schema::information_schema::{InformationTable, Predicates, PARTITIONS};
-use crate::CatalogManager;
+use crate::system_schema::information_schema::{InformationTable, PARTITIONS, Predicates};
 
 const TABLE_CATALOG: &str = "table_catalog";
 const TABLE_SCHEMA: &str = "table_schema";

src/catalog/src/system_schema/information_schema/procedure_info.rs

@@ -22,9 +22,9 @@ use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
 use common_time::timestamp::Timestamp;
 use datafusion::execution::TaskContext;
+use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
 use datatypes::timestamp::TimestampMillisecond;
@@ -33,10 +33,10 @@ use datatypes::vectors::{StringVectorBuilder, TimestampMillisecondVectorBuilder}
 use snafu::ResultExt;
 use store_api::storage::{ScanRequest, TableId};
 
-use crate::error::{CreateRecordBatchSnafu, InternalSnafu, Result};
-use crate::system_schema::information_schema::{InformationTable, Predicates, PROCEDURE_INFO};
-use crate::system_schema::utils;
 use crate::CatalogManager;
+use crate::error::{CreateRecordBatchSnafu, InternalSnafu, Result};
+use crate::system_schema::information_schema::{InformationTable, PROCEDURE_INFO, Predicates};
+use crate::system_schema::utils;
 
 const PROCEDURE_ID: &str = "procedure_id";
 const PROCEDURE_TYPE: &str = "procedure_type";

src/catalog/src/system_schema/information_schema/region_peers.rs

@@ -23,9 +23,9 @@ use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
 use datafusion::common::HashMap;
 use datafusion::execution::TaskContext;
+use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
 use datatypes::value::Value;
@@ -35,13 +35,13 @@ use snafu::{OptionExt, ResultExt};
 use store_api::storage::{RegionId, ScanRequest, TableId};
 use table::metadata::TableType;
 
+use crate::CatalogManager;
 use crate::error::{
     CreateRecordBatchSnafu, FindRegionRoutesSnafu, InternalSnafu, Result,
     UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::kvbackend::KvBackendCatalogManager;
 use crate::system_schema::information_schema::{InformationTable, Predicates, REGION_PEERS};
-use crate::CatalogManager;
 
 pub const TABLE_CATALOG: &str = "table_catalog";
 pub const TABLE_SCHEMA: &str = "table_schema";

src/catalog/src/system_schema/information_schema/region_statistics.rs

@@ -30,11 +30,11 @@ use datatypes::vectors::{StringVectorBuilder, UInt32VectorBuilder, UInt64VectorB
 use snafu::ResultExt;
 use store_api::storage::{ScanRequest, TableId};
 
+use crate::CatalogManager;
 use crate::error::{CreateRecordBatchSnafu, InternalSnafu, Result};
 use crate::information_schema::Predicates;
 use crate::system_schema::information_schema::{InformationTable, REGION_STATISTICS};
 use crate::system_schema::utils;
-use crate::CatalogManager;
 
 const REGION_ID: &str = "region_id";
 const TABLE_ID: &str = "table_id";

src/catalog/src/system_schema/information_schema/runtime_metrics.rs

@@ -21,9 +21,9 @@ use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
 use common_time::util::current_time_millis;
 use datafusion::execution::TaskContext;
+use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datatypes::prelude::{ConcreteDataType, MutableVector};
 use datatypes::scalars::ScalarVectorBuilder;
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};

src/catalog/src/system_schema/information_schema/schemata.rs

@@ -21,9 +21,9 @@ use common_meta::key::schema_name::SchemaNameKey;
 use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
 use datafusion::execution::TaskContext;
+use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
 use datatypes::value::Value;
@@ -31,13 +31,13 @@ use datatypes::vectors::StringVectorBuilder;
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::{ScanRequest, TableId};
 
+use crate::CatalogManager;
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, TableMetadataManagerSnafu,
     UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::system_schema::information_schema::{InformationTable, Predicates, SCHEMATA};
 use crate::system_schema::utils;
-use crate::CatalogManager;
 
 pub const CATALOG_NAME: &str = "catalog_name";
 pub const SCHEMA_NAME: &str = "schema_name";

src/catalog/src/system_schema/information_schema/table_constraints.rs

@@ -20,9 +20,9 @@ use common_error::ext::BoxedError;
 use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
 use datafusion::execution::TaskContext;
+use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datatypes::prelude::{ConcreteDataType, MutableVector};
 use datatypes::scalars::ScalarVectorBuilder;
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
@@ -32,15 +32,15 @@ use futures::TryStreamExt;
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::{ScanRequest, TableId};
 
+use crate::CatalogManager;
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
+use crate::information_schema::Predicates;
 use crate::information_schema::key_column_usage::{
     CONSTRAINT_NAME_PRI, CONSTRAINT_NAME_TIME_INDEX,
 };
-use crate::information_schema::Predicates;
 use crate::system_schema::information_schema::{InformationTable, TABLE_CONSTRAINTS};
-use crate::CatalogManager;
 
 /// The `TABLE_CONSTRAINTS` table describes which tables have constraints.
 #[derive(Debug)]

src/catalog/src/system_schema/information_schema/tables.rs

@@ -23,9 +23,9 @@ use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
 use common_telemetry::error;
 use datafusion::execution::TaskContext;
+use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
 use datatypes::value::Value;
@@ -37,12 +37,12 @@ use snafu::{OptionExt, ResultExt};
 use store_api::storage::{RegionId, ScanRequest, TableId};
 use table::metadata::{TableInfo, TableType};
 
+use crate::CatalogManager;
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::system_schema::information_schema::{InformationTable, Predicates, TABLES};
 use crate::system_schema::utils;
-use crate::CatalogManager;
 
 pub const TABLE_CATALOG: &str = "table_catalog";
 pub const TABLE_SCHEMA: &str = "table_schema";

src/catalog/src/system_schema/information_schema/views.rs

@@ -20,9 +20,9 @@ use common_error::ext::BoxedError;
 use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
 use datafusion::execution::TaskContext;
+use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
 use datatypes::value::Value;
@@ -32,13 +32,13 @@ use snafu::{OptionExt, ResultExt};
 use store_api::storage::{ScanRequest, TableId};
 use table::metadata::TableType;
 
+use crate::CatalogManager;
 use crate::error::{
     CastManagerSnafu, CreateRecordBatchSnafu, GetViewCacheSnafu, InternalSnafu, Result,
     UpgradeWeakCatalogManagerRefSnafu, ViewInfoNotFoundSnafu,
 };
 use crate::kvbackend::KvBackendCatalogManager;
 use crate::system_schema::information_schema::{InformationTable, Predicates, VIEWS};
-use crate::CatalogManager;
 const INIT_CAPACITY: usize = 42;
 
 pub const TABLE_CATALOG: &str = "table_catalog";

src/catalog/src/system_schema/memory_table.rs

@@ -21,9 +21,9 @@ use common_error::ext::BoxedError;
 use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
 use datafusion::execution::TaskContext;
+use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
-use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datatypes::schema::SchemaRef;
 use datatypes::vectors::VectorRef;
 use snafu::ResultExt;

src/catalog/src/system_schema/pg_catalog.rs

@@ -34,10 +34,10 @@ use table::TableRef;
 pub use table_names::*;
 
 use self::pg_namespace::oid_map::{PGNamespaceOidMap, PGNamespaceOidMapRef};
+use crate::CatalogManager;
 use crate::system_schema::memory_table::MemoryTable;
 use crate::system_schema::utils::tables::u32_column;
 use crate::system_schema::{SystemSchemaProvider, SystemSchemaProviderInner, SystemTableRef};
-use crate::CatalogManager;
 
 lazy_static! {
     static ref MEMORY_TABLES: &'static [&'static str] = &[table_names::PG_TYPE];

src/catalog/src/system_schema/pg_catalog/pg_class.rs

@@ -32,15 +32,15 @@ use snafu::{OptionExt, ResultExt};
 use store_api::storage::ScanRequest;
 use table::metadata::TableType;
 
+use crate::CatalogManager;
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::information_schema::Predicates;
-use crate::system_schema::pg_catalog::pg_namespace::oid_map::PGNamespaceOidMapRef;
-use crate::system_schema::pg_catalog::{query_ctx, OID_COLUMN_NAME, PG_CLASS};
-use crate::system_schema::utils::tables::{string_column, u32_column};
 use crate::system_schema::SystemTable;
-use crate::CatalogManager;
+use crate::system_schema::pg_catalog::pg_namespace::oid_map::PGNamespaceOidMapRef;
+use crate::system_schema::pg_catalog::{OID_COLUMN_NAME, PG_CLASS, query_ctx};
+use crate::system_schema::utils::tables::{string_column, u32_column};
 
 // === column name ===
 pub const RELNAME: &str = "relname";

src/catalog/src/system_schema/pg_catalog/pg_database.rs

@@ -29,15 +29,15 @@ use datatypes::vectors::{StringVectorBuilder, UInt32VectorBuilder, VectorRef};
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::ScanRequest;
 
+use crate::CatalogManager;
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::information_schema::Predicates;
-use crate::system_schema::pg_catalog::pg_namespace::oid_map::PGNamespaceOidMapRef;
-use crate::system_schema::pg_catalog::{query_ctx, OID_COLUMN_NAME, PG_DATABASE};
-use crate::system_schema::utils::tables::{string_column, u32_column};
 use crate::system_schema::SystemTable;
-use crate::CatalogManager;
+use crate::system_schema::pg_catalog::pg_namespace::oid_map::PGNamespaceOidMapRef;
+use crate::system_schema::pg_catalog::{OID_COLUMN_NAME, PG_DATABASE, query_ctx};
+use crate::system_schema::utils::tables::{string_column, u32_column};
 
 // === column name ===
 pub const DATNAME: &str = "datname";

src/catalog/src/system_schema/pg_catalog/pg_namespace.rs

@@ -35,16 +35,16 @@ use datatypes::vectors::{StringVectorBuilder, UInt32VectorBuilder, VectorRef};
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::ScanRequest;
 
+use crate::CatalogManager;
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::information_schema::Predicates;
+use crate::system_schema::SystemTable;
 use crate::system_schema::pg_catalog::{
-    query_ctx, PGNamespaceOidMapRef, OID_COLUMN_NAME, PG_NAMESPACE,
+    OID_COLUMN_NAME, PG_NAMESPACE, PGNamespaceOidMapRef, query_ctx,
 };
 use crate::system_schema::utils::tables::{string_column, u32_column};
-use crate::system_schema::SystemTable;
-use crate::CatalogManager;
 
 const NSPNAME: &str = "nspname";
 const INIT_CAPACITY: usize = 42;

src/catalog/src/system_schema/predicate.rs

@@ -339,18 +339,22 @@ mod tests {
         assert!(!p.eval(&wrong_row).unwrap());
         assert!(p.eval(&[]).is_none());
         assert!(p.eval(&[("c", &a_value)]).is_none());
-        assert!(!p
-            .eval(&[(a_col.as_str(), &b_value), (b_col.as_str(), &a_value)])
-            .unwrap());
-        assert!(!p
-            .eval(&[(a_col.as_str(), &b_value), (b_col.as_str(), &b_value)])
-            .unwrap());
-        assert!(p
-            .eval(&[(a_col.as_ref(), &a_value), ("c", &a_value)])
-            .is_none());
-        assert!(!p
-            .eval(&[(a_col.as_ref(), &b_value), ("c", &a_value)])
-            .unwrap());
+        assert!(
+            !p.eval(&[(a_col.as_str(), &b_value), (b_col.as_str(), &a_value)])
+                .unwrap()
+        );
+        assert!(
+            !p.eval(&[(a_col.as_str(), &b_value), (b_col.as_str(), &b_value)])
+                .unwrap()
+        );
+        assert!(
+            p.eval(&[(a_col.as_ref(), &a_value), ("c", &a_value)])
+                .is_none()
+        );
+        assert!(
+            !p.eval(&[(a_col.as_ref(), &b_value), ("c", &a_value)])
+                .unwrap()
+        );
 
         //Predicate::Or
         let p = Predicate::Or(Box::new(p1), Box::new(p2));
@@ -358,18 +362,22 @@ mod tests {
         assert!(p.eval(&wrong_row).unwrap());
         assert!(p.eval(&[]).is_none());
         assert!(p.eval(&[("c", &a_value)]).is_none());
-        assert!(!p
-            .eval(&[(a_col.as_str(), &b_value), (b_col.as_str(), &a_value)])
-            .unwrap());
-        assert!(p
-            .eval(&[(a_col.as_str(), &b_value), (b_col.as_str(), &b_value)])
-            .unwrap());
-        assert!(p
-            .eval(&[(a_col.as_ref(), &a_value), ("c", &a_value)])
-            .unwrap());
-        assert!(p
-            .eval(&[(a_col.as_ref(), &b_value), ("c", &a_value)])
-            .is_none());
+        assert!(
+            !p.eval(&[(a_col.as_str(), &b_value), (b_col.as_str(), &a_value)])
+                .unwrap()
+        );
+        assert!(
+            p.eval(&[(a_col.as_str(), &b_value), (b_col.as_str(), &b_value)])
+                .unwrap()
+        );
+        assert!(
+            p.eval(&[(a_col.as_ref(), &a_value), ("c", &a_value)])
+                .unwrap()
+        );
+        assert!(
+            p.eval(&[(a_col.as_ref(), &b_value), ("c", &a_value)])
+                .is_none()
+        );
     }
 
     #[test]

src/catalog/src/system_schema/utils.rs

@@ -17,10 +17,10 @@ use std::sync::Weak;
 use common_meta::key::TableMetadataManagerRef;
 use snafu::OptionExt;
 
+use crate::CatalogManager;
 use crate::error::{GetInformationExtensionSnafu, Result, UpgradeWeakCatalogManagerRefSnafu};
 use crate::information_schema::InformationExtensionRef;
 use crate::kvbackend::KvBackendCatalogManager;
-use crate::CatalogManager;
 
 pub mod tables;
 

src/catalog/src/table_source.rs

@@ -17,27 +17,27 @@ use std::sync::Arc;
 
 use bytes::Bytes;
 use common_catalog::format_full_table_name;
-use common_query::logical_plan::{rename_logical_plan_columns, SubstraitPlanDecoderRef};
+use common_query::logical_plan::{SubstraitPlanDecoderRef, rename_logical_plan_columns};
 use datafusion::common::{ResolvedTableReference, TableReference};
 use datafusion::datasource::view::ViewTable;
-use datafusion::datasource::{provider_as_source, TableProvider};
+use datafusion::datasource::{TableProvider, provider_as_source};
 use datafusion::logical_expr::TableSource;
 use itertools::Itertools;
 use session::context::QueryContextRef;
-use snafu::{ensure, OptionExt, ResultExt};
+use snafu::{OptionExt, ResultExt, ensure};
 use table::metadata::TableType;
 use table::table::adapter::DfTableProviderAdapter;
 pub mod dummy_catalog;
 use dummy_catalog::DummyCatalogList;
 use table::TableRef;
 
+use crate::CatalogManagerRef;
 use crate::error::{
     CastManagerSnafu, DecodePlanSnafu, GetViewCacheSnafu, ProjectViewColumnsSnafu,
     QueryAccessDeniedSnafu, Result, TableNotExistSnafu, ViewInfoNotFoundSnafu,
     ViewPlanColumnsChangedSnafu,
 };
 use crate::kvbackend::KvBackendCatalogManager;
-use crate::CatalogManagerRef;
 
 pub struct DfTableSourceProvider {
     catalog_manager: CatalogManagerRef,
@@ -272,7 +272,7 @@ mod tests {
     use common_query::logical_plan::SubstraitPlanDecoder;
     use datafusion::catalog::CatalogProviderList;
     use datafusion::logical_expr::builder::LogicalTableSource;
-    use datafusion::logical_expr::{col, lit, LogicalPlan, LogicalPlanBuilder};
+    use datafusion::logical_expr::{LogicalPlan, LogicalPlanBuilder, col, lit};
 
     use crate::information_schema::NoopInformationExtension;
 

src/catalog/src/table_source/dummy_catalog.rs

@@ -25,8 +25,8 @@ use datafusion::datasource::TableProvider;
 use snafu::OptionExt;
 use table::table::adapter::DfTableProviderAdapter;
 
-use crate::error::TableNotExistSnafu;
 use crate::CatalogManagerRef;
+use crate::error::TableNotExistSnafu;
 
 /// Delegate the resolving requests to the `[CatalogManager]` unconditionally.
 #[derive(Clone)]

src/cli/src/bench/metadata.rs

@@ -14,8 +14,8 @@
 
 use std::time::Instant;
 
-use common_meta::key::table_route::TableRouteValue;
 use common_meta::key::TableMetadataManagerRef;
+use common_meta::key::table_route::TableRouteValue;
 use table::table_name::TableName;
 
 use crate::bench::{

src/cli/src/data.rs

@@ -18,9 +18,9 @@ mod import;
 use clap::Subcommand;
 use common_error::ext::BoxedError;
 
+use crate::Tool;
 use crate::data::export::ExportCommand;
 use crate::data::import::ImportCommand;
-use crate::Tool;
 
 /// Command for data operations including exporting data from and importing data into GreptimeDB.
 #[derive(Subcommand)]

src/cli/src/data/export.rs

@@ -24,18 +24,18 @@ use common_error::ext::BoxedError;
 use common_telemetry::{debug, error, info};
 use object_store::layers::LoggingLayer;
 use object_store::services::Oss;
-use object_store::{services, ObjectStore};
+use object_store::{ObjectStore, services};
 use serde_json::Value;
 use snafu::{OptionExt, ResultExt};
 use tokio::sync::Semaphore;
 use tokio::time::Instant;
 
-use crate::database::{parse_proxy_opts, DatabaseClient};
+use crate::database::{DatabaseClient, parse_proxy_opts};
 use crate::error::{
     EmptyResultSnafu, Error, OpenDalSnafu, OutputDirNotSetSnafu, Result, S3ConfigNotSetSnafu,
     SchemaNotFoundSnafu,
 };
-use crate::{database, Tool};
+use crate::{Tool, database};
 
 type TableReference = (String, String, String);
 

src/cli/src/data/import.rs

@@ -25,9 +25,9 @@ use snafu::{OptionExt, ResultExt};
 use tokio::sync::Semaphore;
 use tokio::time::Instant;
 
-use crate::database::{parse_proxy_opts, DatabaseClient};
+use crate::database::{DatabaseClient, parse_proxy_opts};
 use crate::error::{Error, FileIoSnafu, Result, SchemaNotFoundSnafu};
-use crate::{database, Tool};
+use crate::{Tool, database};
 
 #[derive(Debug, Default, Clone, ValueEnum)]
 enum ImportTarget {

src/cli/src/database.rs

@@ -14,15 +14,15 @@
 
 use std::time::Duration;
 
-use base64::engine::general_purpose;
 use base64::Engine;
+use base64::engine::general_purpose;
 use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_error::ext::BoxedError;
 use humantime::format_duration;
 use serde_json::Value;
+use servers::http::GreptimeQueryOutput;
 use servers::http::header::constants::GREPTIME_DB_HEADER_TIMEOUT;
 use servers::http::result::greptime_result_v1::GreptimedbV1Response;
-use servers::http::GreptimeQueryOutput;
 use snafu::ResultExt;
 
 use crate::error::{

src/cli/src/metadata.rs

@@ -21,10 +21,10 @@ mod utils;
 use clap::Subcommand;
 use common_error::ext::BoxedError;
 
+use crate::Tool;
 use crate::metadata::control::{DelCommand, GetCommand};
 use crate::metadata::repair::RepairLogicalTablesCommand;
 use crate::metadata::snapshot::SnapshotCommand;
-use crate::Tool;
 
 /// Command for managing metadata operations,
 /// including saving and restoring metadata snapshots,

src/cli/src/metadata/common.rs

@@ -16,9 +16,9 @@ use std::sync::Arc;
 
 use clap::Parser;
 use common_error::ext::BoxedError;
+use common_meta::kv_backend::KvBackendRef;
 use common_meta::kv_backend::chroot::ChrootKvBackend;
 use common_meta::kv_backend::etcd::EtcdStore;
-use common_meta::kv_backend::KvBackendRef;
 use meta_srv::bootstrap::create_etcd_client_with_tls;
 use meta_srv::metasrv::BackendImpl;
 use servers::tls::{TlsMode, TlsOption};

src/cli/src/metadata/control/del.rs

@@ -18,9 +18,9 @@ mod table;
 use clap::Subcommand;
 use common_error::ext::BoxedError;
 
+use crate::Tool;
 use crate::metadata::control::del::key::DelKeyCommand;
 use crate::metadata::control::del::table::DelTableCommand;
-use crate::Tool;
 
 /// The prefix of the tombstone keys.
 pub(crate) const CLI_TOMBSTONE_PREFIX: &str = "__cli_tombstone/";

src/cli/src/metadata/control/del/key.rs

@@ -19,9 +19,9 @@ use common_meta::key::tombstone::TombstoneManager;
 use common_meta::kv_backend::KvBackendRef;
 use common_meta::rpc::store::RangeRequest;
 
+use crate::Tool;
 use crate::metadata::common::StoreConfig;
 use crate::metadata::control::del::CLI_TOMBSTONE_PREFIX;
-use crate::Tool;
 
 /// Delete key-value pairs logically from the metadata store.
 #[derive(Debug, Default, Parser)]
@@ -102,8 +102,8 @@ mod tests {
     use common_meta::kv_backend::{KvBackend, KvBackendRef};
     use common_meta::rpc::store::RangeRequest;
 
-    use crate::metadata::control::del::key::KeyDeleter;
     use crate::metadata::control::del::CLI_TOMBSTONE_PREFIX;
+    use crate::metadata::control::del::key::KeyDeleter;
     use crate::metadata::control::test_utils::put_key;
 
     #[tokio::test]

src/cli/src/metadata/control/del/table.rs

@@ -18,16 +18,16 @@ use client::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_catalog::format_full_table_name;
 use common_error::ext::BoxedError;
 use common_meta::ddl::utils::get_region_wal_options;
-use common_meta::key::table_name::TableNameManager;
 use common_meta::key::TableMetadataManager;
+use common_meta::key::table_name::TableNameManager;
 use common_meta::kv_backend::KvBackendRef;
 use store_api::storage::TableId;
 
+use crate::Tool;
 use crate::error::{InvalidArgumentsSnafu, TableNotFoundSnafu};
 use crate::metadata::common::StoreConfig;
 use crate::metadata::control::del::CLI_TOMBSTONE_PREFIX;
 use crate::metadata::control::utils::get_table_id_by_name;
-use crate::Tool;
 
 /// Delete table metadata logically from the metadata store.
 #[derive(Debug, Default, Parser)]
@@ -183,15 +183,15 @@ mod tests {
 
     use common_error::ext::ErrorExt;
     use common_error::status_code::StatusCode;
-    use common_meta::key::table_route::TableRouteValue;
     use common_meta::key::TableMetadataManager;
+    use common_meta::key::table_route::TableRouteValue;
     use common_meta::kv_backend::chroot::ChrootKvBackend;
     use common_meta::kv_backend::memory::MemoryKvBackend;
     use common_meta::kv_backend::{KvBackend, KvBackendRef};
     use common_meta::rpc::store::RangeRequest;
 
-    use crate::metadata::control::del::table::TableMetadataDeleter;
     use crate::metadata::control::del::CLI_TOMBSTONE_PREFIX;
+    use crate::metadata::control::del::table::TableMetadataDeleter;
     use crate::metadata::control::test_utils::prepare_physical_table_metadata;
 
     #[tokio::test]

src/cli/src/metadata/control/get.rs

@@ -19,18 +19,18 @@ use clap::{Parser, Subcommand};
 use client::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_catalog::format_full_table_name;
 use common_error::ext::BoxedError;
+use common_meta::key::TableMetadataManager;
 use common_meta::key::table_info::TableInfoKey;
 use common_meta::key::table_route::TableRouteKey;
-use common_meta::key::TableMetadataManager;
 use common_meta::kv_backend::KvBackendRef;
-use common_meta::range_stream::{PaginationStream, DEFAULT_PAGE_SIZE};
+use common_meta::range_stream::{DEFAULT_PAGE_SIZE, PaginationStream};
 use common_meta::rpc::store::RangeRequest;
 use futures::TryStreamExt;
 
+use crate::Tool;
 use crate::error::InvalidArgumentsSnafu;
 use crate::metadata::common::StoreConfig;
 use crate::metadata::control::utils::{decode_key_value, get_table_id_by_name, json_fromatter};
-use crate::Tool;
 
 /// Getting metadata from metadata store.
 #[derive(Subcommand)]

src/cli/src/metadata/repair.rs

@@ -31,18 +31,18 @@ use common_meta::key::TableMetadataManager;
 use common_meta::kv_backend::KvBackendRef;
 use common_meta::node_manager::NodeManagerRef;
 use common_meta::peer::Peer;
-use common_meta::rpc::router::{find_leaders, RegionRoute};
+use common_meta::rpc::router::{RegionRoute, find_leaders};
 use common_telemetry::{error, info, warn};
 use futures::TryStreamExt;
-use snafu::{ensure, ResultExt};
+use snafu::{ResultExt, ensure};
 use store_api::storage::TableId;
 
+use crate::Tool;
 use crate::error::{
     InvalidArgumentsSnafu, Result, SendRequestToDatanodeSnafu, TableMetadataSnafu, UnexpectedSnafu,
 };
 use crate::metadata::common::StoreConfig;
 use crate::metadata::utils::{FullTableMetadata, IteratorInput, TableMetadataIterator};
-use crate::Tool;
 
 /// Repair metadata of logical tables.
 #[derive(Debug, Default, Parser)]
@@ -301,7 +301,10 @@ impl RepairTool {
         warn!(
             "Sending alter table requests to datanodes for table: {} failed for the datanodes: {:?}",
             full_table_name,
-            failed_peers.iter().map(|(peer, _)| peer.id).collect::<Vec<_>>()
+            failed_peers
+                .iter()
+                .map(|(peer, _)| peer.id)
+                .collect::<Vec<_>>()
         );
 
         let create_table_expr =
@@ -320,8 +323,7 @@ impl RepairTool {
             }
             info!(
                 "Region not found for table: {}, datanode: {}, trying to create the logical table on that datanode",
-                full_table_name,
-                peer.id
+                full_table_name, peer.id
             );
 
             // If the alter table request fails for any datanode, we attempt to create the table on that datanode

src/cli/src/metadata/repair/alter_table.rs

@@ -13,11 +13,11 @@
 // limitations under the License.
 
 use client::api::v1::alter_table_expr::Kind;
-use client::api::v1::region::{region_request, AlterRequests, RegionRequest, RegionRequestHeader};
+use client::api::v1::region::{AlterRequests, RegionRequest, RegionRequestHeader, region_request};
 use client::api::v1::{AddColumn, AddColumns, AlterTableExpr};
 use common_meta::ddl::alter_logical_tables::make_alter_region_request;
 use common_meta::peer::Peer;
-use common_meta::rpc::router::{find_leader_regions, RegionRoute};
+use common_meta::rpc::router::{RegionRoute, find_leader_regions};
 use operator::expr_helper::column_schemas_to_defs;
 use snafu::ResultExt;
 use store_api::storage::{RegionId, TableId};

src/cli/src/metadata/repair/create_table.rs

@@ -14,12 +14,12 @@
 
 use std::collections::HashMap;
 
-use client::api::v1::region::{region_request, CreateRequests, RegionRequest, RegionRequestHeader};
 use client::api::v1::CreateTableExpr;
+use client::api::v1::region::{CreateRequests, RegionRequest, RegionRequestHeader, region_request};
 use common_meta::ddl::create_logical_tables::create_region_request_builder;
 use common_meta::ddl::utils::region_storage_path;
 use common_meta::peer::Peer;
-use common_meta::rpc::router::{find_leader_regions, RegionRoute};
+use common_meta::rpc::router::{RegionRoute, find_leader_regions};
 use operator::expr_helper::column_schemas_to_defs;
 use snafu::ResultExt;
 use store_api::storage::{RegionId, TableId};

src/cli/src/metadata/snapshot.rs

@@ -19,13 +19,13 @@ use clap::{Parser, Subcommand};
 use common_base::secrets::{ExposeSecret, SecretString};
 use common_error::ext::BoxedError;
 use common_meta::snapshot::MetadataSnapshotManager;
-use object_store::services::{Fs, S3};
 use object_store::ObjectStore;
+use object_store::services::{Fs, S3};
 use snafu::{OptionExt, ResultExt};
 
+use crate::Tool;
 use crate::error::{InvalidFilePathSnafu, OpenDalSnafu, S3ConfigNotSetSnafu};
 use crate::metadata::common::StoreConfig;
-use crate::Tool;
 
 /// Subcommand for metadata snapshot operations, including saving snapshots, restoring from snapshots, and viewing snapshot information.
 #[derive(Subcommand)]
@@ -262,7 +262,9 @@ impl Tool for MetaRestoreTool {
             );
             Ok(())
         } else {
-            common_telemetry::info!("The target source is not clean, We will restore the metadata snapshot with --force.");
+            common_telemetry::info!(
+                "The target source is not clean, We will restore the metadata snapshot with --force."
+            );
             self.inner
                 .restore(&self.source_file)
                 .await

src/cli/src/metadata/utils.rs

@@ -17,9 +17,9 @@ use std::collections::VecDeque;
 use async_stream::try_stream;
 use common_catalog::consts::METRIC_ENGINE;
 use common_catalog::format_full_table_name;
+use common_meta::key::TableMetadataManager;
 use common_meta::key::table_name::TableNameKey;
 use common_meta::key::table_route::TableRouteValue;
-use common_meta::key::TableMetadataManager;
 use common_meta::kv_backend::KvBackendRef;
 use futures::Stream;
 use snafu::{OptionExt, ResultExt};

src/client/src/client.rs

@@ -14,11 +14,11 @@
 
 use std::sync::Arc;
 
+use api::v1::HealthCheckRequest;
 use api::v1::flow::flow_client::FlowClient as PbFlowClient;
 use api::v1::health_check_client::HealthCheckClient;
 use api::v1::prometheus_gateway_client::PrometheusGatewayClient;
 use api::v1::region::region_client::RegionClient as PbRegionClient;
-use api::v1::HealthCheckRequest;
 use arrow_flight::flight_service_client::FlightServiceClient;
 use common_grpc::channel_manager::{ChannelConfig, ChannelManager, ClientTlsOption};
 use parking_lot::RwLock;
@@ -27,7 +27,7 @@ use tonic::codec::CompressionEncoding;
 use tonic::transport::Channel;
 
 use crate::load_balance::{LoadBalance, Loadbalancer};
-use crate::{error, Result};
+use crate::{Result, error};
 
 pub struct FlightClient {
     addr: String,

src/client/src/client_manager.rs

@@ -21,9 +21,9 @@ use common_meta::node_manager::{DatanodeManager, DatanodeRef, FlownodeManager, F
 use common_meta::peer::Peer;
 use moka::future::{Cache, CacheBuilder};
 
+use crate::Client;
 use crate::flow::FlowRequester;
 use crate::region::RegionRequester;
-use crate::Client;
 
 pub struct NodeClients {
     channel_manager: ChannelManager,

src/client/src/database.rs

@@ -27,8 +27,8 @@ use api::v1::{
 };
 use arrow_flight::{FlightData, Ticket};
 use async_stream::stream;
-use base64::prelude::BASE64_STANDARD;
 use base64::Engine;
+use base64::prelude::BASE64_STANDARD;
 use common_catalog::build_db_string;
 use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_error::ext::BoxedError;
@@ -42,7 +42,7 @@ use common_telemetry::{error, warn};
 use futures::future;
 use futures_util::{Stream, StreamExt, TryStreamExt};
 use prost::Message;
-use snafu::{ensure, OptionExt, ResultExt};
+use snafu::{OptionExt, ResultExt, ensure};
 use tonic::metadata::{AsciiMetadataKey, AsciiMetadataValue, MetadataMap, MetadataValue};
 use tonic::transport::Channel;
 
@@ -50,7 +50,7 @@ use crate::error::{
     ConvertFlightDataSnafu, Error, FlightGetSnafu, IllegalFlightMessagesSnafu,
     InvalidTonicMetadataValueSnafu,
 };
-use crate::{error, from_grpc_response, Client, Result};
+use crate::{Client, Result, error, from_grpc_response};
 
 type FlightDataStream = Pin<Box<dyn Stream<Item = FlightData> + Send>>;
 
@@ -379,11 +379,10 @@ impl Database {
                 tonic_code,
                 e
             );
-            let error = Err(BoxedError::new(e)).with_context(|_| FlightGetSnafu {
+            Err(BoxedError::new(e)).with_context(|_| FlightGetSnafu {
                 addr: client.addr().to_string(),
                 tonic_code,
-            });
-            error
+            })
         })?;
 
         let flight_data_stream = response.into_inner();

src/client/src/error.rs

@@ -18,9 +18,9 @@ use common_error::define_from_tonic_status;
 use common_error::ext::{BoxedError, ErrorExt};
 use common_error::status_code::StatusCode;
 use common_macro::stack_trace_debug;
-use snafu::{location, Location, Snafu};
-use tonic::metadata::errors::InvalidMetadataValue;
+use snafu::{Location, Snafu, location};
 use tonic::Code;
+use tonic::metadata::errors::InvalidMetadataValue;
 
 #[derive(Snafu)]
 #[snafu(visibility(pub))]

src/client/src/flow.rs

@@ -18,8 +18,8 @@ use common_error::ext::BoxedError;
 use common_meta::node_manager::Flownode;
 use snafu::ResultExt;
 
-use crate::error::{FlowServerSnafu, Result};
 use crate::Client;
+use crate::error::{FlowServerSnafu, Result};
 
 #[derive(Debug)]
 pub struct FlowRequester {

src/client/src/region.rs

@@ -15,8 +15,8 @@
 use std::sync::Arc;
 
 use api::region::RegionResponse;
-use api::v1::region::RegionRequest;
 use api::v1::ResponseHeader;
+use api::v1::region::RegionRequest;
 use arc_swap::ArcSwapOption;
 use arrow_flight::Ticket;
 use async_stream::stream;
@@ -33,7 +33,7 @@ use common_telemetry::error;
 use common_telemetry::tracing_context::TracingContext;
 use prost::Message;
 use query::query_engine::DefaultSerializer;
-use snafu::{location, OptionExt, ResultExt};
+use snafu::{OptionExt, ResultExt, location};
 use substrait::{DFLogicalSubstraitConvertor, SubstraitPlan};
 use tokio_stream::StreamExt;
 
@@ -41,7 +41,7 @@ use crate::error::{
     self, ConvertFlightDataSnafu, FlightGetSnafu, IllegalDatabaseResponseSnafu,
     IllegalFlightMessagesSnafu, MissingFieldSnafu, Result, ServerSnafu,
 };
-use crate::{metrics, Client, Error};
+use crate::{Client, Error, metrics};
 
 #[derive(Debug)]
 pub struct RegionRequester {
@@ -115,11 +115,10 @@ impl RegionRequester {
                     flight_client.addr(),
                     tonic_code
                 );
-                let error = Err(BoxedError::new(e)).with_context(|_| FlightGetSnafu {
+                Err(BoxedError::new(e)).with_context(|_| FlightGetSnafu {
                     addr: flight_client.addr().to_string(),
                     tonic_code,
-                });
-                error
+                })
             })?;
 
         let flight_data_stream = response.into_inner();

src/cmd/src/bin/greptime.rs

@@ -18,7 +18,7 @@ use clap::{Parser, Subcommand};
 use cmd::datanode::builder::InstanceBuilder;
 use cmd::error::{InitTlsProviderSnafu, Result};
 use cmd::options::GlobalOptions;
-use cmd::{cli, datanode, flownode, frontend, metasrv, standalone, App};
+use cmd::{App, cli, datanode, flownode, frontend, metasrv, standalone};
 use common_base::Plugins;
 use common_version::{verbose_version, version};
 use servers::install_ring_crypto_provider;
@@ -143,8 +143,10 @@ async fn start(cli: Command) -> Result<()> {
 }
 
 fn setup_human_panic() {
-    human_panic::setup_panic!(human_panic::Metadata::new("GreptimeDB", version())
-        .homepage("https://github.com/GreptimeTeam/greptimedb/discussions"));
+    human_panic::setup_panic!(
+        human_panic::Metadata::new("GreptimeDB", version())
+            .homepage("https://github.com/GreptimeTeam/greptimedb/discussions")
+    );
 
     common_telemetry::set_panic_hook();
 }

src/cmd/src/cli.rs

@@ -20,7 +20,7 @@ use snafu::ResultExt;
 use tracing_appender::non_blocking::WorkerGuard;
 
 use crate::options::GlobalOptions;
-use crate::{error, App, Result};
+use crate::{App, Result, error};
 pub const APP_NAME: &str = "greptime-cli";
 use async_trait::async_trait;
 
@@ -109,7 +109,7 @@ mod tests {
 
     use crate::error::Result as CmdResult;
     use crate::options::GlobalOptions;
-    use crate::{cli, standalone, App};
+    use crate::{App, cli, standalone};
 
     #[tokio::test(flavor = "multi_thread")]
     async fn test_export_create_table_with_quoted_names() -> CmdResult<()> {

src/cmd/src/datanode.rs

@@ -20,20 +20,20 @@ use std::time::Duration;
 use async_trait::async_trait;
 use clap::Parser;
 use common_config::Configurable;
-use common_telemetry::logging::{TracingOptions, DEFAULT_LOGGING_DIR};
+use common_telemetry::logging::{DEFAULT_LOGGING_DIR, TracingOptions};
 use common_telemetry::{info, warn};
 use common_wal::config::DatanodeWalConfig;
 use datanode::datanode::Datanode;
 use meta_client::MetaClientOptions;
-use snafu::{ensure, ResultExt};
+use snafu::{ResultExt, ensure};
 use tracing_appender::non_blocking::WorkerGuard;
 
+use crate::App;
 use crate::datanode::builder::InstanceBuilder;
 use crate::error::{
     LoadLayeredConfigSnafu, MissingConfigSnafu, Result, ShutdownDatanodeSnafu, StartDatanodeSnafu,
 };
 use crate::options::{GlobalOptions, GreptimeOptions};
-use crate::App;
 
 pub const APP_NAME: &str = "greptime-datanode";
 
@@ -187,29 +187,39 @@ impl StartCommand {
         if let Some(addr) = &self.rpc_bind_addr {
             opts.grpc.bind_addr.clone_from(addr);
         } else if let Some(addr) = &opts.rpc_addr {
-            warn!("Use the deprecated attribute `DatanodeOptions.rpc_addr`, please use `grpc.addr` instead.");
+            warn!(
+                "Use the deprecated attribute `DatanodeOptions.rpc_addr`, please use `grpc.addr` instead."
+            );
             opts.grpc.bind_addr.clone_from(addr);
         }
 
         if let Some(server_addr) = &self.rpc_server_addr {
             opts.grpc.server_addr.clone_from(server_addr);
         } else if let Some(server_addr) = &opts.rpc_hostname {
-            warn!("Use the deprecated attribute `DatanodeOptions.rpc_hostname`, please use `grpc.hostname` instead.");
+            warn!(
+                "Use the deprecated attribute `DatanodeOptions.rpc_hostname`, please use `grpc.hostname` instead."
+            );
             opts.grpc.server_addr.clone_from(server_addr);
         }
 
         if let Some(runtime_size) = opts.rpc_runtime_size {
-            warn!("Use the deprecated attribute `DatanodeOptions.rpc_runtime_size`, please use `grpc.runtime_size` instead.");
+            warn!(
+                "Use the deprecated attribute `DatanodeOptions.rpc_runtime_size`, please use `grpc.runtime_size` instead."
+            );
             opts.grpc.runtime_size = runtime_size;
         }
 
         if let Some(max_recv_message_size) = opts.rpc_max_recv_message_size {
-            warn!("Use the deprecated attribute `DatanodeOptions.rpc_max_recv_message_size`, please use `grpc.max_recv_message_size` instead.");
+            warn!(
+                "Use the deprecated attribute `DatanodeOptions.rpc_max_recv_message_size`, please use `grpc.max_recv_message_size` instead."
+            );
             opts.grpc.max_recv_message_size = max_recv_message_size;
         }
 
         if let Some(max_send_message_size) = opts.rpc_max_send_message_size {
-            warn!("Use the deprecated attribute `DatanodeOptions.rpc_max_send_message_size`, please use `grpc.max_send_message_size` instead.");
+            warn!(
+                "Use the deprecated attribute `DatanodeOptions.rpc_max_send_message_size`, please use `grpc.max_send_message_size` instead."
+            );
             opts.grpc.max_send_message_size = max_send_message_size;
         }
 
@@ -430,20 +440,24 @@ mod tests {
 
     #[test]
     fn test_try_from_cmd() {
-        assert!((StartCommand {
+        assert!(
+            (StartCommand {
                 metasrv_addrs: Some(vec!["127.0.0.1:3002".to_string()]),
                 ..Default::default()
             })
             .load_options(&GlobalOptions::default())
-        .is_err());
+            .is_err()
+        );
 
         // Providing node_id but leave metasrv_addr absent is ok since metasrv_addr has default value
-        assert!((StartCommand {
+        assert!(
+            (StartCommand {
                 node_id: Some(42),
                 ..Default::default()
             })
             .load_options(&GlobalOptions::default())
-        .is_ok());
+            .is_ok()
+        );
     }
 
     #[test]

src/cmd/src/datanode/builder.rs

@@ -26,7 +26,7 @@ use meta_client::MetaClientType;
 use snafu::{OptionExt, ResultExt};
 use tracing_appender::non_blocking::WorkerGuard;
 
-use crate::datanode::{DatanodeOptions, Instance, APP_NAME};
+use crate::datanode::{APP_NAME, DatanodeOptions, Instance};
 use crate::error::{MetaClientInitSnafu, MissingConfigSnafu, Result, StartDatanodeSnafu};
 use crate::{create_resource_limit_metrics, log_versions, maybe_activate_heap_profile};
 

src/cmd/src/flownode.rs

@@ -25,20 +25,20 @@ use common_base::Plugins;
 use common_config::{Configurable, DEFAULT_DATA_HOME};
 use common_grpc::channel_manager::ChannelConfig;
 use common_meta::cache::{CacheRegistryBuilder, LayeredCacheRegistryBuilder};
+use common_meta::heartbeat::handler::HandlerGroupExecutor;
 use common_meta::heartbeat::handler::invalidate_table_cache::InvalidateCacheHandler;
 use common_meta::heartbeat::handler::parse_mailbox_message::ParseMailboxMessageHandler;
-use common_meta::heartbeat::handler::HandlerGroupExecutor;
-use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::TableMetadataManager;
+use common_meta::key::flow::FlowMetadataManager;
 use common_telemetry::info;
-use common_telemetry::logging::{TracingOptions, DEFAULT_LOGGING_DIR};
+use common_telemetry::logging::{DEFAULT_LOGGING_DIR, TracingOptions};
 use common_version::{short_version, verbose_version};
 use flow::{
-    get_flow_auth_options, FlownodeBuilder, FlownodeInstance, FlownodeServiceBuilder,
-    FrontendClient, FrontendInvoker,
+    FlownodeBuilder, FlownodeInstance, FlownodeServiceBuilder, FrontendClient, FrontendInvoker,
+    get_flow_auth_options,
 };
 use meta_client::{MetaClientOptions, MetaClientType};
-use snafu::{ensure, OptionExt, ResultExt};
+use snafu::{OptionExt, ResultExt, ensure};
 use tracing_appender::non_blocking::WorkerGuard;
 
 use crate::error::{
@@ -46,7 +46,7 @@ use crate::error::{
     MissingConfigSnafu, Result, ShutdownFlownodeSnafu, StartFlownodeSnafu,
 };
 use crate::options::{GlobalOptions, GreptimeOptions};
-use crate::{create_resource_limit_metrics, log_versions, maybe_activate_heap_profile, App};
+use crate::{App, create_resource_limit_metrics, log_versions, maybe_activate_heap_profile};
 
 pub const APP_NAME: &str = "greptime-flownode";
 

src/cmd/src/frontend.rs

@@ -27,11 +27,11 @@ use common_base::Plugins;
 use common_config::{Configurable, DEFAULT_DATA_HOME};
 use common_grpc::channel_manager::ChannelConfig;
 use common_meta::cache::{CacheRegistryBuilder, LayeredCacheRegistryBuilder};
+use common_meta::heartbeat::handler::HandlerGroupExecutor;
 use common_meta::heartbeat::handler::invalidate_table_cache::InvalidateCacheHandler;
 use common_meta::heartbeat::handler::parse_mailbox_message::ParseMailboxMessageHandler;
-use common_meta::heartbeat::handler::HandlerGroupExecutor;
 use common_telemetry::info;
-use common_telemetry::logging::{TracingOptions, DEFAULT_LOGGING_DIR};
+use common_telemetry::logging::{DEFAULT_LOGGING_DIR, TracingOptions};
 use common_time::timezone::set_default_timezone;
 use common_version::{short_version, verbose_version};
 use frontend::frontend::Frontend;
@@ -48,7 +48,7 @@ use tracing_appender::non_blocking::WorkerGuard;
 
 use crate::error::{self, Result};
 use crate::options::{GlobalOptions, GreptimeOptions};
-use crate::{create_resource_limit_metrics, log_versions, maybe_activate_heap_profile, App};
+use crate::{App, create_resource_limit_metrics, log_versions, maybe_activate_heap_profile};
 
 type FrontendOptions = GreptimeOptions<frontend::frontend::FrontendOptions>;
 

src/cmd/src/lib.rs

@@ -46,7 +46,7 @@ lazy_static::lazy_static! {
 /// wait for the close signal, for unix platform it's SIGINT or SIGTERM
 #[cfg(unix)]
 async fn start_wait_for_close_signal() -> std::io::Result<()> {
-    use tokio::signal::unix::{signal, SignalKind};
+    use tokio::signal::unix::{SignalKind, signal};
     let mut sigint = signal(SignalKind::interrupt())?;
     let mut sigterm = signal(SignalKind::terminate())?;
 
@@ -93,14 +93,14 @@ pub trait App: Send {
 
         self.start().await?;
 
-        if self.wait_signal() {
-            if let Err(e) = start_wait_for_close_signal().await {
+        if self.wait_signal()
+            && let Err(e) = start_wait_for_close_signal().await
+        {
             error!(e; "Failed to listen for close signal");
             // It's unusual to fail to listen for close signal, maybe there's something unexpected in
             // the underlying system. So we stop the app instead of running nonetheless to let people
             // investigate the issue.
         }
-        }
 
         self.stop().await?;
         info!("Goodbye!");

src/cmd/src/metasrv.rs

@@ -21,7 +21,7 @@ use clap::Parser;
 use common_base::Plugins;
 use common_config::Configurable;
 use common_telemetry::info;
-use common_telemetry::logging::{TracingOptions, DEFAULT_LOGGING_DIR};
+use common_telemetry::logging::{DEFAULT_LOGGING_DIR, TracingOptions};
 use common_version::{short_version, verbose_version};
 use meta_srv::bootstrap::MetasrvInstance;
 use meta_srv::metasrv::BackendImpl;
@@ -30,7 +30,7 @@ use tracing_appender::non_blocking::WorkerGuard;
 
 use crate::error::{self, LoadLayeredConfigSnafu, Result, StartMetaServerSnafu};
 use crate::options::{GlobalOptions, GreptimeOptions};
-use crate::{create_resource_limit_metrics, log_versions, maybe_activate_heap_profile, App};
+use crate::{App, create_resource_limit_metrics, log_versions, maybe_activate_heap_profile};
 
 type MetasrvOptions = GreptimeOptions<meta_srv::metasrv::MetasrvOptions>;
 

src/cmd/src/standalone.rs

@@ -24,10 +24,10 @@ use catalog::kvbackend::KvBackendCatalogManagerBuilder;
 use catalog::process_manager::ProcessManager;
 use clap::Parser;
 use client::api::v1::meta::RegionRole;
-use common_base::readable_size::ReadableSize;
 use common_base::Plugins;
+use common_base::readable_size::ReadableSize;
 use common_catalog::consts::{MIN_USER_FLOW_ID, MIN_USER_TABLE_ID};
-use common_config::{metadata_store_dir, Configurable, KvBackendConfig};
+use common_config::{Configurable, KvBackendConfig, metadata_store_dir};
 use common_error::ext::BoxedError;
 use common_meta::cache::LayeredCacheRegistryBuilder;
 use common_meta::cluster::{NodeInfo, NodeStatus};
@@ -36,8 +36,8 @@ use common_meta::ddl::flow_meta::FlowMetadataAllocator;
 use common_meta::ddl::table_meta::TableMetadataAllocator;
 use common_meta::ddl::{DdlContext, NoopRegionFailureDetectorControl};
 use common_meta::ddl_manager::DdlManager;
-use common_meta::key::flow::flow_state::FlowStat;
 use common_meta::key::flow::FlowMetadataManager;
+use common_meta::key::flow::flow_state::FlowStat;
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
 use common_meta::peer::Peer;
@@ -45,12 +45,12 @@ use common_meta::procedure_executor::LocalProcedureExecutor;
 use common_meta::region_keeper::MemoryRegionKeeper;
 use common_meta::region_registry::LeaderRegionRegistry;
 use common_meta::sequence::SequenceBuilder;
-use common_meta::wal_options_allocator::{build_wal_options_allocator, WalOptionsAllocatorRef};
+use common_meta::wal_options_allocator::{WalOptionsAllocatorRef, build_wal_options_allocator};
 use common_options::memory::MemoryOptions;
 use common_procedure::{ProcedureInfo, ProcedureManagerRef};
 use common_telemetry::info;
 use common_telemetry::logging::{
-    LoggingOptions, SlowQueryOptions, TracingOptions, DEFAULT_LOGGING_DIR,
+    DEFAULT_LOGGING_DIR, LoggingOptions, SlowQueryOptions, TracingOptions,
 };
 use common_time::timezone::set_default_timezone;
 use common_version::{short_version, verbose_version};
@@ -85,7 +85,7 @@ use tracing_appender::non_blocking::WorkerGuard;
 
 use crate::error::{Result, StartFlownodeSnafu};
 use crate::options::{GlobalOptions, GreptimeOptions};
-use crate::{create_resource_limit_metrics, error, log_versions, maybe_activate_heap_profile, App};
+use crate::{App, create_resource_limit_metrics, error, log_versions, maybe_activate_heap_profile};
 
 pub const APP_NAME: &str = "greptime-standalone";
 

src/cmd/tests/load_config_test.rs

@@ -18,9 +18,9 @@ use cmd::options::GreptimeOptions;
 use cmd::standalone::StandaloneOptions;
 use common_config::{Configurable, DEFAULT_DATA_HOME};
 use common_options::datanode::{ClientOptions, DatanodeClientOptions};
-use common_telemetry::logging::{LoggingOptions, DEFAULT_LOGGING_DIR, DEFAULT_OTLP_HTTP_ENDPOINT};
-use common_wal::config::raft_engine::RaftEngineConfig;
+use common_telemetry::logging::{DEFAULT_LOGGING_DIR, DEFAULT_OTLP_HTTP_ENDPOINT, LoggingOptions};
 use common_wal::config::DatanodeWalConfig;
+use common_wal::config::raft_engine::RaftEngineConfig;
 use datanode::config::{DatanodeOptions, RegionEngineConfig, StorageConfig};
 use file_engine::config::EngineConfig as FileEngineConfig;
 use flow::FlownodeOptions;

src/common/base/src/cancellation.rs

@@ -19,8 +19,8 @@
 use std::fmt::{Debug, Display, Formatter};
 use std::future::Future;
 use std::pin::Pin;
-use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::task::{Context, Poll};
 
 use futures::task::AtomicWaker;

src/common/base/src/range_read.rs

@@ -17,8 +17,8 @@ use std::io;
 use std::ops::Range;
 use std::path::Path;
 use std::pin::Pin;
-use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::task::{Context, Poll};
 
 use async_trait::async_trait;

src/common/base/src/readable_size.rs

@@ -7,7 +7,7 @@ use std::ops::{Div, Mul};
 use std::str::FromStr;
 
 use serde::de::{Unexpected, Visitor};
-use serde::{de, Deserialize, Deserializer, Serialize, Serializer};
+use serde::{Deserialize, Deserializer, Serialize, Serializer, de};
 
 const UNIT: u64 = 1;
 

src/common/base/src/secrets.rs

@@ -34,7 +34,7 @@
 use std::fmt::Debug;
 use std::{any, fmt};
 
-use serde::{de, ser, Deserialize, Serialize};
+use serde::{Deserialize, Serialize, de, ser};
 use zeroize::{Zeroize, ZeroizeOnDrop};
 
 /// Wrapper type for strings that contains secrets. See also [SecretBox].

src/common/config/src/config.rs

@@ -13,8 +13,8 @@
 // limitations under the License.
 
 use config::{Environment, File, FileFormat};
-use serde::de::DeserializeOwned;
 use serde::Serialize;
+use serde::de::DeserializeOwned;
 use snafu::ResultExt;
 
 use crate::error::{LoadLayeredConfigSnafu, Result, SerdeJsonSnafu, TomlFormatSnafu};

src/common/datasource/src/file_format.rs

@@ -42,11 +42,11 @@ use self::csv::CsvFormat;
 use self::json::JsonFormat;
 use self::orc::OrcFormat;
 use self::parquet::ParquetFormat;
+use crate::DEFAULT_WRITE_BUFFER_SIZE;
 use crate::buffered_writer::{DfRecordBatchEncoder, LazyBufferedWriter};
 use crate::compression::CompressionType;
 use crate::error::{self, Result};
 use crate::share_buffer::SharedBuffer;
-use crate::DEFAULT_WRITE_BUFFER_SIZE;
 
 pub const FORMAT_COMPRESSION_TYPE: &str = "compression_type";
 pub const FORMAT_DELIMITER: &str = "delimiter";
@@ -158,10 +158,10 @@ pub fn open_with_decoder<T: ArrowDecoder, F: Fn() -> DataFusionResult<T>>(
 
         let stream = futures::stream::poll_fn(move |cx| {
             loop {
-                if buffered.is_empty() {
-                    if let Some(result) = futures::ready!(upstream.poll_next_unpin(cx)) {
+                if buffered.is_empty()
+                    && let Some(result) = futures::ready!(upstream.poll_next_unpin(cx))
+                {
                     buffered = result?;
-                    };
                 }
 
                 let decoded = decoder.decode(buffered.as_ref())?;

src/common/datasource/src/file_format/csv.rs

@@ -30,7 +30,7 @@ use tokio_util::io::SyncIoBridge;
 use crate::buffered_writer::DfRecordBatchEncoder;
 use crate::compression::CompressionType;
 use crate::error::{self, Result};
-use crate::file_format::{self, stream_to_file, FileFormat};
+use crate::file_format::{self, FileFormat, stream_to_file};
 use crate::share_buffer::SharedBuffer;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -160,8 +160,8 @@ mod tests {
 
     use super::*;
     use crate::file_format::{
-        FileFormat, FORMAT_COMPRESSION_TYPE, FORMAT_DELIMITER, FORMAT_HAS_HEADER,
-        FORMAT_SCHEMA_INFER_MAX_RECORD,
+        FORMAT_COMPRESSION_TYPE, FORMAT_DELIMITER, FORMAT_HAS_HEADER,
+        FORMAT_SCHEMA_INFER_MAX_RECORD, FileFormat,
     };
     use crate::test_util::{format_schema, test_store};
 

src/common/datasource/src/file_format/json.rs

@@ -17,7 +17,7 @@ use std::io::BufReader;
 use std::str::FromStr;
 
 use arrow::json;
-use arrow::json::reader::{infer_json_schema_from_iterator, ValueIter};
+use arrow::json::reader::{ValueIter, infer_json_schema_from_iterator};
 use arrow::json::writer::LineDelimited;
 use arrow::record_batch::RecordBatch;
 use arrow_schema::Schema;
@@ -32,7 +32,7 @@ use tokio_util::io::SyncIoBridge;
 use crate::buffered_writer::DfRecordBatchEncoder;
 use crate::compression::CompressionType;
 use crate::error::{self, Result};
-use crate::file_format::{self, stream_to_file, FileFormat};
+use crate::file_format::{self, FileFormat, stream_to_file};
 use crate::share_buffer::SharedBuffer;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -133,7 +133,7 @@ mod tests {
     use common_test_util::find_workspace_path;
 
     use super::*;
-    use crate::file_format::{FileFormat, FORMAT_COMPRESSION_TYPE, FORMAT_SCHEMA_INFER_MAX_RECORD};
+    use crate::file_format::{FORMAT_COMPRESSION_TYPE, FORMAT_SCHEMA_INFER_MAX_RECORD, FileFormat};
     use crate::test_util::{format_schema, test_store};
 
     fn test_data_root() -> String {

src/common/datasource/src/file_format/orc.rs

@@ -15,8 +15,8 @@
 use arrow_schema::Schema;
 use async_trait::async_trait;
 use bytes::Bytes;
-use futures::future::BoxFuture;
 use futures::FutureExt;
+use futures::future::BoxFuture;
 use object_store::ObjectStore;
 use orc_rust::arrow_reader::ArrowReaderBuilder;
 use orc_rust::async_arrow_reader::ArrowStreamReader;

src/common/datasource/src/file_format/parquet.rs

@@ -21,29 +21,29 @@ use async_trait::async_trait;
 use datafusion::datasource::physical_plan::{FileMeta, ParquetFileReaderFactory};
 use datafusion::error::Result as DatafusionResult;
 use datafusion::parquet::arrow::async_reader::AsyncFileReader;
-use datafusion::parquet::arrow::{parquet_to_arrow_schema, ArrowWriter};
+use datafusion::parquet::arrow::{ArrowWriter, parquet_to_arrow_schema};
 use datafusion::parquet::errors::{ParquetError, Result as ParquetResult};
 use datafusion::parquet::file::metadata::ParquetMetaData;
 use datafusion::parquet::format::FileMetaData;
-use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
 use datafusion::physical_plan::SendableRecordBatchStream;
+use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
 use datatypes::schema::SchemaRef;
-use futures::future::BoxFuture;
 use futures::StreamExt;
+use futures::future::BoxFuture;
 use object_store::{FuturesAsyncReader, ObjectStore};
-use parquet::arrow::arrow_reader::ArrowReaderOptions;
 use parquet::arrow::AsyncArrowWriter;
+use parquet::arrow::arrow_reader::ArrowReaderOptions;
 use parquet::basic::{Compression, Encoding, ZstdLevel};
 use parquet::file::properties::{WriterProperties, WriterPropertiesBuilder};
 use parquet::schema::types::ColumnPath;
 use snafu::ResultExt;
 use tokio_util::compat::{Compat, FuturesAsyncReadCompatExt, FuturesAsyncWriteCompatExt};
 
+use crate::DEFAULT_WRITE_BUFFER_SIZE;
 use crate::buffered_writer::{ArrowWriterCloser, DfRecordBatchEncoder};
 use crate::error::{self, Result, WriteObjectSnafu, WriteParquetSnafu};
 use crate::file_format::FileFormat;
 use crate::share_buffer::SharedBuffer;
-use crate::DEFAULT_WRITE_BUFFER_SIZE;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
 pub struct ParquetFormat {}

src/common/datasource/src/file_format/tests.rs

@@ -24,8 +24,8 @@ use datafusion::datasource::physical_plan::{
 };
 use datafusion::datasource::source::DataSourceExec;
 use datafusion::execution::context::TaskContext;
-use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
 use datafusion::physical_plan::ExecutionPlan;
+use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
 use datafusion::prelude::SessionContext;
 use datafusion_orc::OrcSource;
 use futures::StreamExt;

src/common/datasource/src/lister.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use futures::{future, TryStreamExt};
+use futures::{TryStreamExt, future};
 use object_store::{Entry, ObjectStore};
 use regex::Regex;
 use snafu::ResultExt;

src/common/datasource/src/object_store/fs.rs

@@ -12,9 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use object_store::ObjectStore;
 use object_store::services::Fs;
 use object_store::util::DefaultLoggingInterceptor;
-use object_store::ObjectStore;
 use snafu::ResultExt;
 
 use crate::error::{BuildBackendSnafu, Result};

src/common/datasource/src/object_store/oss.rs

@@ -14,8 +14,8 @@
 
 use std::collections::HashMap;
 
-use object_store::services::Oss;
 use object_store::ObjectStore;
+use object_store::services::Oss;
 use snafu::ResultExt;
 
 use crate::error::{self, Result};

src/common/datasource/src/object_store/s3.rs

@@ -14,9 +14,9 @@
 
 use std::collections::HashMap;
 
+use object_store::ObjectStore;
 use object_store::services::S3;
 use object_store::util::DefaultLoggingInterceptor;
-use object_store::ObjectStore;
 use snafu::ResultExt;
 
 use crate::error::{self, Result};

src/common/datasource/src/test_util.rs

@@ -15,7 +15,7 @@
 use std::sync::Arc;
 
 use arrow_schema::{DataType, Field, Schema, SchemaRef};
-use common_test_util::temp_dir::{create_temp_dir, TempDir};
+use common_test_util::temp_dir::{TempDir, create_temp_dir};
 use datafusion::datasource::file_format::file_compression_type::FileCompressionType;
 use datafusion::datasource::listing::PartitionedFile;
 use datafusion::datasource::object_store::ObjectStoreUrl;
@@ -24,8 +24,8 @@ use datafusion::datasource::physical_plan::{
     JsonOpener, JsonSource,
 };
 use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
-use object_store::services::Fs;
 use object_store::ObjectStore;
+use object_store::services::Fs;
 
 use crate::file_format::csv::stream_to_csv;
 use crate::file_format::json::stream_to_json;
@@ -111,7 +111,8 @@ pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usi
 
     let output_path = format!("{}/{}", dir.path().display(), "output");
 
-    assert!(stream_to_json(
+    assert!(
+        stream_to_json(
             Box::pin(stream),
             tmp_store.clone(),
             &output_path,
@@ -119,7 +120,8 @@ pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usi
             8
         )
         .await
-    .is_ok());
+        .is_ok()
+    );
 
     let written = tmp_store.read(&output_path).await.unwrap();
     let origin = store.read(origin_path).await.unwrap();
@@ -148,7 +150,8 @@ pub async fn setup_stream_to_csv_test(origin_path: &str, threshold: impl Fn(usiz
 
     let output_path = format!("{}/{}", dir.path().display(), "output");
 
-    assert!(stream_to_csv(
+    assert!(
+        stream_to_csv(
             Box::pin(stream),
             tmp_store.clone(),
             &output_path,
@@ -156,7 +159,8 @@ pub async fn setup_stream_to_csv_test(origin_path: &str, threshold: impl Fn(usiz
             8
         )
         .await
-    .is_ok());
+        .is_ok()
+    );
 
     let written = tmp_store.read(&output_path).await.unwrap();
     let origin = store.read(origin_path).await.unwrap();

src/common/event-recorder/src/recorder.rs

@@ -32,7 +32,7 @@ use humantime::format_duration;
 use itertools::Itertools;
 use serde::{Deserialize, Serialize};
 use store_api::mito_engine_options::{APPEND_MODE_KEY, TTL_KEY};
-use tokio::sync::mpsc::{channel, Receiver, Sender};
+use tokio::sync::mpsc::{Receiver, Sender, channel};
 use tokio::task::JoinHandle;
 use tokio::time::sleep;
 use tokio_util::sync::CancellationToken;

src/common/frontend/src/selector.rs

@@ -18,8 +18,8 @@ use std::time::Duration;
 use common_grpc::channel_manager::{ChannelConfig, ChannelManager};
 use common_meta::cluster::{ClusterInfo, NodeInfo, Role};
 use greptime_proto::v1::frontend::{
-    frontend_client, KillProcessRequest, KillProcessResponse, ListProcessRequest,
-    ListProcessResponse,
+    KillProcessRequest, KillProcessResponse, ListProcessRequest, ListProcessResponse,
+    frontend_client,
 };
 use meta_client::MetaClientRef;
 use snafu::ResultExt;

src/common/frontend/src/slow_query_event.rs

@@ -16,8 +16,8 @@ use std::any::Any;
 
 use api::v1::value::ValueData;
 use api::v1::{ColumnDataType, ColumnSchema, Row, SemanticType};
-use common_event_recorder::error::Result;
 use common_event_recorder::Event;
+use common_event_recorder::error::Result;
 use serde::Serialize;
 
 pub const SLOW_QUERY_TABLE_NAME: &str = "slow_queries";

src/common/function/src/admin/flush_compact_table.rs

@@ -14,7 +14,7 @@
 
 use std::str::FromStr;
 
-use api::v1::region::{compact_request, StrictWindow};
+use api::v1::region::{StrictWindow, compact_request};
 use arrow::datatypes::DataType as ArrowDataType;
 use common_error::ext::BoxedError;
 use common_macro::admin_fn;
@@ -27,7 +27,7 @@ use datafusion_expr::{Signature, Volatility};
 use datatypes::prelude::*;
 use session::context::QueryContextRef;
 use session::table_name::table_name_to_full_name;
-use snafu::{ensure, ResultExt};
+use snafu::{ResultExt, ensure};
 use table::requests::{CompactTableRequest, FlushTableRequest};
 
 use crate::handlers::TableMutationHandlerRef;
@@ -133,13 +133,19 @@ fn parse_compact_params(
             table_name,
             compact_request::Options::Regular(Default::default()),
         ),
-        [ValueRef::String(table_name), ValueRef::String(compact_ty_str)] => {
+        [
+            ValueRef::String(table_name),
+            ValueRef::String(compact_ty_str),
+        ] => {
             let compact_type = parse_compact_type(compact_ty_str, None)?;
             (table_name, compact_type)
         }
 
-        [ValueRef::String(table_name), ValueRef::String(compact_ty_str), ValueRef::String(options_str)] =>
-        {
+        [
+            ValueRef::String(table_name),
+            ValueRef::String(compact_ty_str),
+            ValueRef::String(options_str),
+        ] => {
             let compact_type = parse_compact_type(compact_ty_str, Some(options_str))?;
             (table_name, compact_type)
         }
@@ -148,7 +154,7 @@ fn parse_compact_params(
                 function: "compact_table",
                 datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
             }
-            .fail()
+            .fail();
         }
     };
 
@@ -384,22 +390,26 @@ mod tests {
             ),
         ]);
 
-        assert!(parse_compact_params(
+        assert!(
+            parse_compact_params(
                 &["table", "strict_window", "abc"]
                     .into_iter()
                     .map(ValueRef::String)
                     .collect::<Vec<_>>(),
                 &QueryContext::arc(),
             )
-        .is_err());
+            .is_err()
+        );
 
-        assert!(parse_compact_params(
+        assert!(
+            parse_compact_params(
                 &["a.b.table", "strict_window", "abc"]
                     .into_iter()
                     .map(ValueRef::String)
                     .collect::<Vec<_>>(),
                 &QueryContext::arc(),
             )
-        .is_err());
+            .is_err()
+        );
     }
 }

src/common/function/src/admin/reconcile_table.rs

@@ -54,15 +54,16 @@ pub(crate) async fn reconcile_table(
 ) -> Result<Value> {
     let (table_name, resolve_strategy) = match params {
         [ValueRef::String(table_name)] => (table_name, ResolveStrategy::UseLatest),
-        [ValueRef::String(table_name), ValueRef::String(resolve_strategy)] => {
-            (table_name, parse_resolve_strategy(resolve_strategy)?)
-        }
+        [
+            ValueRef::String(table_name),
+            ValueRef::String(resolve_strategy),
+        ] => (table_name, parse_resolve_strategy(resolve_strategy)?),
         _ => {
             return UnsupportedInputDataTypeSnafu {
                 function: FN_NAME,
                 datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
             }
-            .fail()
+            .fail();
         }
     };
     let (catalog_name, schema_name, table_name) = table_name_to_full_name(table_name, query_ctx)

src/common/function/src/aggrs/aggr_wrapper.rs

@@ -28,8 +28,8 @@ use arrow::array::StructArray;
 use arrow_schema::{FieldRef, Fields};
 use common_telemetry::debug;
 use datafusion::functions_aggregate::all_default_aggregate_functions;
-use datafusion::optimizer::analyzer::type_coercion::TypeCoercion;
 use datafusion::optimizer::AnalyzerRule;
+use datafusion::optimizer::analyzer::type_coercion::TypeCoercion;
 use datafusion::physical_planner::create_aggregate_expr_and_maybe_filter;
 use datafusion_common::{Column, ScalarValue};
 use datafusion_expr::expr::{AggregateFunction, AggregateFunctionParams};
@@ -208,7 +208,8 @@ impl StateMergeHelper {
         if *upper_plan.schema() != *aggr_plan.schema() {
             return Err(datafusion_common::DataFusionError::Internal(format!(
                 "Upper aggregate plan's schema is not the same as the original aggregate plan's schema: \n[transformed]:{}\n[original]:{}",
-                upper_plan.schema(), aggr_plan.schema()
+                upper_plan.schema(),
+                aggr_plan.schema()
             )));
         }
 

src/common/function/src/aggrs/aggr_wrapper/tests.rs

@@ -25,8 +25,8 @@ use datafusion::datasource::DefaultTableSource;
 use datafusion::execution::{RecordBatchStream, SendableRecordBatchStream, TaskContext};
 use datafusion::functions_aggregate::average::avg_udaf;
 use datafusion::functions_aggregate::sum::sum_udaf;
-use datafusion::optimizer::analyzer::type_coercion::TypeCoercion;
 use datafusion::optimizer::AnalyzerRule;
+use datafusion::optimizer::analyzer::type_coercion::TypeCoercion;
 use datafusion::physical_plan::aggregates::AggregateExec;
 use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion::physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties};
@@ -35,7 +35,7 @@ use datafusion::prelude::SessionContext;
 use datafusion_common::{Column, TableReference};
 use datafusion_expr::expr::AggregateFunction;
 use datafusion_expr::sqlparser::ast::NullTreatment;
-use datafusion_expr::{lit, Aggregate, Expr, LogicalPlan, SortExpr, TableScan};
+use datafusion_expr::{Aggregate, Expr, LogicalPlan, SortExpr, TableScan, lit};
 use datafusion_physical_expr::aggregate::AggregateExprBuilder;
 use datafusion_physical_expr::{EquivalenceProperties, Partitioning};
 use datatypes::arrow_array::StringArray;
@@ -264,7 +264,8 @@ async fn test_sum_udaf() {
         Aggregate::try_new(
             Arc::new(expected_lower_plan),
             vec![],
-            vec![Expr::AggregateFunction(AggregateFunction::new_udf(
+            vec![
+                Expr::AggregateFunction(AggregateFunction::new_udf(
                     Arc::new(
                         MergeWrapper::new(
                             sum.clone(),
@@ -293,7 +294,8 @@ async fn test_sum_udaf() {
                     vec![],
                     None,
                 ))
-            .alias("sum(number)")],
+                .alias("sum(number)"),
+            ],
         )
         .unwrap(),
     );
@@ -444,7 +446,8 @@ async fn test_avg_udaf() {
         Aggregate::try_new(
             Arc::new(coerced_aggr_state_plan.clone()),
             vec![],
-            vec![Expr::AggregateFunction(AggregateFunction::new_udf(
+            vec![
+                Expr::AggregateFunction(AggregateFunction::new_udf(
                     Arc::new(expected_merge_fn.into()),
                     vec![Expr::Column(Column::new_unqualified("__avg_state(number)"))],
                     false,
@@ -452,7 +455,8 @@ async fn test_avg_udaf() {
                     vec![],
                     None,
                 ))
-            .alias("avg(number)")],
+                .alias("avg(number)"),
+            ],
         )
         .unwrap(),
     );

src/common/function/src/aggrs/approximate/hll.rs

@@ -105,11 +105,11 @@ impl HllState {
     }
 
     fn merge(&mut self, raw: &[u8]) {
-        if let Ok(serialized) = bincode::deserialize::<HllStateType>(raw) {
-            if let Ok(()) = self.hll.merge(&serialized) {
+        if let Ok(serialized) = bincode::deserialize::<HllStateType>(raw)
+            && let Ok(()) = self.hll.merge(&serialized)
+        {
             return;
         }
-        }
         trace!("Warning: Failed to merge HyperLogLog from {:?}", raw);
     }
 
@@ -153,7 +153,7 @@ impl DfAccumulator for HllState {
                 return not_impl_err!(
                     "HLL functions do not support data type: {}",
                     array.data_type()
-                )
+                );
             }
         }