format files

src/bin/server.rs

@@ -1,10 +1,9 @@
-use tonic::transport::Server;
 use scheduler2::composable_database::scheduler_api_server::SchedulerApiServer;
 use scheduler2::server::SchedulerService;
+use tonic::transport::Server;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-
     let addr = "0.0.0.0:15721".parse().unwrap();
 
     let catalog_path = concat!(env!("CARGO_MANIFEST_DIR"), "/test_files/");

src/frontend.rs

@@ -48,6 +48,7 @@ use crate::composable_database::scheduler_api_client::SchedulerApiClient;
 use crate::composable_database::QueryJobStatusArgs;
 use crate::composable_database::QueryStatus;
 use crate::composable_database::QueryStatus::InProgress;
+use crate::intermediate_results::{get_results, TaskKey};
 use crate::mock_catalog::load_catalog;
 use crate::mock_optimizer::Optimizer;
 use crate::parser::ExecutionPlanParser;
@@ -56,7 +57,6 @@ use datafusion::logical_expr::LogicalPlan;
 use datafusion::prelude::SessionContext;
 use serde::{Deserialize, Serialize};
 use tonic::Request;
-use crate::intermediate_results::{get_results, TaskKey};
 
 #[derive(Clone, Serialize, Deserialize)]
 pub struct JobInfo {
@@ -329,10 +329,15 @@ impl MockFrontend {
                     //         }
                     //     };
 
-                    let results =  get_results(&TaskKey{stage_id: status.stage_id, query_id: status.query_id}).await
-                        .expect("api.rs: query is done but no results in table");
+                    let results = get_results(&TaskKey {
+                        stage_id: status.stage_id,
+                        query_id: status.query_id,
+                    })
+                    .await
+                    .expect("api.rs: query is done but no results in table");
 
-                    let flattened_results: Vec<RecordBatch> = results.into_iter().flat_map(|r| r.into_iter()).collect();
+                    let flattened_results: Vec<RecordBatch> =
+                        results.into_iter().flat_map(|r| r.into_iter()).collect();
 
                     let updated_job_info = JobInfo {
                         query_id,

src/queue.rs

@@ -5,11 +5,11 @@ use crate::task::{
     TaskStatus::{self, *},
 };
 use crate::SchedulerError;
+use datafusion_proto::bytes::physical_plan_to_bytes;
 use std::collections::{BTreeSet, HashMap};
 use std::hash::{Hash, Hasher};
 use std::sync::Arc;
 use std::time::{Duration, SystemTime};
-use datafusion_proto::bytes::physical_plan_to_bytes;
 use tokio::sync::{Mutex, Notify};
 
 // Must implement here since generated TaskId does not derive Hash.

src/server.rs

@@ -30,11 +30,7 @@ pub struct SchedulerService {
 
 impl fmt::Debug for SchedulerService {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(
-            f,
-            "SchedulerService {{ queue: {:?} }}",
-            self.queue,
-        )
+        write!(f, "SchedulerService {{ queue: {:?} }}", self.queue,)
     }
 }
 
@@ -99,8 +95,11 @@ impl SchedulerApi for SchedulerService {
         // Build a query graph, store in query table, enqueue new tasks.
         let qid = self.next_query_id();
         let query = QueryGraph::new(qid, plan).await;
-        self.queue.lock().await.add_query(qid, Arc::new(Mutex::new(query))).await;
-
+        self.queue
+            .lock()
+            .await
+            .add_query(qid, Arc::new(Mutex::new(query)))
+            .await;
 
         let response = ScheduleQueryRet { query_id: qid };
         Ok(Response::new(response))
@@ -128,7 +127,7 @@ impl SchedulerApi for SchedulerService {
             return Ok(Response::new(QueryJobStatusRet {
                 query_status: QueryStatus::Done.into(),
                 stage_id: stage_id,
-                query_id: query_id
+                query_id: query_id,
             }));
             // ****************** END CHANGES FROM INTEGRATION TESTING****************//
         }