cargo fix and fmt

Author: robtandy

src/flight_service/do_get.rs

@@ -1,26 +1,19 @@
-use crate::composed_extension_codec::ComposedPhysicalExtensionCodec;
 use crate::errors::datafusion_error_to_tonic_status;
 use crate::flight_service::service::ArrowFlightEndpoint;
 use crate::plan::DistributedCodec;
-use crate::stage::{stage_from_proto, ExecutionStage, ExecutionStageProto};
+use crate::stage::{stage_from_proto, ExecutionStageProto};
 use arrow_flight::encode::FlightDataEncoderBuilder;
 use arrow_flight::error::FlightError;
 use arrow_flight::flight_service_server::FlightService;
 use arrow_flight::Ticket;
-use datafusion::error::DataFusionError;
 use datafusion::execution::SessionStateBuilder;
 use datafusion::optimizer::OptimizerConfig;
-use datafusion::physical_expr::{Partitioning, PhysicalExpr};
 use datafusion::physical_plan::ExecutionPlan;
-use datafusion_proto::physical_plan::from_proto::parse_physical_exprs;
-use datafusion_proto::physical_plan::to_proto::serialize_physical_exprs;
-use datafusion_proto::physical_plan::{AsExecutionPlan, PhysicalExtensionCodec};
-use datafusion_proto::protobuf::{PhysicalExprNode, PhysicalPlanNode};
+use datafusion_proto::physical_plan::PhysicalExtensionCodec;
 use futures::TryStreamExt;
 use prost::Message;
 use std::sync::Arc;
 use tonic::{Request, Response, Status};
-use uuid::Uuid;
 
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct DoGet {

src/physical_optimizer.rs

@@ -1,19 +1,16 @@
 use std::sync::Arc;
 
 use datafusion::{
-    catalog::memory::DataSourceExec,
     common::{
-        internal_datafusion_err, internal_err,
-        tree_node::{Transformed, TreeNode, TreeNodeRecursion, TreeNodeRewriter, TreeNodeVisitor},
-        DataFusionError,
+        internal_datafusion_err,
+        tree_node::{Transformed, TreeNode, TreeNodeRewriter},
     },
     config::ConfigOptions,
-    datasource::physical_plan::{FileScanConfig, FileSource},
+    datasource::physical_plan::FileSource,
     error::Result,
     physical_optimizer::PhysicalOptimizerRule,
     physical_plan::{
-        displayable, execution_plan::need_data_exchange, repartition::RepartitionExec,
-        ExecutionPlan, ExecutionPlanProperties,
+        displayable, repartition::RepartitionExec, ExecutionPlan, ExecutionPlanProperties,
     },
 };
 use datafusion_proto::physical_plan::PhysicalExtensionCodec;

src/plan/arrow_flight_read.rs

@@ -1,32 +1,21 @@
 use crate::channel_manager::ChannelManager;
-use crate::composed_extension_codec::ComposedPhysicalExtensionCodec;
-use crate::errors::tonic_status_to_datafusion_error;
 use crate::flight_service::DoGet;
-use crate::plan::codec::DistributedCodec;
 use crate::stage::{ExecutionStage, ExecutionStageProto};
-use arrow_flight::decode::FlightRecordBatchStream;
-use arrow_flight::error::FlightError;
-use arrow_flight::{FlightClient, FlightData, Ticket};
+use arrow_flight::{FlightClient, Ticket};
 use datafusion::arrow::datatypes::SchemaRef;
-use datafusion::common::runtime::JoinSet;
-use datafusion::common::{internal_datafusion_err, internal_err, plan_err};
-use datafusion::error::DataFusionError;
+use datafusion::common::{internal_datafusion_err, plan_err};
 use datafusion::error::Result;
 use datafusion::execution::{RecordBatchStream, SendableRecordBatchStream, TaskContext};
 use datafusion::physical_expr::{EquivalenceProperties, Partitioning};
 use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
 use datafusion::physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties};
-use futures::channel::oneshot;
 use futures::{future, TryFutureExt, TryStreamExt};
 use prost::Message;
 use std::any::Any;
 use std::fmt::Formatter;
-use std::future::Future;
-use std::io::Bytes;
-use std::sync::{self, Arc};
+use std::sync::Arc;
 use tonic::transport::Channel;
-use tonic::{IntoRequest, Streaming};
 use url::Url;
 
 use super::combined::CombinedRecordBatchStream;

src/plan/codec.rs

@@ -1,17 +1,14 @@
 use crate::plan::arrow_flight_read::ArrowFlightReadExec;
-use datafusion::arrow::datatypes::{Schema, SchemaRef};
-use datafusion::error::DataFusionError;
+use datafusion::arrow::datatypes::Schema;
 use datafusion::execution::FunctionRegistry;
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion_proto::physical_plan::from_proto::parse_protobuf_partitioning;
 use datafusion_proto::physical_plan::to_proto::serialize_partitioning;
-use datafusion_proto::physical_plan::{DefaultPhysicalExtensionCodec, PhysicalExtensionCodec};
+use datafusion_proto::physical_plan::PhysicalExtensionCodec;
 use datafusion_proto::protobuf;
 use datafusion_proto::protobuf::proto_error;
 use prost::Message;
 use std::sync::Arc;
-use url::Url;
-use uuid::Uuid;
 
 use super::PartitionIsolatorExec;
 

src/plan/combined.rs

@@ -4,13 +4,13 @@ use std::{
     task::{Context, Poll},
 };
 
+use datafusion::error::Result;
 use datafusion::{
     arrow::{array::RecordBatch, datatypes::SchemaRef},
     common::internal_err,
     error::DataFusionError,
     execution::{RecordBatchStream, SendableRecordBatchStream},
 };
-use datafusion::{common::internal_datafusion_err, error::Result};
 use futures::Stream;
 
 pub(crate) struct CombinedRecordBatchStream {

src/plan/isolator.rs

@@ -1,7 +1,6 @@
 use std::{fmt::Formatter, sync::Arc};
 
 use datafusion::{
-    common::internal_datafusion_err,
     error::Result,
     execution::SendableRecordBatchStream,
     physical_plan::{

src/plan/mod.rs

@@ -5,4 +5,4 @@ mod isolator;
 
 pub use arrow_flight_read::ArrowFlightReadExec;
 pub use codec::DistributedCodec;
-pub use isolator::{PartitionGroup, PartitionIsolatorExec};
+pub use isolator::PartitionIsolatorExec;

src/stage/display.rs

@@ -10,23 +10,11 @@
 ///
 /// In the meantime, we can make a dummy ExecutionPlan that will let us render
 /// the Stage tree.
-use std::{
-    fmt::{Display, Formatter, Write},
-    sync::Arc,
-};
+use std::fmt::Write;
 
 use datafusion::{
-    common::{
-        internal_datafusion_err,
-        tree_node::{Transformed, TreeNode, TreeNodeRecursion, TreeNodeRewriter, TreeNodeVisitor},
-    },
     error::Result,
-    physical_expr::EquivalenceProperties,
-    physical_plan::{
-        displayable,
-        execution_plan::{Boundedness, EmissionType},
-        DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PlanProperties,
-    },
+    physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan},
 };
 
 use crate::{

src/stage/proto.rs

@@ -1,13 +1,13 @@
 use std::sync::Arc;
 
 use datafusion::{
-    common::{internal_datafusion_err, internal_err},
+    common::internal_datafusion_err,
     error::{DataFusionError, Result},
     execution::{runtime_env::RuntimeEnv, FunctionRegistry},
     physical_plan::ExecutionPlan,
 };
 use datafusion_proto::{
-    physical_plan::{AsExecutionPlan, DefaultPhysicalExtensionCodec, PhysicalExtensionCodec},
+    physical_plan::{AsExecutionPlan, PhysicalExtensionCodec},
     protobuf::PhysicalPlanNode,
 };
 use prost::Message;

src/stage/stage.rs

@@ -1,18 +1,15 @@
-use std::any::Any;
-use std::cell::RefCell;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 
 use datafusion::common::internal_err;
 use datafusion::error::{DataFusionError, Result};
 use datafusion::execution::TaskContext;
-use datafusion::physical_plan::{displayable, DisplayAs, DisplayFormatType, ExecutionPlan};
+use datafusion::physical_plan::{displayable, ExecutionPlan};
 use datafusion::prelude::SessionContext;
 use datafusion_proto::physical_plan::PhysicalExtensionCodec;
 
 use itertools::Itertools;
 use rand::Rng;
-use tokio::sync::RwLock;
 use url::Url;
 
 use crate::task::ExecutionTask;
@@ -150,7 +147,7 @@ impl ExecutionStage {
     }
 
     pub fn try_assign(
-        mut self,
+        self,
         channel_manager: impl TryInto<ChannelManager, Error = DataFusionError>,
     ) -> Result<Self> {
         let urls: Vec<Url> = channel_manager.try_into()?.get_urls()?;