feat: added clippy::needless_pass_by_value lint rule to datafusion/expr (#18532)

## Which issue does this PR close?
- Closes #18504.

## Rationale for this change
Followed suggestions to not update any public-facing APIs and put the
lint rule in the appropriate spot.

## What changes are included in this PR?
* Add `#![deny(clippy::needless_pass_by_value)]` and `#![cfg_attr(test,
allow(clippy::needless_pass_by_value))]` to `lib.rs`.
* Add `#[allow(clippy::needless_pass_by_value)]` to public functions
* fix `rewrite_in_terms_of_projection()` and
`get_exprs_except_skipped()` to use references per the lint suggestion

## Are these changes tested?
Yes, though the same test failed even without changes to the public
APIs:
`test expr_rewriter::order_by::test::rewrite_sort_cols_by_agg_alias ...
FAILED`
I'll append the logs for your convenience:
```
failures:

---- expr_rewriter::order_by::test::rewrite_sort_cols_by_agg_alias stdout ----
running: 'c1 --> c1  -- column *named* c1 that came out of the projection, (not t.c1)'
running: 'min(c2) --> "min(c2)" -- (column *named* "min(t.c2)"!)'

thread 'expr_rewriter::order_by::test::rewrite_sort_cols_by_agg_alias' (27524241) panicked at datafusion/expr/src/expr_rewriter/order_by.rs:308:13:
assertion `left == right` failed: 

input:Sort { expr: AggregateFunction(AggregateFunction { func: AggregateUDF { inner: Min { name: "min", signature: Signature { type_signature: VariadicAny, volatility: Immutable, parameter_names: None } } }, params: AggregateFunctionParams { args: [Column(Column { relation: None, name: "c2" })], distinct: false, filter: None, order_by: [], null_treatment: None } }), asc: true, nulls_first: true }
rewritten:Sort { expr: Column(Column { relation: None, name: "min(t.c2)" }), asc: true, nulls_first: true }
expected:Sort { expr: Column(Column { relation: Some(Bare { table: "min(t" }), name: "c2)" }), asc: true, nulls_first: true }

  left: Sort { expr: Column(Column { relation: None, name: "min(t.c2)" }), asc: true, nulls_first: true }
 right: Sort { expr: Column(Column { relation: Some(Bare { table: "min(t" }), name: "c2)" }), asc: true, nulls_first: true }
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

failures:
    expr_rewriter::order_by::test::rewrite_sort_cols_by_agg_alias

```
## Are there any user-facing changes?
No, all modification were constrained to internal APIs.

---------

Co-authored-by: Yongting You <[email protected]>

Author: Gohlub

datafusion/expr/src/execution_props.rs

@@ -102,6 +102,7 @@ impl ExecutionProps {
     }
 
     /// Returns the provider for the `var_type`, if any
+    #[allow(clippy::needless_pass_by_value)]
     pub fn get_var_provider(
         &self,
         var_type: VarType,

datafusion/expr/src/expr_rewriter/order_by.rs

@@ -52,7 +52,7 @@ fn rewrite_sort_col_by_aggs(expr: Expr, plan: &LogicalPlan) -> Result<Expr> {
     // on top of them)
     if plan_inputs.len() == 1 {
         let proj_exprs = plan.expressions();
-        rewrite_in_terms_of_projection(expr, proj_exprs, plan_inputs[0])
+        rewrite_in_terms_of_projection(expr, &proj_exprs, plan_inputs[0])
     } else {
         Ok(expr)
     }
@@ -71,7 +71,7 @@ fn rewrite_sort_col_by_aggs(expr: Expr, plan: &LogicalPlan) -> Result<Expr> {
 /// 2. t produces an output schema with two columns "a", "b + c"
 fn rewrite_in_terms_of_projection(
     expr: Expr,
-    proj_exprs: Vec<Expr>,
+    proj_exprs: &[Expr],
     input: &LogicalPlan,
 ) -> Result<Expr> {
     // assumption is that each item in exprs, such as "b + c" is
@@ -104,7 +104,7 @@ fn rewrite_in_terms_of_projection(
 
         // look for the column named the same as this expr
         let mut found = None;
-        for proj_expr in &proj_exprs {
+        for proj_expr in proj_exprs {
             proj_expr.apply(|e| {
                 if expr_match(&search_col, e) {
                     found = Some(e.clone());

datafusion/expr/src/lib.rs

@@ -23,6 +23,9 @@
 // Make sure fast / cheap clones on Arc are explicit:
 // https://github.com/apache/datafusion/issues/11143
 #![deny(clippy::clone_on_ref_ptr)]
+// https://github.com/apache/datafusion/issues/18503
+#![deny(clippy::needless_pass_by_value)]
+#![cfg_attr(test, allow(clippy::needless_pass_by_value))]
 
 //! [DataFusion](https://github.com/apache/datafusion)
 //! is an extensible query execution framework that uses

datafusion/expr/src/literal.rs

@@ -21,10 +21,12 @@ use crate::Expr;
 use datafusion_common::{metadata::FieldMetadata, ScalarValue};
 
 /// Create a literal expression
+#[allow(clippy::needless_pass_by_value)]
 pub fn lit<T: Literal>(n: T) -> Expr {
     n.lit()
 }
 
+#[allow(clippy::needless_pass_by_value)]
 pub fn lit_with_metadata<T: Literal>(n: T, metadata: Option<FieldMetadata>) -> Expr {
     let Some(metadata) = metadata else {
         return n.lit();
@@ -45,6 +47,7 @@ pub fn lit_with_metadata<T: Literal>(n: T, metadata: Option<FieldMetadata>) -> E
 }
 
 /// Create a literal timestamp expression
+#[allow(clippy::needless_pass_by_value)]
 pub fn lit_timestamp_nano<T: TimestampLiteral>(n: T) -> Expr {
     n.lit_timestamp_nano()
 }

datafusion/expr/src/logical_plan/plan.rs

@@ -3481,6 +3481,7 @@ impl Aggregate {
     ///
     /// This method should only be called when you are absolutely sure that the schema being
     /// provided is correct for the aggregate. If in doubt, call [try_new](Self::try_new) instead.
+    #[allow(clippy::needless_pass_by_value)]
     pub fn try_new_with_schema(
         input: Arc<LogicalPlan>,
         group_expr: Vec<Expr>,

datafusion/expr/src/utils.rs

@@ -354,7 +354,7 @@ fn get_excluded_columns(
 /// Returns all `Expr`s in the schema, except the `Column`s in the `columns_to_skip`
 fn get_exprs_except_skipped(
     schema: &DFSchema,
-    columns_to_skip: HashSet<Column>,
+    columns_to_skip: &HashSet<Column>,
 ) -> Vec<Expr> {
     if columns_to_skip.is_empty() {
         schema.iter().map(Expr::from).collect::<Vec<Expr>>()
@@ -419,7 +419,7 @@ pub fn expand_wildcard(
     };
     // Add each excluded `Column` to columns_to_skip
     columns_to_skip.extend(excluded_columns);
-    Ok(get_exprs_except_skipped(schema, columns_to_skip))
+    Ok(get_exprs_except_skipped(schema, &columns_to_skip))
 }
 
 /// Resolves an `Expr::Wildcard` to a collection of qualified `Expr::Column`'s.
@@ -464,7 +464,7 @@ pub fn expand_qualified_wildcard(
     columns_to_skip.extend(excluded_columns);
     Ok(get_exprs_except_skipped(
         &qualified_dfschema,
-        columns_to_skip,
+        &columns_to_skip,
     ))
 }
 
@@ -928,6 +928,7 @@ pub fn find_valid_equijoin_key_pair(
 ///     round(Float64)
 ///     round(Float32)
 /// ```
+#[allow(clippy::needless_pass_by_value)]
 pub fn generate_signature_error_msg(
     func_name: &str,
     func_signature: Signature,