refactor: replace unwrap_or with unwrap_or_else for improved lazy…

Cargo.toml

@@ -211,6 +211,8 @@ strip = false
 # Detects large stack-allocated futures that may cause stack overflow crashes (see threshold in clippy.toml)
 large_futures = "warn"
 used_underscore_binding = "warn"
+or_fun_call = "warn"
+unnecessary_lazy_evaluations = "warn"
 
 [workspace.lints.rust]
 unexpected_cfgs = { level = "warn", check-cfg = ["cfg(tarpaulin)"] }

benchmarks/src/bin/external_aggr.rs

@@ -335,7 +335,7 @@ impl ExternalAggrConfig {
     fn partitions(&self) -> usize {
         self.common
             .partitions
-            .unwrap_or(get_available_parallelism())
+            .unwrap_or_else(get_available_parallelism)
     }
 }
 

benchmarks/src/imdb/run.rs

@@ -471,7 +471,7 @@ impl RunOpt {
     fn partitions(&self) -> usize {
         self.common
             .partitions
-            .unwrap_or(get_available_parallelism())
+            .unwrap_or_else(get_available_parallelism)
     }
 }
 

benchmarks/src/sort.rs

@@ -149,7 +149,7 @@ impl RunOpt {
                 let config = SessionConfig::new().with_target_partitions(
                     self.common
                         .partitions
-                        .unwrap_or(get_available_parallelism()),
+                        .unwrap_or_else(get_available_parallelism),
                 );
                 let ctx = SessionContext::new_with_config(config);
                 let (rows, elapsed) =

benchmarks/src/sort_tpch.rs

@@ -352,6 +352,6 @@ impl RunOpt {
     fn partitions(&self) -> usize {
         self.common
             .partitions
-            .unwrap_or(get_available_parallelism())
+            .unwrap_or_else(get_available_parallelism)
     }
 }

benchmarks/src/tpch/run.rs

@@ -313,7 +313,7 @@ impl RunOpt {
     fn partitions(&self) -> usize {
         self.common
             .partitions
-            .unwrap_or(get_available_parallelism())
+            .unwrap_or_else(get_available_parallelism)
     }
 }
 

benchmarks/src/util/options.rs

@@ -72,11 +72,11 @@ impl CommonOpt {
     /// Modify the existing config appropriately
     pub fn update_config(&self, mut config: SessionConfig) -> SessionConfig {
         if let Some(batch_size) = self.batch_size {
-            config = config.with_batch_size(batch_size)
+            config = config.with_batch_size(batch_size);
         }
 
         if let Some(partitions) = self.partitions {
-            config = config.with_target_partitions(partitions)
+            config = config.with_target_partitions(partitions);
         }
 
         if let Some(sort_spill_reservation_bytes) = self.sort_spill_reservation_bytes {

datafusion/common/src/column.rs

@@ -130,8 +130,8 @@ impl Column {
     /// where `"foo.BAR"` would be parsed to a reference to column named `foo.BAR`
     pub fn from_qualified_name(flat_name: impl Into<String>) -> Self {
         let flat_name = flat_name.into();
-        Self::from_idents(parse_identifiers_normalized(&flat_name, false)).unwrap_or(
-            Self {
+        Self::from_idents(parse_identifiers_normalized(&flat_name, false)).unwrap_or_else(
+            || Self {
                 relation: None,
                 name: flat_name,
                 spans: Spans::new(),
@@ -142,8 +142,8 @@ impl Column {
     /// Deserialize a fully qualified name string into a column preserving column text case
     pub fn from_qualified_name_ignore_case(flat_name: impl Into<String>) -> Self {
         let flat_name = flat_name.into();
-        Self::from_idents(parse_identifiers_normalized(&flat_name, true)).unwrap_or(
-            Self {
+        Self::from_idents(parse_identifiers_normalized(&flat_name, true)).unwrap_or_else(
+            || Self {
                 relation: None,
                 name: flat_name,
                 spans: Spans::new(),

datafusion/common/src/error.rs

@@ -526,7 +526,7 @@ impl DataFusionError {
     pub fn message(&self) -> Cow<str> {
         match *self {
             DataFusionError::ArrowError(ref desc, ref backtrace) => {
-                let backtrace = backtrace.clone().unwrap_or("".to_owned());
+                let backtrace = backtrace.clone().unwrap_or_else(|| "".to_owned());
                 Cow::Owned(format!("{desc}{backtrace}"))
             }
             #[cfg(feature = "parquet")]
@@ -535,7 +535,8 @@ impl DataFusionError {
             DataFusionError::AvroError(ref desc) => Cow::Owned(desc.to_string()),
             DataFusionError::IoError(ref desc) => Cow::Owned(desc.to_string()),
             DataFusionError::SQL(ref desc, ref backtrace) => {
-                let backtrace: String = backtrace.clone().unwrap_or("".to_owned());
+                let backtrace: String =
+                    backtrace.clone().unwrap_or_else(|| "".to_owned());
                 Cow::Owned(format!("{desc:?}{backtrace}"))
             }
             DataFusionError::Configuration(ref desc) => Cow::Owned(desc.to_string()),
@@ -547,7 +548,7 @@ impl DataFusionError {
             DataFusionError::Plan(ref desc) => Cow::Owned(desc.to_string()),
             DataFusionError::SchemaError(ref desc, ref backtrace) => {
                 let backtrace: &str =
-                    &backtrace.as_ref().clone().unwrap_or("".to_owned());
+                    &backtrace.as_ref().clone().unwrap_or_else(|| "".to_owned());
                 Cow::Owned(format!("{desc}{backtrace}"))
             }
             DataFusionError::Execution(ref desc) => Cow::Owned(desc.to_string()),

datafusion/core/src/datasource/file_format/parquet.rs

@@ -1085,7 +1085,7 @@ mod tests {
         let format = state
             .get_file_format_factory("parquet")
             .map(|factory| factory.create(state, &Default::default()).unwrap())
-            .unwrap_or(Arc::new(ParquetFormat::new()));
+            .unwrap_or_else(|| Arc::new(ParquetFormat::new()));
 
         scan_format(
             state, &*format, None, &testdata, file_name, projection, limit,

datafusion/datasource-csv/src/file_format.rs

@@ -414,11 +414,11 @@ impl FileFormat for CsvFormat {
         let has_header = self
             .options
             .has_header
-            .unwrap_or(state.config_options().catalog.has_header);
+            .unwrap_or_else(|| state.config_options().catalog.has_header);
         let newlines_in_values = self
             .options
             .newlines_in_values
-            .unwrap_or(state.config_options().catalog.newlines_in_values);
+            .unwrap_or_else(|| state.config_options().catalog.newlines_in_values);
 
         let conf_builder = FileScanConfigBuilder::from(conf)
             .with_file_compression_type(self.options.compression.into())
@@ -454,11 +454,11 @@ impl FileFormat for CsvFormat {
         let has_header = self
             .options()
             .has_header
-            .unwrap_or(state.config_options().catalog.has_header);
+            .unwrap_or_else(|| state.config_options().catalog.has_header);
         let newlines_in_values = self
             .options()
             .newlines_in_values
-            .unwrap_or(state.config_options().catalog.newlines_in_values);
+            .unwrap_or_else(|| state.config_options().catalog.newlines_in_values);
 
         let options = self
             .options()
@@ -504,7 +504,7 @@ impl CsvFormat {
                         && self
                             .options
                             .has_header
-                            .unwrap_or(state.config_options().catalog.has_header),
+                            .unwrap_or_else(|| state.config_options().catalog.has_header),
                 )
                 .with_delimiter(self.options.delimiter)
                 .with_quote(self.options.quote);

datafusion/datasource/src/file_scan_config.rs

@@ -580,7 +580,7 @@ impl DataSource for FileScanConfig {
                 &file_scan
                     .projection
                     .clone()
-                    .unwrap_or((0..self.file_schema.fields().len()).collect()),
+                    .unwrap_or_else(|| (0..self.file_schema.fields().len()).collect()),
             );
             DataSourceExec::from_data_source(
                 FileScanConfigBuilder::from(file_scan)

datafusion/datasource/src/statistics.rs

@@ -137,7 +137,9 @@ impl MinMaxStatistics {
                 // Reverse the projection to get the index of the column in the full statistics
                 // The file statistics contains _every_ column , but the sort column's index()
                 // refers to the index in projected_schema
-                let i = projection.map(|p| p[c.index()]).unwrap_or(c.index());
+                let i = projection
+                    .map(|p| p[c.index()])
+                    .unwrap_or_else(|| c.index());
 
                 let (min, max) = get_min_max(i).map_err(|e| {
                     e.context(format!("get min/max for column: '{}'", c.name()))

datafusion/datasource/src/write/demux.rs

@@ -513,7 +513,7 @@ fn compute_take_arrays(
         for vals in all_partition_values.iter() {
             part_key.push(vals[i].clone().into());
         }
-        let builder = take_map.entry(part_key).or_insert(UInt64Builder::new());
+        let builder = take_map.entry(part_key).or_insert_with(UInt64Builder::new);
         builder.append_value(i as u64);
     }
     take_map

datafusion/expr-common/src/type_coercion/binary.rs

@@ -1138,7 +1138,7 @@ fn dictionary_comparison_coercion(
 /// 2. Data type of the other side should be able to cast to string type
 fn string_concat_coercion(lhs_type: &DataType, rhs_type: &DataType) -> Option<DataType> {
     use arrow::datatypes::DataType::*;
-    string_coercion(lhs_type, rhs_type).or(match (lhs_type, rhs_type) {
+    string_coercion(lhs_type, rhs_type).or_else(|| match (lhs_type, rhs_type) {
         (Utf8View, from_type) | (from_type, Utf8View) => {
             string_concat_internal_coercion(from_type, &Utf8View)
         }

datafusion/expr/src/expr_fn.rs

@@ -838,7 +838,7 @@ impl ExprFuncBuilder {
                         partition_by: partition_by.unwrap_or_default(),
                         order_by: order_by.unwrap_or_default(),
                         window_frame: window_frame
-                            .unwrap_or(WindowFrame::new(has_order_by)),
+                            .unwrap_or_else(|| WindowFrame::new(has_order_by)),
                         null_treatment,
                     },
                 })

datafusion/functions-aggregate/src/array_agg.rs

@@ -1057,7 +1057,7 @@ mod tests {
 
     fn print_nulls(sort: Vec<Option<String>>) -> Vec<String> {
         sort.into_iter()
-            .map(|v| v.unwrap_or("NULL".to_string()))
+            .map(|v| v.unwrap_or_else(|| "NULL".to_string()))
             .collect()
     }
 

datafusion/functions-window/src/lead_lag.rs

@@ -321,7 +321,7 @@ fn parse_default_value(
     unparsed
         .filter(|v| !v.data_type().is_null())
         .map(|v| v.cast_to(&expr_type))
-        .unwrap_or(ScalarValue::try_from(expr_type))
+        .unwrap_or_else(|| ScalarValue::try_from(expr_type))
 }
 
 #[derive(Debug)]

datafusion/macros/src/user_doc.rs

@@ -206,7 +206,7 @@ pub fn user_doc(args: TokenStream, input: TokenStream) -> TokenStream {
         };
     let doc_section_description = doc_section_desc
         .map(|desc| quote! { Some(#desc)})
-        .unwrap_or(quote! { None });
+        .unwrap_or_else(|| quote! { None });
 
     let sql_example = sql_example.map(|ex| {
         quote! {

datafusion/optimizer/src/simplify_expressions/regex.rs

@@ -331,7 +331,7 @@ fn lower_simple(mode: &OperatorMode, left: &Expr, hir: &Hir) -> Option<Expr> {
         }
         HirKind::Concat(inner) => {
             if let Some(pattern) = partial_anchored_literal_to_like(inner)
-                .or(collect_concat_to_like_string(inner))
+                .or_else(|| collect_concat_to_like_string(inner))
             {
                 return Some(mode.expr(Box::new(left.clone()), pattern));
             }

datafusion/physical-expr/src/analysis.rs

@@ -112,7 +112,7 @@ impl ExprBoundaries {
                 .min_value
                 .get_value()
                 .cloned()
-                .unwrap_or(empty_field.clone()),
+                .unwrap_or_else(|| empty_field.clone()),
             col_stats
                 .max_value
                 .get_value()

datafusion/physical-expr/src/equivalence/properties/mod.rs

@@ -1364,7 +1364,7 @@ impl EquivalenceProperties {
             .transform_up(|expr| update_properties(expr, self))
             .data()
             .map(|node| node.data)
-            .unwrap_or(ExprProperties::new_unknown())
+            .unwrap_or_else(|_| ExprProperties::new_unknown())
     }
 
     /// Transforms this `EquivalenceProperties` into a new `EquivalenceProperties`

datafusion/physical-optimizer/src/enforce_distribution.rs

@@ -42,7 +42,6 @@ use datafusion_physical_expr::utils::map_columns_before_projection;
 use datafusion_physical_expr::{
     physical_exprs_equal, EquivalenceProperties, PhysicalExpr, PhysicalExprRef,
 };
-use datafusion_physical_expr_common::sort_expr::LexOrdering;
 use datafusion_physical_plan::aggregates::{
     AggregateExec, AggregateMode, PhysicalGroupBy,
 };
@@ -950,11 +949,7 @@ fn add_spm_on_top(input: DistributionContext) -> DistributionContext {
 
         let new_plan = if should_preserve_ordering {
             Arc::new(SortPreservingMergeExec::new(
-                input
-                    .plan
-                    .output_ordering()
-                    .unwrap_or(&LexOrdering::default())
-                    .clone(),
+                input.plan.output_ordering().cloned().unwrap_or_default(),
                 Arc::clone(&input.plan),
             )) as _
         } else {

datafusion/physical-optimizer/src/enforce_sorting/mod.rs

@@ -505,7 +505,7 @@ fn analyze_immediate_sort_removal(
             sort_exec
                 .properties()
                 .output_ordering()
-                .unwrap_or(LexOrdering::empty()),
+                .unwrap_or_else(|| LexOrdering::empty()),
         ) {
             node.plan = if !sort_exec.preserve_partitioning()
                 && sort_input.output_partitioning().partition_count() > 1

datafusion/physical-optimizer/src/enforce_sorting/replace_with_order_preserving_variants.rs

@@ -27,8 +27,10 @@ use crate::utils::{
 
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::Transformed;
-use datafusion_common::{internal_err, Result};
-use datafusion_physical_expr_common::sort_expr::LexOrdering;
+use datafusion_physical_expr::LexOrdering;
+use datafusion_physical_plan::internal_err;
+
+use datafusion_common::Result;
 use datafusion_physical_plan::coalesce_partitions::CoalescePartitionsExec;
 use datafusion_physical_plan::execution_plan::EmissionType;
 use datafusion_physical_plan::repartition::RepartitionExec;
@@ -286,7 +288,7 @@ pub fn replace_with_order_preserving_variants(
             requirements
                 .plan
                 .output_ordering()
-                .unwrap_or(LexOrdering::empty()),
+                .unwrap_or_else(|| LexOrdering::empty()),
         )
     {
         for child in alternate_plan.children.iter_mut() {

datafusion/physical-optimizer/src/enforce_sorting/sort_pushdown.rs

@@ -233,7 +233,7 @@ fn pushdown_requirement_to_children(
                 .properties()
                 .output_ordering()
                 .cloned()
-                .unwrap_or(LexOrdering::default()),
+                .unwrap_or_else(LexOrdering::default),
         );
         if sort_exec
             .properties()
@@ -258,7 +258,7 @@ fn pushdown_requirement_to_children(
             plan.properties()
                 .output_ordering()
                 .cloned()
-                .unwrap_or(LexOrdering::default()),
+                .unwrap_or_else(LexOrdering::default),
         );
         // Push down through operator with fetch when:
         // - requirement is aligned with output ordering

datafusion/physical-optimizer/src/update_aggr_exprs.rs

@@ -24,10 +24,11 @@ use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
 use datafusion_common::{plan_datafusion_err, Result};
 use datafusion_physical_expr::aggregate::AggregateFunctionExpr;
+use datafusion_physical_expr::LexRequirement;
 use datafusion_physical_expr::{
     reverse_order_bys, EquivalenceProperties, PhysicalSortRequirement,
 };
-use datafusion_physical_expr::{LexOrdering, LexRequirement};
+use datafusion_physical_expr_common::sort_expr::LexOrdering;
 use datafusion_physical_plan::aggregates::concat_slices;
 use datafusion_physical_plan::windows::get_ordered_partition_by_indices;
 use datafusion_physical_plan::{
@@ -159,7 +160,9 @@ fn try_convert_aggregate_if_better(
     aggr_exprs
         .into_iter()
         .map(|aggr_expr| {
-            let aggr_sort_exprs = aggr_expr.order_bys().unwrap_or(LexOrdering::empty());
+            let aggr_sort_exprs = aggr_expr
+                .order_bys()
+                .unwrap_or_else(|| LexOrdering::empty());
             let reverse_aggr_sort_exprs = reverse_order_bys(aggr_sort_exprs);
             let aggr_sort_reqs = LexRequirement::from(aggr_sort_exprs.clone());
             let reverse_aggr_req = LexRequirement::from(reverse_aggr_sort_exprs);

datafusion/physical-plan/src/projection.rs

@@ -525,7 +525,7 @@ pub fn remove_unnecessary_projections(
         } else {
             return Ok(Transformed::no(plan));
         };
-    Ok(maybe_modified.map_or(Transformed::no(plan), Transformed::yes))
+    Ok(maybe_modified.map_or_else(|| Transformed::no(plan), Transformed::yes))
 }
 
 /// Compare the inputs and outputs of the projection. All expressions must be

datafusion/physical-plan/src/streaming.rs

@@ -302,7 +302,7 @@ impl ExecutionPlan for StreamingTableExec {
         let new_projections = new_projections_for_columns(
             projection,
             &streaming_table_projections
-                .unwrap_or((0..self.schema().fields().len()).collect()),
+                .unwrap_or_else(|| (0..self.schema().fields().len()).collect()),
         );
 
         let mut lex_orderings = vec![];