More clippy issues and use global context everywehere

Author: thinkharderdev

ballista/rust/scheduler/src/lib.rs

@@ -746,19 +746,19 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan> SchedulerGrpc
     ) -> std::result::Result<Response<ExecuteQueryResult>, tonic::Status> {
         if let ExecuteQueryParams {
             query: Some(query),
-            settings,
+            settings: _,
         } = request.into_inner()
         {
             // parse config
-            let mut config_builder = BallistaConfig::builder();
-            for kv_pair in &settings {
-                config_builder = config_builder.set(&kv_pair.key, &kv_pair.value);
-            }
-            let config = config_builder.build().map_err(|e| {
-                let msg = format!("Could not parse configs: {}", e);
-                error!("{}", msg);
-                tonic::Status::internal(msg)
-            })?;
+            // let mut config_builder = BallistaConfig::builder();
+            // for kv_pair in &settings {
+            //     config_builder = config_builder.set(&kv_pair.key, &kv_pair.value);
+            // }
+            // let config = config_builder.build().map_err(|e| {
+            //     let msg = format!("Could not parse configs: {}", e);
+            //     error!("{}", msg);
+            //     tonic::Status::internal(msg)
+            // })?;
 
             let plan = match query {
                 Query::LogicalPlan(message) => {
@@ -817,9 +817,9 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan> SchedulerGrpc
                     Some(self.scheduler_env.as_ref().unwrap().tx_job.clone())
                 }
             };
+            let datafusion_ctx = self.ctx.read().await.clone();
             tokio::spawn(async move {
                 // create physical plan using DataFusion
-                let datafusion_ctx = create_datafusion_context(&config);
                 macro_rules! fail_job {
                     ($code :expr) => {{
                         match $code {

ballista/rust/scheduler/src/state/mod.rs

@@ -299,7 +299,7 @@ impl<T: 'static + AsExecutionPlan> SchedulerState<T> {
             )));
         }
         let value = T::try_decode(value.as_slice())?;
-        let plan = value.try_into_physical_plan(&ctx)?;
+        let plan = value.try_into_physical_plan(ctx)?;
         // let value: PhysicalPlanNode = decode_protobuf(value)?;
         // Ok((&value).try_into()?)
         Ok(plan)
@@ -366,7 +366,7 @@ impl<T: 'static + AsExecutionPlan> SchedulerState<T> {
         ctx: &ExecutionContext,
     ) -> Result<Option<(TaskStatus, Arc<dyn ExecutionPlan>)>> {
         let tasks = self.get_all_tasks().await?;
-        self.assign_next_schedulable_task_inner(executor_id, tasks, &ctx)
+        self.assign_next_schedulable_task_inner(executor_id, tasks, ctx)
             .await
     }
 
@@ -377,7 +377,7 @@ impl<T: 'static + AsExecutionPlan> SchedulerState<T> {
         ctx: &ExecutionContext,
     ) -> Result<Option<(TaskStatus, Arc<dyn ExecutionPlan>)>> {
         let job_tasks = self.get_job_tasks(job_id).await?;
-        self.assign_next_schedulable_task_inner(executor_id, job_tasks, &ctx)
+        self.assign_next_schedulable_task_inner(executor_id, job_tasks, ctx)
             .await
     }
 
@@ -387,7 +387,7 @@ impl<T: 'static + AsExecutionPlan> SchedulerState<T> {
         tasks: HashMap<String, TaskStatus>,
         ctx: &ExecutionContext,
     ) -> Result<Option<(TaskStatus, Arc<dyn ExecutionPlan>)>> {
-        match self.get_next_schedulable_task(tasks, &ctx).await? {
+        match self.get_next_schedulable_task(tasks, ctx).await? {
             Some((status, plan)) => {
                 let mut status = status.clone();
                 status.status = Some(task_status::Status::Running(RunningTask {
@@ -413,7 +413,7 @@ impl<T: 'static + AsExecutionPlan> SchedulerState<T> {
             if status.status.is_none() {
                 let partition = status.partition_id.as_ref().unwrap();
                 let plan = self
-                    .get_stage_plan(&partition.job_id, partition.stage_id as usize, &ctx)
+                    .get_stage_plan(&partition.job_id, partition.stage_id as usize, ctx)
                     .await?;
 
                 // Let's try to resolve any unresolved shuffles we find