Merge pull request #52 from datafusion-contrib/ntran/cu

Fix compile and clippy warnings

Author: NGA-TRAN

src/planning.rs

@@ -460,7 +460,7 @@ pub async fn distribute_stages(
 
         // all stages to workers
         let (task_datas, final_addrs) =
-            assign_to_workers(query_id, &stages, workers.values().collect(), codec)?;
+            assign_to_workers(query_id, stages, workers.values().collect(), codec)?;
 
         // we retry this a few times to ensure that the workers are ready
         // and can accept the stages

src/proxy_service.rs

@@ -79,21 +79,25 @@ impl DDProxyHandler {
         }
     }
 
-    pub fn create_flight_info_response(&self, query_plan: QueryPlan) -> Result<FlightInfo, Status> {
+    pub fn create_flight_info_response(
+        &self,
+        query_plan: QueryPlan,
+    ) -> Result<FlightInfo, Box<Status>> {
         let mut flight_info = FlightInfo::new()
             .try_with_schema(&query_plan.schema)
-            .map_err(|e| Status::internal(format!("Could not create flight info {e:?}")))?;
+            .map_err(|e| {
+                Box::new(Status::internal(format!(
+                    "Could not create flight info {e:?}"
+                )))
+            })?;
 
         let ticket_data = TicketStatementData {
             query_id: query_plan.query_id,
             stage_id: query_plan.final_stage_id,
             stage_addrs: Some(query_plan.worker_addresses.into()),
-            schema: Some(
-                query_plan
-                    .schema
-                    .try_into()
-                    .map_err(|e| Status::internal(format!("Could not convert schema {e:?}")))?,
-            ),
+            schema: Some(query_plan.schema.try_into().map_err(|e| {
+                Box::new(Status::internal(format!("Could not convert schema {e:?}")))
+            })?),
         };
 
         let ticket = Ticket::new(
@@ -153,18 +157,18 @@ impl DDProxyHandler {
         &self,
         addrs: &Addrs,
         expected_stage_id: u64,
-    ) -> Result<(), Status> {
+    ) -> Result<(), Box<Status>> {
         if addrs.len() != 1 {
-            return Err(Status::internal(format!(
+            return Err(Box::new(Status::internal(format!(
                 "Expected exactly one stage in addrs, got {}",
                 addrs.len()
-            )));
+            ))));
         }
         if !addrs.contains_key(&expected_stage_id) {
-            return Err(Status::internal(format!(
+            return Err(Box::new(Status::internal(format!(
                 "No addresses found for stage_id {} in addrs",
                 expected_stage_id
-            )));
+            ))));
         }
         Ok(())
     }
@@ -257,7 +261,8 @@ impl FlightSqlHandler for DDProxyHandler {
         trace!("calculated addrs: {:?}", addrs);
 
         // Validate that addrs contains exactly one stage
-        self.validate_single_stage_addrs(&addrs, tsd.stage_id)?;
+        self.validate_single_stage_addrs(&addrs, tsd.stage_id)
+            .map_err(|e| *e)?;
 
         let stage_partition_addrs = addrs.get(&tsd.stage_id).ok_or_else(|| {
             Status::internal(format!(
@@ -357,6 +362,7 @@ impl DDProxyService {
             handler: self.handler.clone(),
         };
 
+        #[allow(clippy::result_large_err)]
         fn intercept(req: Request<()>) -> Result<Request<()>, Status> {
             println!("Intercepting request: {:?}", req);
             debug!("Intercepting request: {:?}", req);

src/query_planner.rs

@@ -10,7 +10,6 @@ use datafusion::{
 
 use datafusion_proto::physical_plan::{DefaultPhysicalExtensionCodec, PhysicalExtensionCodec};
 use datafusion_substrait::{logical_plan::consumer::from_substrait_plan, substrait::proto::Plan};
-use insta::assert_snapshot;
 use tokio_stream::StreamExt;
 
 use crate::{
@@ -288,10 +287,12 @@ impl QueryPlanner {
     }
 }
 
-pub mod tests {
+#[cfg(test)]
+mod tests {
     use super::*;
     use arrow::datatypes::{DataType, Field, Schema};
     use datafusion::physical_plan::displayable;
+    use insta::assert_snapshot;
     use std::io::BufReader;
     use std::{fs::File, path::Path};
 

src/worker_service.rs

@@ -99,6 +99,7 @@ struct DDWorkerHandler {
     done: Arc<Mutex<bool>>,
 
     /// Optional customizer for our context and proto serde
+    #[allow(dead_code)]
     pub customizer: Option<Arc<dyn Customizer>>,
 
     codec: Arc<dyn PhysicalExtensionCodec>,
@@ -165,8 +166,7 @@ impl DDWorkerHandler {
             customizer
                 .clone()
                 .map(|c| c as Arc<dyn PhysicalExtensionCodec>)
-                .or(Some(Arc::new(DefaultPhysicalExtensionCodec {})))
-                .unwrap(),
+                .unwrap_or(Arc::new(DefaultPhysicalExtensionCodec {})),
         ));
 
         Self {
@@ -502,6 +502,7 @@ impl DDWorkerHandler {
 
         Ok(Box::pin(out_stream))
     }
+    #[allow(clippy::result_large_err)]
     fn do_action_get_host(&self) -> Result<Response<crate::flight::DoActionStream>, Status> {
         let addr = self.addr.clone();
         let name = self.name.clone();