chore: upgrade rust toolchain to latest nightly (#2049)

* chore: upgrade rust toolchain to latest nightly

* rebase develop

* update rust toolchain in ci

src/frontend/src/catalog.rs

@@ -407,18 +407,26 @@ impl CatalogManager for FrontendCatalogManager {
         }
 
         let key = TableNameKey::new(catalog, schema, table_name);
-        let Some(table_name_value) = self.table_metadata_manager
+        let Some(table_name_value) = self
+            .table_metadata_manager
             .table_name_manager()
             .get(key)
             .await
-            .context(TableMetadataManagerSnafu)? else { return Ok(None) };
+            .context(TableMetadataManagerSnafu)?
+        else {
+            return Ok(None);
+        };
         let table_id = table_name_value.table_id();
 
-        let Some(table_info_value) = self.table_metadata_manager
+        let Some(table_info_value) = self
+            .table_metadata_manager
             .table_info_manager()
             .get(table_id)
             .await
-            .context(TableMetadataManagerSnafu)? else { return Ok(None) };
+            .context(TableMetadataManagerSnafu)?
+        else {
+            return Ok(None);
+        };
         let table_info = Arc::new(
             table_info_value
                 .table_info

src/frontend/src/expr_factory.rs

@@ -254,7 +254,7 @@ pub(crate) fn column_schemas_to_defs(
 
     column_schemas
         .iter()
-        .zip(column_datatypes.into_iter())
+        .zip(column_datatypes)
         .map(|(schema, datatype)| {
             Ok(api::v1::ColumnDef {
                 name: schema.name.clone(),
@@ -341,7 +341,9 @@ mod tests {
             .pop()
             .unwrap();
 
-        let Statement::CreateTable(create_table) = stmt else { unreachable!() };
+        let Statement::CreateTable(create_table) = stmt else {
+            unreachable!()
+        };
         let expr = create_to_expr(&create_table, QueryContext::arc()).unwrap();
         assert_eq!("3days", expr.table_options.get("ttl").unwrap());
         assert_eq!(

src/frontend/src/heartbeat/handler/invalidate_table_cache.rs

@@ -44,7 +44,9 @@ impl HeartbeatResponseHandler for InvalidateTableCacheHandler {
 
     async fn handle(&self, ctx: &mut HeartbeatResponseHandlerContext) -> MetaResult<HandleControl> {
         // TODO(weny): considers introducing a macro
-        let Some((meta, Instruction::InvalidateTableCache(table_ident))) = ctx.incoming_message.take() else {
+        let Some((meta, Instruction::InvalidateTableCache(table_ident))) =
+            ctx.incoming_message.take()
+        else {
             unreachable!("InvalidateTableCacheHandler: should be guarded by 'is_acceptable'");
         };
 

src/frontend/src/instance.rs

@@ -193,7 +193,7 @@ impl Instance {
         plugins.insert::<StatementExecutorRef>(statement_executor.clone());
 
         let handlers_executor = HandlerGroupExecutor::new(vec![
-            Arc::new(ParseMailboxMessageHandler::default()),
+            Arc::new(ParseMailboxMessageHandler),
             Arc::new(InvalidateTableCacheHandler::new(
                 meta_backend,
                 partition_manager,

src/frontend/src/instance/distributed.rs

@@ -300,7 +300,9 @@ impl DistInstance {
             .iter()
             .filter(|route| {
                 route.region_routes.iter().any(|r| {
-                    let Some(n) = region_number else { return true; };
+                    let Some(n) = region_number else {
+                        return true;
+                    };
                     n == r.region.id.region_number()
                 })
             })

src/frontend/src/instance/prom_store.rs

@@ -67,7 +67,9 @@ fn negotiate_response_type(accepted_response_types: &[i32]) -> ServerResult<Resp
 }
 
 async fn to_query_result(table_name: &str, output: Output) -> ServerResult<QueryResult> {
-    let Output::Stream(stream) = output else { unreachable!() };
+    let Output::Stream(stream) = output else {
+        unreachable!()
+    };
     let recordbatches = RecordBatches::try_collect(stream)
         .await
         .context(error::CollectRecordbatchSnafu)?;

src/frontend/src/statement/copy_table_from.rs

@@ -315,7 +315,7 @@ impl StatementExecutor {
                 let columns_values = fields
                     .iter()
                     .cloned()
-                    .zip(vectors.into_iter())
+                    .zip(vectors)
                     .collect::<HashMap<_, _>>();
 
                 pending.push(table.insert(InsertRequest {

src/frontend/src/table.rs

@@ -218,7 +218,9 @@ impl Table for DistTable {
             .await
             .map_err(BoxedError::new)
             .context(TableOperationSnafu)?;
-        let Output::AffectedRows(rows) = output else { unreachable!() };
+        let Output::AffectedRows(rows) = output else {
+            unreachable!()
+        };
         Ok(rows)
     }
 }

src/frontend/src/table/delete.rs

@@ -30,7 +30,7 @@ impl DistTable {
         let regions = requests.iter().map(|x| x.region_number).collect::<Vec<_>>();
         let instances = self.find_datanode_instances(&regions).await?;
 
-        let results = future::try_join_all(instances.into_iter().zip(requests.into_iter()).map(
+        let results = future::try_join_all(instances.into_iter().zip(requests).map(
             |(instance, request)| {
                 common_runtime::spawn_write(async move {
                     instance

src/frontend/src/table/scan.rs

@@ -63,7 +63,9 @@ impl DatanodeInstance {
             .logical_plan(substrait_plan.to_vec(), None)
             .await
             .context(error::RequestDatanodeSnafu)?;
-        let Output::RecordBatches(record_batches) = result else { unreachable!() };
+        let Output::RecordBatches(record_batches) = result else {
+            unreachable!()
+        };
         Ok(record_batches)
     }