coralogix
diff --git a/‎ballista/core/src/serde/mod.rs
-1 b/‎ballista/core/src/serde/mod.rs
-1
diff --git a/‎ballista/scheduler/scheduler_config_spec.toml
+7 b/‎ballista/scheduler/scheduler_config_spec.toml
+7
diff --git a/‎ballista/scheduler/src/cluster/storage/etcd.rs
+7-7 b/‎ballista/scheduler/src/cluster/storage/etcd.rs
+7-7
diff --git a/‎ballista/scheduler/src/cluster/storage/memory.rs
+7-7 b/‎ballista/scheduler/src/cluster/storage/memory.rs
+7-7
diff --git a/‎ballista/scheduler/src/flight_sql.rs
+2 b/‎ballista/scheduler/src/flight_sql.rs
+2
diff --git a/‎ballista/scheduler/src/metrics/prometheus.rs
+1-1 b/‎ballista/scheduler/src/metrics/prometheus.rs
+1-1
diff --git a/‎ballista/scheduler/src/scheduler_server/grpc.rs
+3-2 b/‎ballista/scheduler/src/scheduler_server/grpc.rs
+3-2
diff --git a/‎ballista/scheduler/src/scheduler_server/mod.rs
+3-2 b/‎ballista/scheduler/src/scheduler_server/mod.rs
+3-2
@@ -32,7 +32,6 @@ use datafusion_proto::{
     logical_plan::{AsLogicalPlan, DefaultLogicalExtensionCodec, LogicalExtensionCodec},
     physical_plan::{AsExecutionPlan, PhysicalExtensionCodec},
 };
-
 use prost::Message;
 use std::fmt::Debug;
 use std::marker::PhantomData;
 
@@ -36,6 +36,13 @@ type = "ballista_scheduler::cluster::ClusterStorage"
 doc = "The configuration backend for the scheduler cluster state, possible values: etcd, memory, sled. Default: sled"
 default = "ballista_scheduler::cluster::ClusterStorage::Sled"
 
+[[param]]
+abbr = "c"
+name = "cluster_backend"
+type = "ballista_scheduler::state::backend::StateBackend"
+doc = "The configuration backend for the scheduler cluster state, possible values: etcd, memory, sled. Default: sled"
+default = "ballista_scheduler::state::backend::StateBackend::Sled"
+
 [[param]]
 abbr = "n"
 name = "namespace"
 
@@ -57,7 +57,7 @@ impl StateBackendClient for EtcdClient {
             .clone()
             .get(key, None)
             .await
-            .map_err(|e| ballista_error(&format!("etcd error {:?}", e)))?
+            .map_err(|e| ballista_error(&format!("etcd error {e:?}")))?
             .kvs()
             .get(0)
             .map(|kv| kv.value().to_owned())
@@ -76,7 +76,7 @@ impl StateBackendClient for EtcdClient {
             .clone()
             .get(prefix, Some(GetOptions::new().with_prefix()))
             .await
-            .map_err(|e| ballista_error(&format!("etcd error {:?}", e)))?
+            .map_err(|e| ballista_error(&format!("etcd error {e:?}")))?
             .kvs()
             .iter()
             .map(|kv| (kv.key_str().unwrap().to_owned(), kv.value().to_owned()))
@@ -101,7 +101,7 @@ impl StateBackendClient for EtcdClient {
             .clone()
             .get(prefix, Some(options))
             .await
-            .map_err(|e| ballista_error(&format!("etcd error {:?}", e)))?
+            .map_err(|e| ballista_error(&format!("etcd error {e:?}")))?
             .kvs()
             .iter()
             .map(|kv| (kv.key_str().unwrap().to_owned(), kv.value().to_owned()))
@@ -118,7 +118,7 @@ impl StateBackendClient for EtcdClient {
             .clone()
             .get(prefix.clone(), Some(options))
             .await
-            .map_err(|e| ballista_error(&format!("etcd error {:?}", e)))?
+            .map_err(|e| ballista_error(&format!("etcd error {e:?}")))?
             .kvs()
             .iter()
             .map(|kv| {
@@ -139,7 +139,7 @@ impl StateBackendClient for EtcdClient {
             .await
             .map_err(|e| {
                 warn!("etcd put failed: {}", e);
-                ballista_error(&format!("etcd put failed: {}", e))
+                ballista_error(&format!("etcd put failed: {e}"))
             })
             .map(|_| ())
     }
@@ -163,7 +163,7 @@ impl StateBackendClient for EtcdClient {
             .await
             .map_err(|e| {
                 error!("etcd operation failed: {}", e);
-                ballista_error(&format!("etcd operation failed: {}", e))
+                ballista_error(&format!("etcd operation failed: {e}"))
             })
             .map(|_| ())
     }
@@ -181,7 +181,7 @@ impl StateBackendClient for EtcdClient {
         let current_value = etcd
             .get(from_key.as_str(), None)
             .await
-            .map_err(|e| ballista_error(&format!("etcd error {:?}", e)))?
+            .map_err(|e| ballista_error(&format!("etcd error {e:?}")))?
             .kvs()
             .get(0)
             .map(|kv| kv.value().to_owned());
 
@@ -46,11 +46,11 @@ impl MemoryBackendClient {
     }
 
     fn get_space_key(keyspace: &Keyspace) -> String {
-        format!("/{:?}", keyspace)
+        format!("/{keyspace:?}")
     }
 
     fn get_flat_key(keyspace: &Keyspace, key: &str) -> String {
-        format!("/{:?}/{}", keyspace, key)
+        format!("/{keyspace:?}/{key}")
     }
 }
 
@@ -140,7 +140,7 @@ impl StateBackendClient for MemoryBackendClient {
             .insert(key.clone(), value.clone());
 
         // Notify subscribers
-        let full_key = format!("{}/{}", space_key, key);
+        let full_key = format!("{space_key}/{key}");
         if let Some(res) = self.subscribers.reserve(&full_key) {
             let event = WatchEvent::Put(full_key, value);
             res.complete(&event);
@@ -215,7 +215,7 @@ impl StateBackendClient for MemoryBackendClient {
     }
 
     async fn watch(&self, keyspace: Keyspace, prefix: String) -> Result<Box<dyn Watch>> {
-        let prefix = format!("/{:?}/{}", keyspace, prefix);
+        let prefix = format!("/{keyspace:?}/{prefix}");
 
         Ok(Box::new(MemoryWatch {
             subscriber: self.subscribers.register(prefix.as_bytes()),
@@ -227,7 +227,7 @@ impl StateBackendClient for MemoryBackendClient {
         if let Some(mut space_state) = self.states.get_mut(&space_key) {
             if space_state.value_mut().remove(key).is_some() {
                 // Notify subscribers
-                let full_key = format!("{}/{}", space_key, key);
+                let full_key = format!("{space_key}/{key}");
                 if let Some(res) = self.subscribers.reserve(&full_key) {
                     let event = WatchEvent::Delete(full_key);
                     res.complete(&event);
@@ -345,10 +345,10 @@ mod tests {
         let key = "key";
         let value = "value".as_bytes();
         client
-            .put(Keyspace::Slots, format!("{}/1", key), value.to_vec())
+            .put(Keyspace::Slots, format!("{key}/1"), value.to_vec())
             .await?;
         client
-            .put(Keyspace::Slots, format!("{}/2", key), value.to_vec())
+            .put(Keyspace::Slots, format!("{key}/2"), value.to_vec())
             .await?;
         assert_eq!(
             client.get_from_prefix(Keyspace::Slots, key).await?,
 
@@ -88,6 +88,7 @@ impl FlightSqlServiceImpl {
         }
     }
 
+    #[allow(deprecated)]
     #[allow(deprecated)]
     fn tables(&self, ctx: Arc<SessionContext>) -> Result<RecordBatch, ArrowError> {
         let schema = Arc::new(Schema::new(vec![
@@ -157,6 +158,7 @@ impl FlightSqlServiceImpl {
         let bearer = "Bearer ";
         if !authorization.starts_with(bearer) {
             Err(Status::internal("Invalid auth header!"))?;
+            Err(Status::internal("Invalid auth header!"))?;
         }
         let auth = authorization[bearer.len()..].to_string();
 
 
@@ -127,7 +127,7 @@ impl PrometheusMetricsCollector {
             registry
         )
         .map_err(|e| {
-            BallistaError::Internal(format!("Error registering metric: {:?}", e))
+            BallistaError::Internal(format!("Error registering metric: {e:?}"))
         })?;
 
         Ok(Self {
 
@@ -29,7 +29,7 @@ use ballista_core::serde::protobuf::{
     HeartBeatResult, PollWorkParams, PollWorkResult, RegisterExecutorParams,
     RegisterExecutorResult, UpdateTaskStatusParams, UpdateTaskStatusResult,
 };
-use ballista_core::serde::scheduler::ExecutorMetadata;
+use ballista_core::serde::scheduler::ExecutorMetadata;use ballista_core::serde::AsExecutionPlan;
 
 use datafusion::datasource::file_format::parquet::ParquetFormat;
 use datafusion::datasource::file_format::FileFormat;
@@ -312,7 +312,7 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan> SchedulerGrpc
 
         let GetFileMetadataParams { path, file_type } = request.into_inner();
         let file_format: Arc<dyn FileFormat> = match file_type.as_str() {
-            "parquet" => Ok(Arc::new(ParquetFormat::new(config))),
+            "parquet" => Ok(Arc::new(ParquetFormat::new())),
             // TODO implement for CSV
             _ => Err(tonic::Status::unimplemented(
                 "get_file_metadata unsupported file type",
@@ -601,6 +601,7 @@ mod test {
     use ballista_core::serde::scheduler::ExecutorSpecification;
     use ballista_core::serde::BallistaCodec;
 
+    use crate::state::backend::cluster::DefaultClusterState;
     use crate::state::executor_manager::DEFAULT_EXECUTOR_TIMEOUT_SECONDS;
     use crate::state::SchedulerState;
     use crate::test_utils::test_cluster_context;
 
@@ -23,7 +23,7 @@ use ballista_core::event_loop::{EventLoop, EventSender};
 use ballista_core::serde::protobuf::{StopExecutorParams, TaskStatus};
 use ballista_core::serde::BallistaCodec;
 use ballista_core::serde::protobuf::{JobStatus, StopExecutorParams, TaskStatus};
-use ballista_core::serde::{AsExecutionPlan, BallistaCodec};
+use ballista_core::serde::BallistaCodec;
 use ballista_core::utils::default_session_builder;
 
 use datafusion::execution::context::SessionState;
@@ -558,7 +558,8 @@ mod test {
         match status.status {
             Some(job_status::Status::Successful(SuccessfulJob {
                 partition_location,
-                ..
+                queued_at: _,
+                completed_at: _,
             })) => {
                 assert_eq!(partition_location.len(), 4);
             }
Original file line number	Diff line number	Diff line change
`@@ -88,6 +88,7 @@ impl FlightSqlServiceImpl {`
`88`	`88`	`}`
`89`	`89`	`}`
`90`	`90`
	`91`	`+ #[allow(deprecated)]`
`91`	`92`	`#[allow(deprecated)]`
`92`	`93`	`fn tables(&self, ctx: Arc<SessionContext>) -> Result<RecordBatch, ArrowError> {`
`93`	`94`	`let schema = Arc::new(Schema::new(vec![`
`@@ -157,6 +158,7 @@ impl FlightSqlServiceImpl {`
`157`	`158`	`let bearer = "Bearer ";`
`158`	`159`	`if !authorization.starts_with(bearer) {`
`159`	`160`	`Err(Status::internal("Invalid auth header!"))?;`
	`161`	`+ Err(Status::internal("Invalid auth header!"))?;`
`160`	`162`	`}`
`161`	`163`	`let auth = authorization[bearer.len()..].to_string();`
`162`	`164`
Original file line number	Diff line number	Diff line change
`@@ -127,7 +127,7 @@ impl PrometheusMetricsCollector {`
`127`	`127`	`registry`
`128`	`128`	`)`
`129`	`129`	`.map_err(\|e\| {`
`130`		`- BallistaError::Internal(format!("Error registering metric: {:?}", e))`
	`130`	`+ BallistaError::Internal(format!("Error registering metric: {e:?}"))`
`131`	`131`	`})?;`
`132`	`132`
`133`	`133`	`Ok(Self {`