7.x: fix clippy

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

@@ -198,7 +198,7 @@ impl From<&DataType> for protobuf::arrow_type::ArrowTypeEnum {
             DataType::Timestamp(time_unit, timezone) => {
                 ArrowTypeEnum::Timestamp(protobuf::Timestamp {
                     time_unit: protobuf::TimeUnit::from_arrow_time_unit(time_unit) as i32,
-                    timezone: timezone.to_owned().unwrap_or_else(String::new),
+                    timezone: timezone.to_owned().unwrap_or_default(),
                 })
             }
             DataType::Date32 => ArrowTypeEnum::Date32(EmptyMessage {}),

datafusion/src/execution/context.rs

@@ -1151,7 +1151,7 @@ impl ExecutionProps {
         var_type: VarType,
         provider: Arc<dyn VarProvider + Send + Sync>,
     ) -> Option<Arc<dyn VarProvider + Send + Sync>> {
-        let mut var_providers = self.var_providers.take().unwrap_or_else(HashMap::new);
+        let mut var_providers = self.var_providers.take().unwrap_or_default();
 
         let old_provider = var_providers.insert(var_type, provider);
 
@@ -3602,7 +3602,7 @@ mod tests {
         let logical_plan = ctx.create_logical_plan(sql)?;
         let logical_plan = ctx.optimize(&logical_plan)?;
         let physical_plan = ctx.create_physical_plan(&logical_plan).await?;
-        ctx.write_csv(physical_plan, out_dir.to_string()).await
+        ctx.write_csv(physical_plan, out_dir).await
     }
 
     /// Execute SQL and write results to partitioned parquet files
@@ -3615,7 +3615,7 @@ mod tests {
         let logical_plan = ctx.create_logical_plan(sql)?;
         let logical_plan = ctx.optimize(&logical_plan)?;
         let physical_plan = ctx.create_physical_plan(&logical_plan).await?;
-        ctx.write_parquet(physical_plan, out_dir.to_string(), writer_properties)
+        ctx.write_parquet(physical_plan, out_dir, writer_properties)
             .await
     }
 

datafusion/src/logical_plan/builder.rs

@@ -1041,14 +1041,13 @@ pub(crate) fn expand_wildcard(
     let columns_to_skip = using_columns
         .into_iter()
         // For each USING JOIN condition, only expand to one column in projection
-        .map(|cols| {
+        .flat_map(|cols| {
             let mut cols = cols.into_iter().collect::<Vec<_>>();
             // sort join columns to make sure we consistently keep the same
             // qualified column
             cols.sort();
             cols.into_iter().skip(1)
         })
-        .flatten()
         .collect::<HashSet<_>>();
 
     if columns_to_skip.is_empty() {

datafusion/src/logical_plan/plan.rs

@@ -543,8 +543,7 @@ impl LogicalPlan {
                 {
                     self.using_columns.push(
                         on.iter()
-                            .map(|entry| [&entry.0, &entry.1])
-                            .flatten()
+                            .flat_map(|entry| [&entry.0, &entry.1])
                             .cloned()
                             .collect::<HashSet<Column>>(),
                     );

datafusion/src/optimizer/filter_push_down.rs

@@ -228,14 +228,13 @@ fn get_pushable_join_predicates<'a>(
     let schema_columns = schema
         .fields()
         .iter()
-        .map(|f| {
+        .flat_map(|f| {
             [
                 f.qualified_column(),
                 // we need to push down filter using unqualified column as well
                 f.unqualified_column(),
             ]
         })
-        .flatten()
         .collect::<HashSet<_>>();
 
     state

datafusion/src/physical_plan/hash_aggregate.rs

@@ -1176,12 +1176,11 @@ mod tests {
             _partition: usize,
             _runtime: Arc<RuntimeEnv>,
         ) -> Result<SendableRecordBatchStream> {
-            let stream;
-            if self.yield_first {
-                stream = TestYieldingStream::New;
+            let stream = if self.yield_first {
+                TestYieldingStream::New
             } else {
-                stream = TestYieldingStream::Yielded;
-            }
+                TestYieldingStream::Yielded
+            };
             Ok(Box::pin(stream))
         }
 

datafusion/src/physical_plan/regex_expressions.rs

@@ -139,7 +139,7 @@ pub fn regexp_replace<T: StringOffsetSizeTrait>(args: &[ArrayRef]) -> Result<Arr
                     let (pattern, replace_all) = if flags == "g" {
                         (pattern.to_string(), true)
                     } else if flags.contains('g') {
-                        (format!("(?{}){}", flags.to_string().replace("g", ""), pattern), true)
+                        (format!("(?{}){}", flags.to_string().replace('g', ""), pattern), true)
                     } else {
                         (format!("(?{}){}", flags, pattern), false)
                     };

datafusion/src/physical_plan/union.rs

@@ -201,14 +201,12 @@ fn col_stats_union(
         .min_value
         .zip(right.min_value)
         .map(|(a, b)| expressions::helpers::min(&a, &b))
-        .map(Result::ok)
-        .flatten();
+        .and_then(Result::ok);
     left.max_value = left
         .max_value
         .zip(right.max_value)
         .map(|(a, b)| expressions::helpers::max(&a, &b))
-        .map(Result::ok)
-        .flatten();
+        .and_then(Result::ok);
     left.null_count = left.null_count.zip(right.null_count).map(|(a, b)| a + b);
 
     left

datafusion/src/physical_plan/values.rs

@@ -190,7 +190,7 @@ mod tests {
     async fn values_empty_case() -> Result<()> {
         let schema = test_util::aggr_test_schema();
         let empty = ValuesExec::try_new(schema, vec![]);
-        assert!(!empty.is_ok());
+        assert!(empty.is_err());
         Ok(())
     }
 }

datafusion/tests/merge_fuzz.rs

@@ -104,8 +104,7 @@ async fn run_merge_test(input: Vec<Vec<RecordBatch>>) {
     for batch_size in batch_sizes {
         let first_batch = input
             .iter()
-            .map(|p| p.iter())
-            .flatten()
+            .flat_map(|p| p.iter())
             .next()
             .expect("at least one batch");
         let schema = first_batch.schema();

datafusion/tests/order_spill_fuzz.rs

@@ -58,8 +58,7 @@ async fn run_sort(pool_size: usize, size_spill: Vec<(usize, bool)>) {
         let input = vec![make_staggered_batches(size)];
         let first_batch = input
             .iter()
-            .map(|p| p.iter())
-            .flatten()
+            .flat_map(|p| p.iter())
             .next()
             .expect("at least one batch");
         let schema = first_batch.schema();