fix thiserror/anyhow, fmt, some clippy

risingwavelabs · Sep 9, 2023 · ab8c61c · ab8c61c
1 parent 5ab88a8
commit ab8c61c
Show file tree

Hide file tree

Showing 9 changed files with 27 additions and 27 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/src/connector/build.rs b/src/connector/build.rs
@@ -17,7 +17,7 @@ fn main() {
 
     println!("cargo:rerun-if-changed={}", proto_dir);
 
-    let proto_files = vec!["recursive"];
+    let proto_files = ["recursive"];
     let protos: Vec<String> = proto_files
         .iter()
         .map(|f| format!("{}/{}.proto", proto_dir, f))

diff --git a/src/frontend/src/binder/expr/function.rs b/src/frontend/src/binder/expr/function.rs
@@ -185,10 +185,16 @@ impl Binder {
             }
         };
 
-        let ast::FunctionArgExpr::Expr(ast::Expr::LambdaFunction { args: lambda_args, body: lambda_body }) = lambda.get_expr() else {
+        let ast::FunctionArgExpr::Expr(ast::Expr::LambdaFunction {
+            args: lambda_args,
+            body: lambda_body,
+        }) = lambda.get_expr()
+        else {
             return Err(ErrorCode::BindError(
-                "The `lambda` argument for `array_transform` should be a lambda function".to_string()
-            ).into());
+                "The `lambda` argument for `array_transform` should be a lambda function"
+                    .to_string(),
+            )
+            .into());
         };
 
         let [lambda_arg] = <[Ident; 1]>::try_from(lambda_args).map_err(|args| -> RwError {

diff --git a/src/jni_core/src/lib.rs b/src/jni_core/src/lib.rs
@@ -13,7 +13,6 @@
 // limitations under the License.
 
 #![feature(error_generic_member_access)]
-#![feature(provide_any)]
 #![feature(lazy_cell)]
 #![feature(once_cell_try)]
 #![feature(type_alias_impl_trait)]

diff --git a/src/storage/src/hummock/event_handler/hummock_event_handler.rs b/src/storage/src/hummock/event_handler/hummock_event_handler.rs
@@ -511,7 +511,7 @@ impl HummockEventHandler {
             UploaderEvent::ImmMerged(merge_output) => {
                 // update read version for corresponding table shards
                 let read_guard = self.read_version_mapping.read();
-                read_guard.get(&merge_output.table_id).map_or((), |shards| {
+                if let Some(shards) = read_guard.get(&merge_output.table_id) {
                     shards.get(&merge_output.instance_id).map_or_else(
                         || {
                             warn!(
@@ -525,7 +525,7 @@ impl HummockEventHandler {
                             ));
                         },
                     )
-                });
+                }
             }
         }
     }

diff --git a/src/storage/src/monitor/traced_store.rs b/src/storage/src/monitor/traced_store.rs
@@ -104,8 +104,7 @@ impl<S> TracedStateStore<S> {
     }
 }
 
-type TracedStateStoreIterStream<S: StateStoreIterItemStream> =
-    impl StateStoreIterItemStream;
+type TracedStateStoreIterStream<S: StateStoreIterItemStream> = impl StateStoreIterItemStream;
 
 impl<S: LocalStateStore> LocalStateStore for TracedStateStore<S> {
     type IterStream<'a> = impl StateStoreIterItemStream + 'a;

diff --git a/src/stream/src/executor/over_window/general.rs b/src/stream/src/executor/over_window/general.rs
@@ -307,15 +307,15 @@ impl<S: StateStore> OverWindowExecutor<S> {
             match record {
                 Record::Insert { new_row } => {
                     let part_key = this.get_partition_key(new_row).into();
-                    let part_delta = deltas.entry(part_key).or_insert(PartitionDelta::new());
+                    let part_delta = deltas.entry(part_key).or_default();
                     part_delta.insert(
                         this.row_to_cache_key(new_row)?,
                         Change::Insert(new_row.into_owned_row()),
                     );
                 }
                 Record::Delete { old_row } => {
                     let part_key = this.get_partition_key(old_row).into();
-                    let part_delta = deltas.entry(part_key).or_insert(PartitionDelta::new());
+                    let part_delta = deltas.entry(part_key).or_default();
                     part_delta.insert(this.row_to_cache_key(old_row)?, Change::Delete);
                 }
                 Record::Update { old_row, new_row } => {
@@ -325,27 +325,23 @@ impl<S: StateStore> OverWindowExecutor<S> {
                     let new_state_key = this.row_to_cache_key(new_row)?;
                     if old_part_key == new_part_key && old_state_key == new_state_key {
                         // not a key-change update
-                        let part_delta =
-                            deltas.entry(old_part_key).or_insert(PartitionDelta::new());
+                        let part_delta = deltas.entry(old_part_key).or_default();
                         part_delta.insert(old_state_key, Change::Insert(new_row.into_owned_row()));
                     } else if old_part_key == new_part_key {
                         // order-change update, split into delete + insert, will be merged after
                         // building changes
                         key_change_updated_pks.insert(this.get_input_pk(old_row));
-                        let part_delta =
-                            deltas.entry(old_part_key).or_insert(PartitionDelta::new());
+                        let part_delta = deltas.entry(old_part_key).or_default();
                         part_delta.insert(old_state_key, Change::Delete);
                         part_delta.insert(new_state_key, Change::Insert(new_row.into_owned_row()));
                     } else {
                         // partition-change update, split into delete + insert
                         // NOTE(rc): Since we append partition key to logical pk, we can't merge the
                         // delete + insert back to update later.
                         // TODO: IMO this behavior is problematic. Deep discussion is needed.
-                        let old_part_delta =
-                            deltas.entry(old_part_key).or_insert(PartitionDelta::new());
+                        let old_part_delta = deltas.entry(old_part_key).or_default();
                         old_part_delta.insert(old_state_key, Change::Delete);
-                        let new_part_delta =
-                            deltas.entry(new_part_key).or_insert(PartitionDelta::new());
+                        let new_part_delta = deltas.entry(new_part_key).or_default();
                         new_part_delta
                             .insert(new_state_key, Change::Insert(new_row.into_owned_row()));
                     }

diff --git a/src/stream/src/executor/over_window/over_partition.rs b/src/stream/src/executor/over_window/over_partition.rs
@@ -1083,7 +1083,7 @@ mod find_affected_ranges_tests {
     ) {
         result
             .into_iter()
-            .zip_eq(expected.into_iter())
+            .zip_eq(expected)
             .for_each(|(result, expected)| {
                 assert_eq!(
                     result.0.as_normal_expect().pk.0,

diff --git a/src/stream/src/executor/source/source_executor.rs b/src/stream/src/executor/source/source_executor.rs
@@ -257,7 +257,7 @@ impl<S: StateStore> SourceExecutor<S> {
         // fetch the newest offset, either it's in cache (before barrier)
         // or in state table (just after barrier)
         let target_state = if core.state_cache.is_empty() {
-            for ele in split_info.iter_mut() {
+            for ele in &mut *split_info {
                 if let Some(recover_state) = core
                     .split_state_store
                     .try_recover_from_state_store(ele)