Run cargo fmt

Author: gabotechs

src/plan/arrow_flight_read.rs

@@ -1,11 +1,13 @@
 use super::combined::CombinedRecordBatchStream;
+use crate::channel_manager_ext::get_channel_resolver;
 use crate::common::ComposedPhysicalExtensionCodec;
 use crate::config_extension_ext::ContextGrpcMetadata;
 use crate::errors::tonic_status_to_datafusion_error;
 use crate::flight_service::{DoGet, StageKey};
 use crate::plan::DistributedCodec;
 use crate::stage::{proto_from_stage, ExecutionStage};
 use crate::user_codec_ext::get_user_codec;
+use crate::ChannelResolver;
 use arrow_flight::decode::FlightRecordBatchStream;
 use arrow_flight::error::FlightError;
 use arrow_flight::flight_service_client::FlightServiceClient;
@@ -27,8 +29,6 @@ use std::sync::Arc;
 use tonic::metadata::MetadataMap;
 use tonic::Request;
 use url::Url;
-use crate::channel_manager_ext::get_channel_resolver;
-use crate::ChannelResolver;
 
 /// This node has two variants.
 /// 1. Pending: it acts as a placeholder for the distributed optimization step to mark it as ready.
@@ -160,7 +160,9 @@ impl ExecutionPlan for ArrowFlightReadExec {
 
         // get the channel manager and current stage from our context
         let Some(channel_resolver) = get_channel_resolver(context.session_config()) else {
-            return exec_err!("ArrowFlightReadExec requires a ChannelResolver in the session config");       
+            return exec_err!(
+                "ArrowFlightReadExec requires a ChannelResolver in the session config"
+            );
         };
 
         let stage = context

src/stage/execution_stage.rs

@@ -6,13 +6,13 @@ use datafusion::execution::TaskContext;
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::prelude::SessionContext;
 
+use crate::channel_manager_ext::get_channel_resolver;
+use crate::task::ExecutionTask;
+use crate::ChannelResolver;
 use itertools::Itertools;
 use rand::Rng;
 use url::Url;
 use uuid::Uuid;
-use crate::channel_manager_ext::get_channel_resolver;
-use crate::ChannelResolver;
-use crate::task::ExecutionTask;
 
 /// A unit of isolation for a portion of a physical execution plan
 /// that can be executed independently and across a network boundary.  
@@ -171,10 +171,7 @@ impl ExecutionStage {
         format!("Stage {:<3}{}", self.num, child_str)
     }
 
-    pub fn try_assign(
-        self,
-        channel_resolver: &impl ChannelResolver
-    ) -> Result<Self> {
+    pub fn try_assign(self, channel_resolver: &impl ChannelResolver) -> Result<Self> {
         let urls: Vec<Url> = channel_resolver.get_urls()?;
         if urls.is_empty() {
             return internal_err!("No URLs found in ChannelManager");