fix clippy

Author: yjshen

ballista/rust/client/src/context.rs

@@ -142,7 +142,7 @@ impl BallistaContext {
         let path = fs::canonicalize(&path)?;
 
         // use local DataFusion context for now but later this might call the scheduler
-        let mut ctx = create_datafusion_context(&self.state.lock().unwrap().config());
+        let mut ctx = create_datafusion_context(self.state.lock().unwrap().config());
         let df = ctx.read_parquet(path.to_str().unwrap())?;
         Ok(df)
     }
@@ -159,7 +159,7 @@ impl BallistaContext {
         let path = fs::canonicalize(&path)?;
 
         // use local DataFusion context for now but later this might call the scheduler
-        let mut ctx = create_datafusion_context(&self.state.lock().unwrap().config());
+        let mut ctx = create_datafusion_context(self.state.lock().unwrap().config());
         let df = ctx.read_csv(path.to_str().unwrap(), options)?;
         Ok(df)
     }
@@ -193,7 +193,7 @@ impl BallistaContext {
         // use local DataFusion context for now but later this might call the scheduler
         // register tables
         let state = self.state.lock().unwrap();
-        let mut ctx = create_datafusion_context(&state.config());
+        let mut ctx = create_datafusion_context(state.config());
         for (name, plan) in &state.tables {
             let plan = ctx.optimize(plan)?;
             let execution_plan = ctx.create_physical_plan(&plan)?;

ballista/rust/core/src/serde/physical_plan/from_proto.rs

@@ -131,9 +131,9 @@ impl TryInto<Arc<dyn ExecutionPlan>> for &protobuf::PhysicalPlanNode {
             }
             PhysicalPlanType::ParquetScan(scan) => {
                 let projection = scan.projection.iter().map(|i| *i as usize).collect();
-                let path: &str = &scan.filename[0].as_str();
+                let path: &str = scan.filename[0].as_str();
                 Ok(Arc::new(ParquetExec::try_from_path(
-                    &path,
+                    path,
                     Some(projection),
                     None,
                     scan.batch_size as usize,

ballista/rust/core/src/serde/physical_plan/to_proto.rs

@@ -259,7 +259,7 @@ impl TryInto<protobuf::PhysicalPlanNode> for Arc<dyn ExecutionPlan> {
             let filenames = exec
                 .partitions()
                 .iter()
-                .flat_map(|part| part.filenames().to_owned())
+                .flat_map(|part| part.filenames())
                 .collect();
             Ok(protobuf::PhysicalPlanNode {
                 physical_plan_type: Some(PhysicalPlanType::ParquetScan(

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

@@ -316,7 +316,7 @@ impl SchedulerState {
                             ))
                             .unwrap();
                         let task_is_dead = self
-                            .reschedule_dead_task(&referenced_task, &executors)
+                            .reschedule_dead_task(referenced_task, &executors)
                             .await?;
                         if task_is_dead {
                             continue 'tasks;

benchmarks/src/bin/tpch.rs

@@ -484,7 +484,7 @@ fn get_table(
             let path = format!("{}/{}", path, table);
             Ok(Arc::new(ParquetTable::try_new(
                 &path,
-                create_datafusion_context_concurrency(max_concurrency),
+                ExecutionContext::with_concurrency(max_concurrency),
             )?))
         }
         other => {
@@ -493,11 +493,6 @@ fn get_table(
     }
 }
 
-fn create_datafusion_context_concurrency(n: usize) -> ExecutionContext {
-    let config = ExecutionConfig::new().with_concurrency(n);
-    ExecutionContext::with_config(config)
-}
-
 fn get_schema(table: &str) -> Schema {
     // note that the schema intentionally uses signed integers so that any generated Parquet
     // files can also be used to benchmark tools that only support signed integers, such as

datafusion/src/datasource/csv.rs

@@ -66,8 +66,8 @@ impl CsvFile {
         let schema = Arc::new(match options.schema {
             Some(s) => s.clone(),
             None => {
-                let filenames =
-                    LocalFileSystem {}.list_all_files(&path, options.file_extension)?;
+                let filenames = LocalFileSystem
+                    .list_all_files(path.as_str(), options.file_extension)?;
                 if filenames.is_empty() {
                     return Err(DataFusionError::Plan(format!(
                         "No files found at {path} with file extension {file_extension}",

datafusion/src/datasource/json.rs

@@ -59,7 +59,7 @@ impl NdJsonFile {
             schema
         } else {
             let filenames =
-                LocalFileSystem {}.list_all_files(&path, options.file_extension)?;
+                LocalFileSystem.list_all_files(path, options.file_extension)?;
             if filenames.is_empty() {
                 return Err(DataFusionError::Plan(format!(
                     "No files found at {path} with file extension {file_extension}",

datafusion/src/datasource/local.rs

@@ -33,7 +33,7 @@ pub struct LocalFileSystem;
 
 impl ObjectStore for LocalFileSystem {
     fn as_any(&self) -> &dyn Any {
-        return self;
+        self
     }
 
     fn list_all_files(&self, path: &str, ext: &str) -> Result<Vec<String>> {

datafusion/src/datasource/object_store.rs

@@ -78,7 +78,7 @@ impl ObjectStoreRegistry {
         store: Arc<dyn ObjectStore>,
     ) -> Option<Arc<dyn ObjectStore>> {
         let mut stores = self.object_stores.write().unwrap();
-        stores.insert(scheme.to_string(), store)
+        stores.insert(scheme, store)
     }
 
     /// Get the store registered for scheme

datafusion/src/datasource/parquet.rs

@@ -58,7 +58,7 @@ impl ParquetTable {
         let max_concurrency = context.state.lock().unwrap().config.concurrency;
         let root_desc = ParquetRootDesc::new(path.as_str(), context);
         Ok(Self {
-            path: path.clone(),
+            path,
             desc: Arc::new(root_desc?),
             max_concurrency,
             enable_pruning: true,
@@ -222,7 +222,7 @@ impl SourceRootDescBuilder for ParquetRootDesc {
         };
 
         Ok(PartitionedFile {
-            file_path: file_path.clone(),
+            file_path,
             schema,
             statistics,
             partition_value: None,

datafusion/src/physical_plan/csv.rs

@@ -143,8 +143,7 @@ impl CsvExec {
     ) -> Result<Self> {
         let file_extension = String::from(options.file_extension);
 
-        let filenames =
-            LocalFileSystem {}.list_all_files(&path, options.file_extension)?;
+        let filenames = LocalFileSystem.list_all_files(path, options.file_extension)?;
         if filenames.is_empty() {
             return Err(DataFusionError::Execution(format!(
                 "No files found at {path} with file extension {file_extension}",

datafusion/src/physical_plan/json.rs

@@ -89,8 +89,7 @@ impl NdJsonExec {
     ) -> Result<Self> {
         let file_extension = options.file_extension.to_string();
 
-        let filenames =
-            LocalFileSystem {}.list_all_files(&path, options.file_extension)?;
+        let filenames = LocalFileSystem.list_all_files(path, options.file_extension)?;
 
         if filenames.is_empty() {
             return Err(DataFusionError::Execution(format!(

datafusion/src/physical_plan/parquet.rs

@@ -279,13 +279,12 @@ impl ParquetPartition {
     }
 
     /// The Parquet filename for this partition
-    pub fn filenames(&self) -> &[String] {
-        todo!()
-    }
-
-    /// Statistics for this partition
-    pub fn statistics(&self) -> &Statistics {
-        todo!()
+    pub fn filenames(&self) -> Vec<String> {
+        self.file_partition
+            .files
+            .iter()
+            .map(|f| f.file_path.clone())
+            .collect()
     }
 }