chore(rustfmt): format let-chains and let-else

src/common/src/cache.rs

@@ -757,7 +757,7 @@ impl<K: LruKey, T: LruValue> LruCache<K, T> {
             shard.release(handle)
         };
         // do not deallocate data with holding mutex.
-        if let Some((key,value)) = data && let Some(listener) = &self.listener {
+        if let Some((key, value)) = data && let Some(listener) = &self.listener {
             listener.on_release(key, value);
         }
     }
@@ -819,7 +819,7 @@ impl<K: LruKey, T: LruValue> LruCache<K, T> {
             shard.erase(hash, key)
         };
         // do not deallocate data with holding mutex.
-        if let Some((key,value)) = data && let Some(listener) = &self.listener {
+        if let Some((key, value)) = data && let Some(listener) = &self.listener {
             listener.on_release(key, value);
         }
     }

src/common/src/hash/consistent_hash/vnode.rs

@@ -113,11 +113,12 @@ impl VirtualNode {
     // be the one that contains RowId, and use a special method to skip the calculation of Hash
     // and directly extract the `VirtualNode` from `RowId`.
     pub fn compute_chunk(data_chunk: &DataChunk, keys: &[usize]) -> Vec<VirtualNode> {
-        if let Ok(idx) = keys.iter().exactly_one() &&
-            let ArrayImpl::Serial(serial_array) = data_chunk.column_at(*idx).array_ref() {
-
-            return serial_array.iter()
-                .map(|serial|extract_vnode_id_from_row_id(serial.unwrap().as_row_id()))
+        if let Ok(idx) = keys.iter().exactly_one()
+            && let ArrayImpl::Serial(serial_array) = data_chunk.column_at(*idx).array_ref()
+        {
+            return serial_array
+                .iter()
+                .map(|serial| extract_vnode_id_from_row_id(serial.unwrap().as_row_id()))
                 .collect();
         }
 

src/common/src/types/interval.rs

@@ -1324,7 +1324,7 @@ impl Interval {
     fn parse_postgres(s: &str) -> Result<Self> {
         use DateTimeField::*;
         let mut tokens = parse_interval(s)?;
-        if tokens.len()%2!=0 && let Some(TimeStrToken::Num(_)) = tokens.last() {
+        if tokens.len() % 2 != 0 && let Some(TimeStrToken::Num(_)) = tokens.last() {
             tokens.push(TimeStrToken::TimeUnit(DateTimeField::Second));
         }
         if tokens.len() % 2 != 0 {
@@ -1356,22 +1356,30 @@ impl Interval {
                         }
                     })()
                     .and_then(|rhs| result.checked_add(&rhs))
-                    .ok_or_else(|| ErrorCode::InvalidInputSyntax(format!("Invalid interval {}.", s)))?;
+                    .ok_or_else(|| {
+                        ErrorCode::InvalidInputSyntax(format!("Invalid interval {}.", s))
+                    })?;
                 }
                 (TimeStrToken::Second(second), TimeStrToken::TimeUnit(interval_unit)) => {
                     result = match interval_unit {
                         Second => {
-                            // If unsatisfied precision is passed as input, we should not return None (Error).
-                            let usecs = (second.into_inner() * (USECS_PER_SEC as f64)).round_ties_even() as i64;
+                            // If unsatisfied precision is passed as input, we should not return
+                            // None (Error).
+                            let usecs = (second.into_inner() * (USECS_PER_SEC as f64))
+                                .round_ties_even() as i64;
                             Some(Interval::from_month_day_usec(0, 0, usecs))
                         }
                         _ => None,
                     }
                     .and_then(|rhs| result.checked_add(&rhs))
-                    .ok_or_else(|| ErrorCode::InvalidInputSyntax(format!("Invalid interval {}.", s)))?;
+                    .ok_or_else(|| {
+                        ErrorCode::InvalidInputSyntax(format!("Invalid interval {}.", s))
+                    })?;
                 }
                 _ => {
-                    return Err(ErrorCode::InvalidInputSyntax(format!("Invalid interval {}.", &s)).into());
+                    return Err(
+                        ErrorCode::InvalidInputSyntax(format!("Invalid interval {}.", &s)).into(),
+                    );
                 }
             }
         }

src/connector/src/parser/debezium/mongo_json_parser.rs

@@ -77,32 +77,30 @@ fn parse_bson_value(
             )),
             _ => Err(RwError::from(ProtocolError(format!(
                 "Can not convert bson {:?} to {:?}",
-                id_field,
-                id_type
+                id_field, id_type
             ))))?,
         },
         DataType::Int32 => {
-            if let serde_json::Value::Object(ref obj) = id_field  && obj.contains_key("$numberInt"){
-                    let int_str = obj["$numberInt"].as_str().unwrap_or_default();
-                    Some(ScalarImpl::Int32(int_str.parse().unwrap_or_default()))
-                } else {
-                    Err(RwError::from(ProtocolError(format!(
-                        "Can not convert bson {:?} to {:?}",
-                        id_field,
-                        id_type
-                    ))))?
+            if let serde_json::Value::Object(ref obj) = id_field && obj.contains_key("$numberInt") {
+                let int_str = obj["$numberInt"].as_str().unwrap_or_default();
+                Some(ScalarImpl::Int32(int_str.parse().unwrap_or_default()))
+            } else {
+                Err(RwError::from(ProtocolError(format!(
+                    "Can not convert bson {:?} to {:?}",
+                    id_field, id_type
+                ))))?
             }
         }
         DataType::Int64 => {
-            if let serde_json::Value::Object(ref obj) = id_field && obj.contains_key("$numberLong") {
-                    let int_str = obj["$numberLong"].as_str().unwrap_or_default();
-                    Some(ScalarImpl::Int64(int_str.parse().unwrap_or_default()))
-                } else {
-                    Err(RwError::from(ProtocolError(format!(
-                        "Can not convert bson {:?} to {:?}",
-                        id_field,
-                        id_type
-                    ))))?
+            if let serde_json::Value::Object(ref obj) = id_field && obj.contains_key("$numberLong")
+            {
+                let int_str = obj["$numberLong"].as_str().unwrap_or_default();
+                Some(ScalarImpl::Int64(int_str.parse().unwrap_or_default()))
+            } else {
+                Err(RwError::from(ProtocolError(format!(
+                    "Can not convert bson {:?} to {:?}",
+                    id_field, id_type
+                ))))?
             }
         }
         _ => unreachable!("DebeziumMongoJsonParser::new must ensure _id column datatypes."),

src/connector/src/parser/json_parser.rs

@@ -107,8 +107,8 @@ impl JsonParser {
         let value: BorrowedValue<'_> = simd_json::to_borrowed_value(&mut payload_mut)
             .map_err(|e| RwError::from(ProtocolError(e.to_string())))?;
 
-        if let BorrowedValue::Array(ref objects) = value  && matches!(op, Op::Insert) {
-             at_least_one_ok(
+        if let BorrowedValue::Array(ref objects) = value && matches!(op, Op::Insert) {
+            at_least_one_ok(
                 objects
                     .iter()
                     .map(|obj| Self::parse_single_value(obj, &mut writer))
@@ -119,13 +119,10 @@ impl JsonParser {
                 simd_json_parse_value(
                     &SourceFormat::Json,
                     &desc.data_type,
-                    json_object_smart_get_value(&value,desc.name.as_str().into())
+                    json_object_smart_get_value(&value, desc.name.as_str().into()),
                 )
                 .map_err(|e| {
-                    tracing::error!(
-                        "failed to process value: {}",
-                        e
-                    );
+                    tracing::error!("failed to process value: {}", e);
                     e.into()
                 })
             };

src/connector/src/source/base.rs

@@ -116,10 +116,13 @@ impl SourceContext {
             return Err(e);
         }
         let mut err_str = e.inner().to_string();
-        if let Some(suppressor) = &self.error_suppressor &&
-            suppressor.lock().suppress_error(&err_str)
+        if let Some(suppressor) = &self.error_suppressor
+            && suppressor.lock().suppress_error(&err_str)
         {
-            err_str = format!("error msg suppressed (due to per-actor error limit: {})", suppressor.lock().max());
+            err_str = format!(
+                "error msg suppressed (due to per-actor error limit: {})",
+                suppressor.lock().max()
+            );
         }
         self.metrics
             .user_source_error_count

src/connector/src/source/datagen/source/reader.rs

@@ -280,11 +280,12 @@ fn generator_from_data_type(
         }
         _ => {
             let kind_key = format!("fields.{}.kind", name);
-            if let Some(kind) = fields_option_map.get(&kind_key) && kind.as_str() == SEQUENCE_FIELD_KIND {
+            if let Some(kind) = fields_option_map.get(&kind_key)
+                && kind.as_str() == SEQUENCE_FIELD_KIND
+            {
                 let start_key = format!("fields.{}.start", name);
                 let end_key = format!("fields.{}.end", name);
-                let start_value =
-                    fields_option_map.get(&start_key).map(|s| s.to_string());
+                let start_value = fields_option_map.get(&start_key).map(|s| s.to_string());
                 let end_value = fields_option_map.get(&end_key).map(|s| s.to_string());
                 FieldGeneratorImpl::with_number_sequence(
                     data_type,
@@ -299,12 +300,7 @@ fn generator_from_data_type(
                 let max_key = format!("fields.{}.max", name);
                 let min_value = fields_option_map.get(&min_key).map(|s| s.to_string());
                 let max_value = fields_option_map.get(&max_key).map(|s| s.to_string());
-                FieldGeneratorImpl::with_number_random(
-                    data_type,
-                    min_value,
-                    max_value,
-                    random_seed
-                )
+                FieldGeneratorImpl::with_number_random(data_type, min_value, max_value, random_seed)
             }
         }
     }

src/connector/src/source/google_pubsub/source/reader.rs

@@ -100,8 +100,7 @@ impl PubsubSplitReader {
             yield chunk;
 
             // Stop if we've approached the stop_offset
-            if let Some(stop_offset) = self.stop_offset
-            && latest_offset >= stop_offset {
+            if let Some(stop_offset) = self.stop_offset && latest_offset >= stop_offset {
                 return Ok(());
             }
         }

src/connector/src/source/kafka/source/reader.rs

@@ -152,7 +152,7 @@ impl KafkaSplitReader {
     #[try_stream(boxed, ok = Vec<SourceMessage>, error = anyhow::Error)]
     pub async fn into_data_stream(self) {
         if let Some(stop_offset) = self.stop_offset {
-            if let Some(start_offset) = self.start_offset && (start_offset+1) >= stop_offset {
+            if let Some(start_offset) = self.start_offset && (start_offset + 1) >= stop_offset {
                 yield Vec::new();
                 return Ok(());
             } else if stop_offset == 0 {

src/expr/macro/src/parse.rs

@@ -75,9 +75,15 @@ impl UserFunctionAttr {
 /// Check if the last argument is `&mut dyn Write`.
 fn last_arg_is_write(item: &syn::ItemFn) -> bool {
     let Some(syn::FnArg::Typed(arg)) = item.sig.inputs.last() else { return false };
-    let syn::Type::Reference(syn::TypeReference { elem, .. }) = arg.ty.as_ref() else { return false };
-    let syn::Type::TraitObject(syn::TypeTraitObject { bounds, .. }) = elem.as_ref() else { return false };
-    let Some(syn::TypeParamBound::Trait(syn::TraitBound { path, .. })) = bounds.first() else { return false };
+    let syn::Type::Reference(syn::TypeReference { elem, .. }) = arg.ty.as_ref() else {
+        return false;
+    };
+    let syn::Type::TraitObject(syn::TypeTraitObject { bounds, .. }) = elem.as_ref() else {
+        return false;
+    };
+    let Some(syn::TypeParamBound::Trait(syn::TraitBound { path, .. })) = bounds.first() else {
+        return false;
+    };
     path.segments.last().map_or(false, |s| s.ident == "Write")
 }
 

src/expr/src/expr/build.rs

@@ -242,7 +242,9 @@ fn lexer(input: &str) -> Vec<Token> {
             _ => {
                 let mut literal = String::new();
                 literal.push(c);
-                while let Some(&c) = chars.peek() && !matches!(c, '(' | ')' | ':' | ' ' | '\t' | '\r' | '\n') {
+                while let Some(&c) = chars.peek()
+                    && !matches!(c, '(' | ')' | ':' | ' ' | '\t' | '\r' | '\n')
+                {
                     literal.push(chars.next().unwrap());
                 }
                 Token::Literal(literal)

src/expr/src/expr/expr_some_all.rs

@@ -88,9 +88,7 @@ impl Expression for SomeAllExpression {
         let mut num_array = Vec::with_capacity(data_chunk.capacity());
 
         let arr_right_inner = arr_right.as_list();
-        let DataType::List { datatype } = arr_right_inner.data_type() else {
-            unreachable!()
-        };
+        let DataType::List { datatype } = arr_right_inner.data_type() else { unreachable!() };
         let capacity = arr_right_inner
             .iter()
             .flatten()
@@ -219,9 +217,11 @@ impl<'a> TryFrom<&'a ExprNode> for SomeAllExpression {
         let left_expr = build_from_prost(&inner_children[0])?;
         let right_expr = build_from_prost(&inner_children[1])?;
 
-        let DataType::List { datatype: right_expr_return_type } = right_expr.return_type() else {
-        bail!("Expect Array Type");
-    };
+        let DataType::List {
+            datatype: right_expr_return_type,
+        } = right_expr.return_type() else {
+            bail!("Expect Array Type");
+        };
 
         let eval_func = {
             let left_expr_input_ref = ExprNode {

src/expr/src/vector_op/overlay.rs

@@ -108,7 +108,9 @@ pub fn overlay_for(
 
     let Ok(count) = count.try_into() else {
         // For negative `count`, which is rare in practice, we hand over to `substr`
-        let start_right = start.checked_add(count).ok_or(ExprError::NumericOutOfRange)?;
+        let start_right = start
+            .checked_add(count)
+            .ok_or(ExprError::NumericOutOfRange)?;
         return super::substr::substr_start(s, start_right, writer);
     };
 

src/frontend/planner_test/src/lib.rs

@@ -202,7 +202,8 @@ impl TestCaseResult {
         if original_test_case.planner_error.is_none() && let Some(ref err) = self.planner_error {
             return Err(anyhow!("unexpected planner error: {}", err));
         }
-        if original_test_case.optimizer_error.is_none() && let Some(ref err) = self.optimizer_error {
+        if original_test_case.optimizer_error.is_none() && let Some(ref err) = self.optimizer_error
+        {
             return Err(anyhow!("unexpected optimizer error: {}", err));
         }
 

src/frontend/src/binder/expr/mod.rs

@@ -493,7 +493,9 @@ impl Binder {
     }
 
     pub fn bind_cast_inner(&mut self, expr: Expr, data_type: DataType) -> Result<ExprImpl> {
-        if let Expr::Array(Array {elem: ref expr, ..}) = expr && matches!(&data_type, DataType::List{ .. } ) {
+        if let Expr::Array(Array { elem: ref expr, .. }) = expr
+            && matches!(&data_type, DataType::List { .. })
+        {
             return self.bind_array_cast(expr.clone(), data_type);
         }
         let lhs = self.bind_expr_inner(expr)?;

src/frontend/src/binder/expr/value.rs

@@ -241,7 +241,7 @@ fn unescape_c_style(s: &str) -> Result<String> {
         for _ in 0..len {
             if let Some(c) = chars.peek() && c.is_ascii_hexdigit() {
                 unicode_seq.push(chars.next().unwrap());
-            }else{
+            } else {
                 break;
             }
         }
@@ -285,7 +285,7 @@ fn unescape_c_style(s: &str) -> Result<String> {
         for _ in 0..2 {
             if let Some(c) = chars.peek() && matches!(*c, '0'..='7') {
                 unicode_seq.push(chars.next().unwrap());
-            }else{
+            } else {
                 break;
             }
         }

src/frontend/src/binder/query.rs

@@ -236,9 +236,17 @@ impl Binder {
     ) -> Result<ColumnOrder> {
         let order_type = OrderType::from_bools(asc, nulls_first);
         let column_index = match expr {
-            Expr::Identifier(name) if let Some(index) = name_to_index.get(&name.real_value()) => match *index != usize::MAX {
-                true => *index,
-                false => return Err(ErrorCode::BindError(format!("ORDER BY \"{}\" is ambiguous", name.real_value())).into()),
+            Expr::Identifier(name) if let Some(index) = name_to_index.get(&name.real_value()) => {
+                match *index != usize::MAX {
+                    true => *index,
+                    false => {
+                        return Err(ErrorCode::BindError(format!(
+                            "ORDER BY \"{}\" is ambiguous",
+                            name.real_value()
+                        ))
+                        .into())
+                    }
+                }
             }
             Expr::Value(Value::Number(number)) => match number.parse::<usize>() {
                 Ok(index) if 1 <= index && index <= visible_output_num => index - 1,