turn on clippy rule for needless borrow

ballista/rust/core/src/execution_plans/query_stage.rs

@@ -139,7 +139,7 @@ impl ExecutionPlan for QueryStageExec {
                 info!("Writing results to {}", path);
 
                 // stream results to disk
-                let stats = utils::write_stream_to_disk(&mut stream, &path)
+                let stats = utils::write_stream_to_disk(&mut stream, path)
                     .await
                     .map_err(|e| DataFusionError::Execution(format!("{:?}", e)))?;
 

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

@@ -1033,9 +1033,7 @@ impl TryInto<protobuf::LogicalExprNode> for &Expr {
                     .map(|e| e.try_into())
                     .collect::<Result<Vec<_>, _>>()?;
                 let window_frame = window_frame.map(|window_frame| {
-                    protobuf::window_expr_node::WindowFrame::Frame(
-                        window_frame.clone().into(),
-                    )
+                    protobuf::window_expr_node::WindowFrame::Frame(window_frame.into())
                 });
                 let window_expr = Box::new(protobuf::WindowExprNode {
                     expr: Some(Box::new(arg.try_into()?)),

ballista/rust/executor/src/flight_service.rs

@@ -279,7 +279,7 @@ fn create_flight_iter(
     options: &IpcWriteOptions,
 ) -> Box<dyn Iterator<Item = Result<FlightData, Status>>> {
     let (flight_dictionaries, flight_batch) =
-        arrow_flight::utils::flight_data_from_arrow_batch(batch, &options);
+        arrow_flight::utils::flight_data_from_arrow_batch(batch, options);
     Box::new(
         flight_dictionaries
             .into_iter()

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

@@ -223,7 +223,7 @@ impl SchedulerState {
             .collect();
         let executors = self.get_executors_metadata().await?;
         'tasks: for (_key, value) in kvs.iter() {
-            let mut status: TaskStatus = decode_protobuf(&value)?;
+            let mut status: TaskStatus = decode_protobuf(value)?;
             if status.status.is_none() {
                 let partition = status.partition_id.as_ref().unwrap();
                 let plan = self

benchmarks/src/bin/tpch.rs

@@ -350,7 +350,7 @@ async fn execute_query(
     if debug {
         println!("Logical plan:\n{:?}", plan);
     }
-    let plan = ctx.optimize(&plan)?;
+    let plan = ctx.optimize(plan)?;
     if debug {
         println!("Optimized logical plan:\n{:?}", plan);
     }
@@ -921,9 +921,9 @@ mod tests {
                 .iter()
                 .map(|field| {
                     Field::new(
-                        Field::name(&field),
+                        Field::name(field),
                         DataType::Utf8,
-                        Field::is_nullable(&field),
+                        Field::is_nullable(field),
                     )
                 })
                 .collect::<Vec<Field>>(),
@@ -939,8 +939,8 @@ mod tests {
                 .iter()
                 .map(|field| {
                     Field::new(
-                        Field::name(&field),
-                        Field::data_type(&field).to_owned(),
+                        Field::name(field),
+                        Field::data_type(field).to_owned(),
                         true,
                     )
                 })
@@ -990,10 +990,10 @@ mod tests {
                     .map(|field| {
                         Expr::Alias(
                             Box::new(Cast {
-                                expr: Box::new(trim(col(Field::name(&field)))),
-                                data_type: Field::data_type(&field).to_owned(),
+                                expr: Box::new(trim(col(Field::name(field)))),
+                                data_type: Field::data_type(field).to_owned(),
                             }),
-                            Field::name(&field).to_string(),
+                            Field::name(field).to_string(),
                         )
                     })
                     .collect::<Vec<Expr>>(),

datafusion/benches/aggregate_query_sql.rs

@@ -47,7 +47,7 @@ fn query(ctx: Arc<Mutex<ExecutionContext>>, sql: &str) {
     let rt = Runtime::new().unwrap();
 
     // execute the query
-    let df = ctx.lock().unwrap().sql(&sql).unwrap();
+    let df = ctx.lock().unwrap().sql(sql).unwrap();
     rt.block_on(df.collect()).unwrap();
 }
 

datafusion/benches/filter_query_sql.rs

@@ -28,7 +28,7 @@ use std::sync::Arc;
 
 async fn query(ctx: &mut ExecutionContext, sql: &str) {
     // execute the query
-    let df = ctx.sql(&sql).unwrap();
+    let df = ctx.sql(sql).unwrap();
     let results = df.collect().await.unwrap();
 
     // display the relation

datafusion/benches/math_query_sql.rs

@@ -40,7 +40,7 @@ fn query(ctx: Arc<Mutex<ExecutionContext>>, sql: &str) {
     let rt = Runtime::new().unwrap();
 
     // execute the query
-    let df = ctx.lock().unwrap().sql(&sql).unwrap();
+    let df = ctx.lock().unwrap().sql(sql).unwrap();
     rt.block_on(df.collect()).unwrap();
 }
 

datafusion/benches/sort_limit_query_sql.rs

@@ -35,7 +35,7 @@ fn query(ctx: Arc<Mutex<ExecutionContext>>, sql: &str) {
     let rt = Runtime::new().unwrap();
 
     // execute the query
-    let df = ctx.lock().unwrap().sql(&sql).unwrap();
+    let df = ctx.lock().unwrap().sql(sql).unwrap();
     rt.block_on(df.collect()).unwrap();
 }
 

datafusion/src/datasource/csv.rs

@@ -204,7 +204,7 @@ impl TableProvider for CsvFile {
                 }
             }
             Source::Path(p) => {
-                CsvExec::try_new(&p, opts, projection.clone(), batch_size, limit)?
+                CsvExec::try_new(p, opts, projection.clone(), batch_size, limit)?
             }
         };
         Ok(Arc::new(exec))

datafusion/src/datasource/json.rs

@@ -149,7 +149,7 @@ impl TableProvider for NdJsonFile {
                 }
             }
             Source::Path(p) => {
-                NdJsonExec::try_new(&p, opts, projection.clone(), batch_size, limit)?
+                NdJsonExec::try_new(p, opts, projection.clone(), batch_size, limit)?
             }
         };
         Ok(Arc::new(exec))

datafusion/src/execution/context.rs

@@ -275,7 +275,7 @@ impl ExecutionContext {
     ) -> Result<Arc<dyn DataFrame>> {
         Ok(Arc::new(DataFrameImpl::new(
             self.state.clone(),
-            &LogicalPlanBuilder::scan_csv(&filename, options, None)?.build()?,
+            &LogicalPlanBuilder::scan_csv(filename, options, None)?.build()?,
         )))
     }
 
@@ -284,7 +284,7 @@ impl ExecutionContext {
         Ok(Arc::new(DataFrameImpl::new(
             self.state.clone(),
             &LogicalPlanBuilder::scan_parquet(
-                &filename,
+                filename,
                 None,
                 self.state.lock().unwrap().config.concurrency,
             )?
@@ -328,7 +328,7 @@ impl ExecutionContext {
     /// executed against this context.
     pub fn register_parquet(&mut self, name: &str, filename: &str) -> Result<()> {
         let table = ParquetTable::try_new(
-            &filename,
+            filename,
             self.state.lock().unwrap().config.concurrency,
         )?;
         self.register_table(name, Arc::new(table))?;
@@ -3205,7 +3205,7 @@ mod tests {
             .expect("Executing CREATE EXTERNAL TABLE");
 
         let sql = "SELECT * from csv_with_timestamps";
-        let result = plan_and_collect(&mut ctx, &sql).await.unwrap();
+        let result = plan_and_collect(&mut ctx, sql).await.unwrap();
         let expected = vec![
             "+--------+-------------------------+",
             "| name   | ts                      |",

datafusion/src/execution/dataframe_impl.rs

@@ -373,7 +373,7 @@ mod tests {
         ctx.register_csv(
             "aggregate_test_100",
             &format!("{}/csv/aggregate_test_100.csv", testdata),
-            CsvReadOptions::new().schema(&schema.as_ref()),
+            CsvReadOptions::new().schema(schema.as_ref()),
         )?;
         Ok(())
     }

datafusion/src/lib.rs

@@ -14,7 +14,7 @@
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.
-#![warn(missing_docs)]
+#![warn(missing_docs, clippy::needless_borrow)]
 // Clippy lints, some should be disabled incrementally
 #![allow(
     clippy::float_cmp,

datafusion/src/logical_plan/dfschema.rs

@@ -325,12 +325,12 @@ impl DFField {
 
     /// Returns an immutable reference to the `DFField`'s unqualified name
     pub fn name(&self) -> &String {
-        &self.field.name()
+        self.field.name()
     }
 
     /// Returns an immutable reference to the `DFField`'s data-type
     pub fn data_type(&self) -> &DataType {
-        &self.field.data_type()
+        self.field.data_type()
     }
 
     /// Indicates whether this `DFField` supports null values

datafusion/src/logical_plan/plan.rs

@@ -221,23 +221,23 @@ impl LogicalPlan {
     /// Get a reference to the logical plan's schema
     pub fn schema(&self) -> &DFSchemaRef {
         match self {
-            LogicalPlan::EmptyRelation { schema, .. } => &schema,
+            LogicalPlan::EmptyRelation { schema, .. } => schema,
             LogicalPlan::TableScan {
                 projected_schema, ..
-            } => &projected_schema,
-            LogicalPlan::Projection { schema, .. } => &schema,
+            } => projected_schema,
+            LogicalPlan::Projection { schema, .. } => schema,
             LogicalPlan::Filter { input, .. } => input.schema(),
-            LogicalPlan::Window { schema, .. } => &schema,
-            LogicalPlan::Aggregate { schema, .. } => &schema,
+            LogicalPlan::Window { schema, .. } => schema,
+            LogicalPlan::Aggregate { schema, .. } => schema,
             LogicalPlan::Sort { input, .. } => input.schema(),
-            LogicalPlan::Join { schema, .. } => &schema,
-            LogicalPlan::CrossJoin { schema, .. } => &schema,
+            LogicalPlan::Join { schema, .. } => schema,
+            LogicalPlan::CrossJoin { schema, .. } => schema,
             LogicalPlan::Repartition { input, .. } => input.schema(),
             LogicalPlan::Limit { input, .. } => input.schema(),
-            LogicalPlan::CreateExternalTable { schema, .. } => &schema,
-            LogicalPlan::Explain { schema, .. } => &schema,
-            LogicalPlan::Extension { node } => &node.schema(),
-            LogicalPlan::Union { schema, .. } => &schema,
+            LogicalPlan::CreateExternalTable { schema, .. } => schema,
+            LogicalPlan::Explain { schema, .. } => schema,
+            LogicalPlan::Extension { node } => node.schema(),
+            LogicalPlan::Union { schema, .. } => schema,
         }
     }
 
@@ -246,12 +246,12 @@ impl LogicalPlan {
         match self {
             LogicalPlan::TableScan {
                 projected_schema, ..
-            } => vec![&projected_schema],
+            } => vec![projected_schema],
             LogicalPlan::Window { input, schema, .. }
             | LogicalPlan::Aggregate { input, schema, .. }
             | LogicalPlan::Projection { input, schema, .. } => {
                 let mut schemas = input.all_schemas();
-                schemas.insert(0, &schema);
+                schemas.insert(0, schema);
                 schemas
             }
             LogicalPlan::Join {
@@ -267,16 +267,16 @@ impl LogicalPlan {
             } => {
                 let mut schemas = left.all_schemas();
                 schemas.extend(right.all_schemas());
-                schemas.insert(0, &schema);
+                schemas.insert(0, schema);
                 schemas
             }
             LogicalPlan::Union { schema, .. } => {
                 vec![schema]
             }
-            LogicalPlan::Extension { node } => vec![&node.schema()],
+            LogicalPlan::Extension { node } => vec![node.schema()],
             LogicalPlan::Explain { schema, .. }
             | LogicalPlan::EmptyRelation { schema, .. }
-            | LogicalPlan::CreateExternalTable { schema, .. } => vec![&schema],
+            | LogicalPlan::CreateExternalTable { schema, .. } => vec![schema],
             LogicalPlan::Limit { input, .. }
             | LogicalPlan::Repartition { input, .. }
             | LogicalPlan::Sort { input, .. }

datafusion/src/optimizer/filter_push_down.rs

@@ -137,7 +137,7 @@ fn get_join_predicates<'a>(
             let all_in_right = right.len() == columns.len();
             !all_in_left && !all_in_right
         })
-        .map(|((ref a, ref b), _)| (a, b))
+        .map(|((a, b), _)| (a, b))
         .unzip();
     (pushable_to_left, pushable_to_right, keep)
 }
@@ -151,7 +151,7 @@ fn push_down(state: &State, plan: &LogicalPlan) -> Result<LogicalPlan> {
         .collect::<Result<Vec<_>>>()?;
 
     let expr = plan.expressions();
-    utils::from_plan(&plan, &expr, &new_inputs)
+    utils::from_plan(plan, &expr, &new_inputs)
 }
 
 /// returns a new [LogicalPlan] that wraps `plan` in a [LogicalPlan::Filter] with
@@ -225,8 +225,8 @@ fn split_members<'a>(predicate: &'a Expr, predicates: &mut Vec<&'a Expr>) {
             op: Operator::And,
             left,
         } => {
-            split_members(&left, predicates);
-            split_members(&right, predicates);
+            split_members(left, predicates);
+            split_members(right, predicates);
         }
         other => predicates.push(other),
     }
@@ -297,7 +297,7 @@ fn optimize(plan: &LogicalPlan, mut state: State) -> Result<LogicalPlan> {
             // optimize inner
             let new_input = optimize(input, state)?;
 
-            utils::from_plan(&plan, &expr, &[new_input])
+            utils::from_plan(plan, expr, &[new_input])
         }
         LogicalPlan::Aggregate {
             input, aggr_expr, ..
@@ -335,7 +335,7 @@ fn optimize(plan: &LogicalPlan, mut state: State) -> Result<LogicalPlan> {
         LogicalPlan::Join { left, right, .. }
         | LogicalPlan::CrossJoin { left, right, .. } => {
             let (pushable_to_left, pushable_to_right, keep) =
-                get_join_predicates(&state, &left.schema(), &right.schema());
+                get_join_predicates(&state, left.schema(), right.schema());
 
             let mut left_state = state.clone();
             left_state.filters = keep_filters(&left_state.filters, &pushable_to_left);
@@ -347,7 +347,7 @@ fn optimize(plan: &LogicalPlan, mut state: State) -> Result<LogicalPlan> {
 
             // create a new Join with the new `left` and `right`
             let expr = plan.expressions();
-            let plan = utils::from_plan(&plan, &expr, &[left, right])?;
+            let plan = utils::from_plan(plan, &expr, &[left, right])?;
 
             if keep.0.is_empty() {
                 Ok(plan)
@@ -437,11 +437,11 @@ impl FilterPushDown {
 
 /// replaces columns by its name on the projection.
 fn rewrite(expr: &Expr, projection: &HashMap<String, Expr>) -> Result<Expr> {
-    let expressions = utils::expr_sub_expressions(&expr)?;
+    let expressions = utils::expr_sub_expressions(expr)?;
 
     let expressions = expressions
         .iter()
-        .map(|e| rewrite(e, &projection))
+        .map(|e| rewrite(e, projection))
         .collect::<Result<Vec<_>>>()?;
 
     if let Expr::Column(name) = expr {
@@ -450,7 +450,7 @@ fn rewrite(expr: &Expr, projection: &HashMap<String, Expr>) -> Result<Expr> {
         }
     }
 
-    utils::rewrite_expression(&expr, &expressions)
+    utils::rewrite_expression(expr, &expressions)
 }
 
 #[cfg(test)]