[MINOR]: Resolve linter errors in the main

datafusion/core/benches/topk_aggregate.rs

@@ -28,6 +28,7 @@ use datafusion_execution::config::SessionConfig;
 use datafusion_execution::TaskContext;
 use rand_distr::Distribution;
 use rand_distr::{Normal, Pareto};
+use std::fmt::Write;
 use std::sync::Arc;
 use tokio::runtime::Runtime;
 
@@ -130,8 +131,10 @@ fn make_data(
         let gen_id = |rng: &mut rand::rngs::SmallRng| {
             rng.gen::<[u8; 16]>()
                 .iter()
-                .map(|b| format!("{:02x}", b))
-                .collect::<String>()
+                .fold(String::new(), |mut output, b| {
+                    let _ = write!(output, "{b:02X}");
+                    output
+                })
         };
         let gen_sample_cnt =
             |mut rng: &mut rand::rngs::SmallRng| pareto.sample(&mut rng).ceil() as u32;

datafusion/core/src/catalog/mod.rs

@@ -31,7 +31,7 @@ use std::sync::Arc;
 
 /// Represent a list of named catalogs
 pub trait CatalogList: Sync + Send {
-    /// Returns the catalog list as [`Any`](std::any::Any)
+    /// Returns the catalog list as [`Any`]
     /// so that it can be downcast to a specific implementation.
     fn as_any(&self) -> &dyn Any;
 
@@ -101,7 +101,7 @@ impl Default for MemoryCatalogProvider {
 
 /// Represents a catalog, comprising a number of named schemas.
 pub trait CatalogProvider: Sync + Send {
-    /// Returns the catalog provider as [`Any`](std::any::Any)
+    /// Returns the catalog provider as [`Any`]
     /// so that it can be downcast to a specific implementation.
     fn as_any(&self) -> &dyn Any;
 

datafusion/core/src/datasource/default_table_source.rs

@@ -43,7 +43,7 @@ impl DefaultTableSource {
 }
 
 impl TableSource for DefaultTableSource {
-    /// Returns the table source as [`Any`](std::any::Any) so that it can be
+    /// Returns the table source as [`Any`] so that it can be
     /// downcast to a specific implementation.
     fn as_any(&self) -> &dyn Any {
         self

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

@@ -1101,7 +1101,7 @@ pub(crate) mod test_util {
         Ok((meta, files))
     }
 
-    //// write batches chunk_size rows at a time
+    /// write batches chunk_size rows at a time
     fn write_in_chunks<W: std::io::Write + Send>(
         writer: &mut ArrowWriter<W>,
         batch: &RecordBatch,

datafusion/core/src/physical_optimizer/enforce_distribution.rs

@@ -1529,7 +1529,7 @@ impl DistributionContext {
         self.plan
             .children()
             .into_iter()
-            .map(|child| DistributionContext::new(child))
+            .map(DistributionContext::new)
             .collect()
     }
 }

datafusion/core/src/physical_optimizer/enforce_sorting.rs

@@ -151,7 +151,7 @@ impl PlanWithCorrespondingSort {
         self.plan
             .children()
             .into_iter()
-            .map(|child| PlanWithCorrespondingSort::new(child))
+            .map(PlanWithCorrespondingSort::new)
             .collect()
     }
 }
@@ -267,7 +267,7 @@ impl PlanWithCorrespondingCoalescePartitions {
         self.plan
             .children()
             .into_iter()
-            .map(|child| PlanWithCorrespondingCoalescePartitions::new(child))
+            .map(PlanWithCorrespondingCoalescePartitions::new)
             .collect()
     }
 }
@@ -602,7 +602,7 @@ fn analyze_window_sort_removal(
         let reqs = window_exec
             .required_input_ordering()
             .swap_remove(0)
-            .unwrap_or(vec![]);
+            .unwrap_or_default();
         let sort_expr = PhysicalSortRequirement::to_sort_exprs(reqs);
         // Satisfy the ordering requirement so that the window can run:
         add_sort_above(&mut window_child, sort_expr, None)?;

datafusion/core/src/physical_optimizer/pipeline_checker.rs

@@ -108,7 +108,7 @@ impl TreeNode for PipelineStatePropagator {
         if !children.is_empty() {
             let new_children = children
                 .into_iter()
-                .map(|child| PipelineStatePropagator::new(child))
+                .map(PipelineStatePropagator::new)
                 .map(transform)
                 .collect::<Result<Vec<_>>>()?;
             let children_unbounded = new_children

datafusion/core/src/physical_optimizer/replace_with_order_preserving_variants.rs

@@ -112,7 +112,7 @@ impl OrderPreservationContext {
         self.plan
             .children()
             .into_iter()
-            .map(|child| OrderPreservationContext::new(child))
+            .map(OrderPreservationContext::new)
             .collect()
     }
 }

datafusion/core/src/physical_optimizer/test_utils.rs

@@ -140,14 +140,13 @@ impl QueryCase {
     async fn run_case(&self, ctx: SessionContext, error: Option<&String>) -> Result<()> {
         let dataframe = ctx.sql(self.sql.as_str()).await?;
         let plan = dataframe.create_physical_plan().await;
-        if error.is_some() {
+        if let Some(error) = error {
             let plan_error = plan.unwrap_err();
-            let initial = error.unwrap().to_string();
             assert!(
-                plan_error.to_string().contains(initial.as_str()),
+                plan_error.to_string().contains(error.as_str()),
                 "plan_error: {:?} doesn't contain message: {:?}",
                 plan_error,
-                initial.as_str()
+                error.as_str()
             );
         } else {
             assert!(plan.is_ok())

datafusion/core/tests/custom_sources.rs

@@ -49,7 +49,7 @@ mod custom_sources_cases;
 
 use async_trait::async_trait;
 
-//// Custom source dataframe tests ////
+//--- Custom source dataframe tests ---//
 
 struct CustomTableProvider;
 #[derive(Debug, Clone)]

datafusion/execution/src/memory_pool/mod.rs

@@ -46,9 +46,9 @@ pub use pool::*;
 ///
 /// The following memory pool implementations are available:
 ///
-/// * [`UnboundedMemoryPool`](pool::UnboundedMemoryPool)
-/// * [`GreedyMemoryPool`](pool::GreedyMemoryPool)
-/// * [`FairSpillPool`](pool::FairSpillPool)
+/// * [`UnboundedMemoryPool`]
+/// * [`GreedyMemoryPool`]
+/// * [`FairSpillPool`]
 pub trait MemoryPool: Send + Sync + std::fmt::Debug {
     /// Registers a new [`MemoryConsumer`]
     ///

datafusion/execution/src/stream.rs

@@ -29,5 +29,5 @@ pub trait RecordBatchStream: Stream<Item = Result<RecordBatch>> {
     fn schema(&self) -> SchemaRef;
 }
 
-/// Trait for a [`Stream`](futures::stream::Stream) of [`RecordBatch`]es
+/// Trait for a [`Stream`] of [`RecordBatch`]es
 pub type SendableRecordBatchStream = Pin<Box<dyn RecordBatchStream + Send>>;

datafusion/expr/src/expr.rs

@@ -39,7 +39,7 @@ use std::sync::Arc;
 /// represent logical expressions such as `A + 1`, or `CAST(c1 AS
 /// int)`.
 ///
-/// An `Expr` can compute its [DataType](arrow::datatypes::DataType)
+/// An `Expr` can compute its [DataType]
 /// and nullability, and has functions for building up complex
 /// expressions.
 ///

datafusion/optimizer/src/push_down_filter.rs

@@ -488,7 +488,7 @@ fn push_down_join(
         .filter
         .as_ref()
         .map(|e| utils::split_conjunction_owned(e.clone()))
-        .unwrap_or_else(Vec::new);
+        .unwrap_or_default();
 
     let mut is_inner_join = false;
     let infer_predicates = if join.join_type == JoinType::Inner {

datafusion/physical-expr/src/aggregate/mod.rs

@@ -68,7 +68,7 @@ pub(crate) mod variance;
 /// `PartialEq<dyn Any>` to allows comparing equality between the
 /// trait objects.
 pub trait AggregateExpr: Send + Sync + Debug + PartialEq<dyn Any> {
-    /// Returns the aggregate expression as [`Any`](std::any::Any) so that it can be
+    /// Returns the aggregate expression as [`Any`] so that it can be
     /// downcast to a specific implementation.
     fn as_any(&self) -> &dyn Any;
 

datafusion/physical-expr/src/aggregate/utils.rs

@@ -161,7 +161,7 @@ pub fn adjust_output_array(
 }
 
 /// Downcast a `Box<dyn AggregateExpr>` or `Arc<dyn AggregateExpr>`
-/// and return the inner trait object as [`Any`](std::any::Any) so
+/// and return the inner trait object as [`Any`] so
 /// that it can be downcast to a specific implementation.
 ///
 /// This method is used when implementing the `PartialEq<dyn Any>`

datafusion/physical-expr/src/expressions/binary.rs

@@ -299,7 +299,7 @@ impl PhysicalExpr for BinaryExpr {
         };
 
         if let Some(result) = scalar_result {
-            return result.map(|a| ColumnarValue::Array(a));
+            return result.map(ColumnarValue::Array);
         }
 
         // if both arrays or both literals - extract arrays and continue execution
@@ -308,7 +308,7 @@ impl PhysicalExpr for BinaryExpr {
             rhs.into_array(batch.num_rows()),
         );
         self.evaluate_with_resolved_args(left, &left_data_type, right, &right_data_type)
-            .map(|a| ColumnarValue::Array(a))
+            .map(ColumnarValue::Array)
     }
 
     fn children(&self) -> Vec<Arc<dyn PhysicalExpr>> {

datafusion/physical-expr/src/physical_expr.rs

@@ -35,7 +35,7 @@ use std::sync::Arc;
 /// Expression that can be evaluated against a RecordBatch
 /// A Physical expression knows its type, nullability and how to evaluate itself.
 pub trait PhysicalExpr: Send + Sync + Display + Debug + PartialEq<dyn Any> {
-    /// Returns the physical expression as [`Any`](std::any::Any) so that it can be
+    /// Returns the physical expression as [`Any`] so that it can be
     /// downcast to a specific implementation.
     fn as_any(&self) -> &dyn Any;
     /// Get the data type of this expression, given the schema of the input

datafusion/physical-expr/src/sort_properties.rs

@@ -172,7 +172,7 @@ impl ExprOrdering {
         self.expr
             .children()
             .into_iter()
-            .map(|e| ExprOrdering::new(e))
+            .map(ExprOrdering::new)
             .collect()
     }
 

datafusion/physical-expr/src/window/built_in_window_function_expr.rs

@@ -37,7 +37,7 @@ use std::sync::Arc;
 /// `nth_value` need the value.
 #[allow(rustdoc::private_intra_doc_links)]
 pub trait BuiltInWindowFunctionExpr: Send + Sync + std::fmt::Debug {
-    /// Returns the aggregate expression as [`Any`](std::any::Any) so that it can be
+    /// Returns the aggregate expression as [`Any`] so that it can be
     /// downcast to a specific implementation.
     fn as_any(&self) -> &dyn Any;
 

datafusion/physical-expr/src/window/window_expr.rs

@@ -59,7 +59,7 @@ use std::sync::Arc;
 /// [`PlainAggregateWindowExpr`]: crate::window::PlainAggregateWindowExpr
 /// [`SlidingAggregateWindowExpr`]: crate::window::SlidingAggregateWindowExpr
 pub trait WindowExpr: Send + Sync + Debug {
-    /// Returns the window expression as [`Any`](std::any::Any) so that it can be
+    /// Returns the window expression as [`Any`] so that it can be
     /// downcast to a specific implementation.
     fn as_any(&self) -> &dyn Any;
 

datafusion/physical-plan/src/aggregates/mod.rs

@@ -681,7 +681,7 @@ impl AggregateExec {
         for (expression, name) in group_by.expr.iter() {
             if let Some(column) = expression.as_any().downcast_ref::<Column>() {
                 let new_col_idx = schema.index_of(name)?;
-                let entry = columns_map.entry(column.clone()).or_insert_with(Vec::new);
+                let entry = columns_map.entry(column.clone()).or_default();
                 entry.push(Column::new(name, new_col_idx));
             };
         }
@@ -1868,7 +1868,7 @@ mod tests {
         }
     }
 
-    //// Tests ////
+    //--- Tests ---//
 
     #[tokio::test]
     async fn aggregate_source_not_yielding() -> Result<()> {

datafusion/physical-plan/src/lib.rs

@@ -62,7 +62,7 @@ pub use stream::EmptyRecordBatchStream;
 /// return value from [`displayable`] in addition to the (normally
 /// quite verbose) `Debug` output.
 pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
-    /// Returns the execution plan as [`Any`](std::any::Any) so that it can be
+    /// Returns the execution plan as [`Any`] so that it can be
     /// downcast to a specific implementation.
     fn as_any(&self) -> &dyn Any;
 

datafusion/physical-plan/src/memory.rs

@@ -81,7 +81,7 @@ impl DisplayAs for MemoryExec {
                             output_ordering.iter().map(|e| e.to_string()).collect();
                         format!(", output_ordering={}", order_strings.join(","))
                     })
-                    .unwrap_or_else(|| "".to_string());
+                    .unwrap_or_default();
 
                 write!(
                     f,

datafusion/physical-plan/src/projection.rs

@@ -111,9 +111,7 @@ impl ProjectionExec {
                 let idx = column.index();
                 let matching_input_field = input_schema.field(idx);
                 let matching_input_column = Column::new(matching_input_field.name(), idx);
-                let entry = columns_map
-                    .entry(matching_input_column)
-                    .or_insert_with(Vec::new);
+                let entry = columns_map.entry(matching_input_column).or_default();
                 entry.push(Column::new(name, expr_idx));
             };
         }

datafusion/physical-plan/src/visitor.rs

@@ -38,7 +38,7 @@ pub fn accept<V: ExecutionPlanVisitor>(
 /// depth first walk of `ExecutionPlan` nodes. `pre_visit` is called
 /// before any children are visited, and then `post_visit` is called
 /// after all children have been visited.
-////
+///
 /// To use, define a struct that implements this trait and then invoke
 /// ['accept'].
 ///

datafusion/sql/src/expr/mod.rs

@@ -719,7 +719,7 @@ fn rewrite_placeholder(expr: &mut Expr, other: &Expr, schema: &DFSchema) -> Resu
             let other_dt = other.get_type(schema);
             match other_dt {
                 Err(e) => {
-                    return Err(e.context(format!(
+                    Err(e.context(format!(
                         "Can not find type of {other} needed to infer type of {expr}"
                     )))?;
                 }

datafusion/sql/src/parser.rs

@@ -253,8 +253,7 @@ impl fmt::Display for Statement {
 
 /// Datafusion SQL Parser based on [`sqlparser`]
 ///
-/// Parses DataFusion's SQL dialect, often delegating to [`sqlparser`]'s
-/// [`Parser`](sqlparser::parser::Parser).
+/// Parses DataFusion's SQL dialect, often delegating to [`sqlparser`]'s [`Parser`].
 ///
 /// DataFusion mostly follows existing SQL dialects via
 /// `sqlparser`. However, certain statements such as `COPY` and