Remove unnecessary clones with clippy (#12197)

* Remove unnecessary clones with clippy

This is automated change done with

```
cargo clippy --fix -- -Aclippy::all -Wclippy::redundant_clone
cargo fmt
 # manually fix few new clippy errors introduced by clippy itself
```

Note: it doesn't remove all unnecessary clones because the command
reported error and backed out for the common crate.

* Prevent clippy from removing intentional clone

clippy can be run with `--fix` and then it won't obey the code comment
instructing not to delete the clone.

* Remove unnecessary clones pointed out by clippy

Change code as instructed by

```
cargo clippy --fix -- -Aclippy::all -Wclippy::redundant_clone
```

where clippy didn't apply the suggested changes by itself.

datafusion-examples/examples/advanced_parquet_index.rs

@@ -300,7 +300,7 @@ impl IndexTableProvider {
         // analyze the predicate. In a real system, using
         // `PruningPredicate::prune` would likely be easier to do.
         let pruning_predicate =
-            PruningPredicate::try_new(Arc::clone(predicate), self.schema().clone())?;
+            PruningPredicate::try_new(Arc::clone(predicate), self.schema())?;
 
         // The PruningPredicate's guarantees must all be satisfied in order for
         // the predicate to possibly evaluate to true.

datafusion-examples/examples/file_stream_provider.rs

@@ -100,7 +100,7 @@ mod non_windows {
     ) {
         // Timeout for a long period of BrokenPipe error
         let broken_pipe_timeout = Duration::from_secs(10);
-        let sa = file_path.clone();
+        let sa = file_path;
         // Spawn a new thread to write to the FIFO file
         #[allow(clippy::disallowed_methods)] // spawn allowed only in tests
         tasks.spawn_blocking(move || {

datafusion/common/src/dfschema.rs

@@ -1242,10 +1242,9 @@ mod tests {
     #[test]
     fn into() {
         // Demonstrate how to convert back and forth between Schema, SchemaRef, DFSchema, and DFSchemaRef
-        let metadata = test_metadata();
         let arrow_schema = Schema::new_with_metadata(
             vec![Field::new("c0", DataType::Int64, true)],
-            metadata.clone(),
+            test_metadata(),
         );
         let arrow_schema_ref = Arc::new(arrow_schema.clone());
 

datafusion/common/src/file_options/parquet_writer.rs

@@ -565,7 +565,7 @@ mod tests {
                 column_options_with_non_defaults(&parquet_options),
             )]
             .into(),
-            key_value_metadata: [(key.clone(), value.clone())].into(),
+            key_value_metadata: [(key, value)].into(),
         };
 
         let writer_props = WriterPropertiesBuilder::try_from(&table_parquet_opts)

datafusion/common/src/scalar/mod.rs

@@ -4905,7 +4905,7 @@ mod tests {
         let data_type =
             DataType::List(Arc::new(Field::new("item", DataType::Int32, true)));
 
-        assert_eq!(non_null_list_scalar.data_type(), data_type.clone());
+        assert_eq!(non_null_list_scalar.data_type(), data_type);
         assert_eq!(null_list_scalar.data_type(), data_type);
     }
 
@@ -5582,13 +5582,13 @@ mod tests {
 
         // Define list-of-structs scalars
 
-        let nl0_array = ScalarValue::iter_to_array(vec![s0.clone(), s1.clone()]).unwrap();
+        let nl0_array = ScalarValue::iter_to_array(vec![s0, s1.clone()]).unwrap();
         let nl0 = ScalarValue::List(Arc::new(array_into_list_array_nullable(nl0_array)));
 
-        let nl1_array = ScalarValue::iter_to_array(vec![s2.clone()]).unwrap();
+        let nl1_array = ScalarValue::iter_to_array(vec![s2]).unwrap();
         let nl1 = ScalarValue::List(Arc::new(array_into_list_array_nullable(nl1_array)));
 
-        let nl2_array = ScalarValue::iter_to_array(vec![s1.clone()]).unwrap();
+        let nl2_array = ScalarValue::iter_to_array(vec![s1]).unwrap();
         let nl2 = ScalarValue::List(Arc::new(array_into_list_array_nullable(nl2_array)));
 
         // iter_to_array for list-of-struct

datafusion/common/src/stats.rs

@@ -557,6 +557,7 @@ mod tests {
         let precision: Precision<ScalarValue> =
             Precision::Exact(ScalarValue::Int64(Some(42)));
         // Clippy would complain about this if it were Copy
+        #[allow(clippy::redundant_clone)]
         let p2 = precision.clone();
         assert_eq!(precision, p2);
     }

datafusion/core/src/datasource/listing/helpers.rs

@@ -282,7 +282,7 @@ async fn prune_partitions(
         Default::default(),
     )?;
 
-    let batch = RecordBatch::try_new(schema.clone(), arrays)?;
+    let batch = RecordBatch::try_new(schema, arrays)?;
 
     // TODO: Plumb this down
     let props = ExecutionProps::new();

datafusion/core/src/datasource/listing/table.rs

@@ -1016,7 +1016,7 @@ impl ListingTable {
             .collected_statistics
             .get_with_extra(&part_file.object_meta.location, &part_file.object_meta)
         {
-            Some(statistics) => Ok(statistics.clone()),
+            Some(statistics) => Ok(statistics),
             None => {
                 let statistics = self
                     .options

datafusion/core/src/datasource/physical_plan/file_groups.rs

@@ -394,7 +394,7 @@ mod test {
     #[test]
     fn repartition_empty_file_only() {
         let partitioned_file_empty = pfile("empty", 0);
-        let file_group = vec![vec![partitioned_file_empty.clone()]];
+        let file_group = vec![vec![partitioned_file_empty]];
 
         let partitioned_files = FileGroupPartitioner::new()
             .with_target_partitions(4)
@@ -817,10 +817,7 @@ mod test {
             .with_preserve_order_within_groups(true)
             .repartition_file_groups(&file_groups);
 
-        assert_partitioned_files(
-            repartitioned.clone(),
-            repartitioned_preserving_sort.clone(),
-        );
+        assert_partitioned_files(repartitioned.clone(), repartitioned_preserving_sort);
         repartitioned
     }
 }

datafusion/core/src/datasource/physical_plan/file_scan_config.rs

@@ -908,7 +908,7 @@ mod tests {
             schema.clone(),
             Some(vec![0, 3, 5, schema.fields().len()]),
             Statistics::new_unknown(&schema),
-            to_partition_cols(partition_cols.clone()),
+            to_partition_cols(partition_cols),
         )
         .projected_file_schema();
 
@@ -941,7 +941,7 @@ mod tests {
             schema.clone(),
             None,
             Statistics::new_unknown(&schema),
-            to_partition_cols(partition_cols.clone()),
+            to_partition_cols(partition_cols),
         )
         .projected_file_schema();
 

datafusion/core/src/datasource/schema_adapter.rs

@@ -369,7 +369,7 @@ mod tests {
             let f1 = Field::new("id", DataType::Int32, true);
             let f2 = Field::new("extra_column", DataType::Utf8, true);
 
-            let schema = Arc::new(Schema::new(vec![f1.clone(), f2.clone()]));
+            let schema = Arc::new(Schema::new(vec![f1, f2]));
 
             let extra_column = Arc::new(StringArray::from(vec!["foo"]));
             let mut new_columns = batch.columns().to_vec();

datafusion/core/src/execution/context/mod.rs

@@ -1165,7 +1165,7 @@ impl SessionContext {
         // check schema uniqueness
         let mut batches = batches.into_iter().peekable();
         let schema = if let Some(batch) = batches.peek() {
-            batch.schema().clone()
+            batch.schema()
         } else {
             Arc::new(Schema::empty())
         };

datafusion/core/src/physical_optimizer/enforce_distribution.rs

@@ -3908,7 +3908,7 @@ pub(crate) mod tests {
         let alias = vec![("a".to_string(), "a".to_string())];
         let plan_parquet =
             aggregate_exec_with_alias(parquet_exec_multiple(), alias.clone());
-        let plan_csv = aggregate_exec_with_alias(csv_exec_multiple(), alias.clone());
+        let plan_csv = aggregate_exec_with_alias(csv_exec_multiple(), alias);
 
         let expected_parquet = [
             "AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[]",
@@ -3934,7 +3934,7 @@ pub(crate) mod tests {
         let alias = vec![("a".to_string(), "a".to_string())];
         let plan_parquet =
             aggregate_exec_with_alias(parquet_exec_multiple(), alias.clone());
-        let plan_csv = aggregate_exec_with_alias(csv_exec_multiple(), alias.clone());
+        let plan_csv = aggregate_exec_with_alias(csv_exec_multiple(), alias);
 
         let expected_parquet = [
             "AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[]",
@@ -3964,7 +3964,7 @@ pub(crate) mod tests {
             options: SortOptions::default(),
         }];
         let plan_parquet = limit_exec(sort_exec(sort_key.clone(), parquet_exec(), false));
-        let plan_csv = limit_exec(sort_exec(sort_key.clone(), csv_exec(), false));
+        let plan_csv = limit_exec(sort_exec(sort_key, csv_exec(), false));
 
         let expected_parquet = &[
             "GlobalLimitExec: skip=0, fetch=100",
@@ -4000,8 +4000,7 @@ pub(crate) mod tests {
             parquet_exec(),
             false,
         )));
-        let plan_csv =
-            limit_exec(filter_exec(sort_exec(sort_key.clone(), csv_exec(), false)));
+        let plan_csv = limit_exec(filter_exec(sort_exec(sort_key, csv_exec(), false)));
 
         let expected_parquet = &[
             "GlobalLimitExec: skip=0, fetch=100",
@@ -4042,7 +4041,7 @@ pub(crate) mod tests {
         );
         let plan_csv = aggregate_exec_with_alias(
             limit_exec(filter_exec(limit_exec(csv_exec()))),
-            alias.clone(),
+            alias,
         );
 
         let expected_parquet = &[
@@ -4126,7 +4125,7 @@ pub(crate) mod tests {
         );
         let plan_csv = sort_preserving_merge_exec(
             sort_key.clone(),
-            csv_exec_with_sort(vec![sort_key.clone()]),
+            csv_exec_with_sort(vec![sort_key]),
         );
 
         // parallelization is not beneficial for SortPreservingMerge
@@ -4154,7 +4153,7 @@ pub(crate) mod tests {
             union_exec(vec![parquet_exec_with_sort(vec![sort_key.clone()]); 2]);
         let input_csv = union_exec(vec![csv_exec_with_sort(vec![sort_key.clone()]); 2]);
         let plan_parquet = sort_preserving_merge_exec(sort_key.clone(), input_parquet);
-        let plan_csv = sort_preserving_merge_exec(sort_key.clone(), input_csv);
+        let plan_csv = sort_preserving_merge_exec(sort_key, input_csv);
 
         // should not repartition (union doesn't benefit from increased parallelism)
         // should not sort (as the data was already sorted)
@@ -4224,8 +4223,8 @@ pub(crate) mod tests {
             ("c".to_string(), "c2".to_string()),
         ];
         let proj_parquet = projection_exec_with_alias(
-            parquet_exec_with_sort(vec![sort_key.clone()]),
-            alias_pairs.clone(),
+            parquet_exec_with_sort(vec![sort_key]),
+            alias_pairs,
         );
         let sort_key_after_projection = vec![PhysicalSortExpr {
             expr: col("c2", &proj_parquet.schema()).unwrap(),
@@ -4560,7 +4559,7 @@ pub(crate) mod tests {
         }];
         let alias = vec![("a".to_string(), "a".to_string())];
         let input = parquet_exec_with_sort(vec![sort_key]);
-        let physical_plan = aggregate_exec_with_alias(input, alias.clone());
+        let physical_plan = aggregate_exec_with_alias(input, alias);
 
         let expected = &[
             "AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[]",
@@ -4584,7 +4583,7 @@ pub(crate) mod tests {
         let alias = vec![("a".to_string(), "a".to_string())];
         let input = parquet_exec_multiple_sorted(vec![sort_key]);
         let aggregate = aggregate_exec_with_alias(input, alias.clone());
-        let physical_plan = aggregate_exec_with_alias(aggregate, alias.clone());
+        let physical_plan = aggregate_exec_with_alias(aggregate, alias);
 
         let expected = &[
             "AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[]",

datafusion/core/src/physical_optimizer/join_selection.rs

@@ -908,7 +908,7 @@ mod tests_statistical {
         );
 
         let optimized_join = JoinSelection::new()
-            .optimize(join.clone(), &ConfigOptions::new())
+            .optimize(join, &ConfigOptions::new())
             .unwrap();
 
         let swapping_projection = optimized_join
@@ -964,7 +964,7 @@ mod tests_statistical {
         );
 
         let optimized_join = JoinSelection::new()
-            .optimize(join.clone(), &ConfigOptions::new())
+            .optimize(join, &ConfigOptions::new())
             .unwrap();
 
         let swapped_join = optimized_join
@@ -1140,7 +1140,7 @@ mod tests_statistical {
         );
 
         let optimized_join = JoinSelection::new()
-            .optimize(join.clone(), &ConfigOptions::new())
+            .optimize(join, &ConfigOptions::new())
             .unwrap();
 
         let swapped_join = optimized_join
@@ -1180,7 +1180,7 @@ mod tests_statistical {
         );
 
         let optimized_join = JoinSelection::new()
-            .optimize(join.clone(), &ConfigOptions::new())
+            .optimize(join, &ConfigOptions::new())
             .unwrap();
 
         let swapping_projection = optimized_join
@@ -1356,7 +1356,7 @@ mod tests_statistical {
             Arc::new(Column::new_with_schema("small_col", &small.schema()).unwrap()) as _,
         )];
         check_join_partition_mode(
-            big.clone(),
+            big,
             small.clone(),
             join_on,
             true,
@@ -1380,8 +1380,8 @@ mod tests_statistical {
             Arc::new(Column::new_with_schema("small_col", &small.schema()).unwrap()) as _,
         )];
         check_join_partition_mode(
-            empty.clone(),
-            small.clone(),
+            empty,
+            small,
             join_on,
             true,
             PartitionMode::CollectLeft,
@@ -1424,7 +1424,7 @@ mod tests_statistical {
             Arc::new(Column::new_with_schema("big_col", &big.schema()).unwrap()) as _,
         )];
         check_join_partition_mode(
-            bigger.clone(),
+            bigger,
             big.clone(),
             join_on,
             true,
@@ -1472,7 +1472,7 @@ mod tests_statistical {
         );
 
         let optimized_join = JoinSelection::new()
-            .optimize(join.clone(), &ConfigOptions::new())
+            .optimize(join, &ConfigOptions::new())
             .unwrap();
 
         if !is_swapped {
@@ -1913,8 +1913,7 @@ mod hash_join_tests {
             false,
         )?);
 
-        let optimized_join_plan =
-            hash_join_swap_subrule(join.clone(), &ConfigOptions::new())?;
+        let optimized_join_plan = hash_join_swap_subrule(join, &ConfigOptions::new())?;
 
         // If swap did happen
         let projection_added = optimized_join_plan.as_any().is::<ProjectionExec>();

datafusion/core/src/physical_optimizer/projection_pushdown.rs

@@ -1692,12 +1692,9 @@ mod tests {
         ]));
         Arc::new(
             CsvExec::builder(
-                FileScanConfig::new(
-                    ObjectStoreUrl::parse("test:///").unwrap(),
-                    schema.clone(),
-                )
-                .with_file(PartitionedFile::new("x".to_string(), 100))
-                .with_projection(Some(vec![0, 1, 2, 3, 4])),
+                FileScanConfig::new(ObjectStoreUrl::parse("test:///").unwrap(), schema)
+                    .with_file(PartitionedFile::new("x".to_string(), 100))
+                    .with_projection(Some(vec![0, 1, 2, 3, 4])),
             )
             .with_has_header(false)
             .with_delimeter(0)
@@ -1719,12 +1716,9 @@ mod tests {
         ]));
         Arc::new(
             CsvExec::builder(
-                FileScanConfig::new(
-                    ObjectStoreUrl::parse("test:///").unwrap(),
-                    schema.clone(),
-                )
-                .with_file(PartitionedFile::new("x".to_string(), 100))
-                .with_projection(Some(vec![3, 2, 1])),
+                FileScanConfig::new(ObjectStoreUrl::parse("test:///").unwrap(), schema)
+                    .with_file(PartitionedFile::new("x".to_string(), 100))
+                    .with_projection(Some(vec![3, 2, 1])),
             )
             .with_has_header(false)
             .with_delimeter(0)

datafusion/core/src/physical_optimizer/pruning.rs

@@ -1369,7 +1369,6 @@ fn build_predicate_expression(
             let change_expr = in_list
                 .list()
                 .iter()
-                .cloned()
                 .map(|e| {
                     Arc::new(phys_expr::BinaryExpr::new(
                         in_list.expr().clone(),

datafusion/core/src/physical_planner.rs

@@ -701,7 +701,7 @@ impl DefaultPhysicalPlanner {
                 let initial_aggr = Arc::new(AggregateExec::try_new(
                     AggregateMode::Partial,
                     groups.clone(),
-                    aggregates.clone(),
+                    aggregates,
                     filters.clone(),
                     input_exec,
                     physical_input_schema.clone(),
@@ -2569,7 +2569,7 @@ mod tests {
 
     impl NoOpExecutionPlan {
         fn new(schema: SchemaRef) -> Self {
-            let cache = Self::compute_properties(schema.clone());
+            let cache = Self::compute_properties(schema);
             Self { cache }
         }
 

datafusion/core/tests/fuzz_cases/join_fuzz.rs

@@ -350,12 +350,10 @@ impl JoinFuzzTestCase {
     fn left_right(&self) -> (Arc<MemoryExec>, Arc<MemoryExec>) {
         let schema1 = self.input1[0].schema();
         let schema2 = self.input2[0].schema();
-        let left = Arc::new(
-            MemoryExec::try_new(&[self.input1.clone()], schema1.clone(), None).unwrap(),
-        );
-        let right = Arc::new(
-            MemoryExec::try_new(&[self.input2.clone()], schema2.clone(), None).unwrap(),
-        );
+        let left =
+            Arc::new(MemoryExec::try_new(&[self.input1.clone()], schema1, None).unwrap());
+        let right =
+            Arc::new(MemoryExec::try_new(&[self.input2.clone()], schema2, None).unwrap());
         (left, right)
     }